File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/client.tar
Back
agent.py 0000644 00000003351 15030320052 0006204 0 ustar 00 # -*- test-case-name: twisted.conch.test.test_default -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Accesses the key agent for user authentication. Maintainer: Paul Swartz """ import os from twisted.conch.ssh import agent, channel, keys from twisted.internet import protocol, reactor from twisted.logger import Logger class SSHAgentClient(agent.SSHAgentClient): _log = Logger() def __init__(self): agent.SSHAgentClient.__init__(self) self.blobs = [] def getPublicKeys(self): return self.requestIdentities().addCallback(self._cbPublicKeys) def _cbPublicKeys(self, blobcomm): self._log.debug("got {num_keys} public keys", num_keys=len(blobcomm)) self.blobs = [x[0] for x in blobcomm] def getPublicKey(self): """ Return a L{Key} from the first blob in C{self.blobs}, if any, or return L{None}. """ if self.blobs: return keys.Key.fromString(self.blobs.pop(0)) return None class SSHAgentForwardingChannel(channel.SSHChannel): def channelOpen(self, specificData): cc = protocol.ClientCreator(reactor, SSHAgentForwardingLocal) d = cc.connectUNIX(os.environ["SSH_AUTH_SOCK"]) d.addCallback(self._cbGotLocal) d.addErrback(lambda x: self.loseConnection()) self.buf = "" def _cbGotLocal(self, local): self.local = local self.dataReceived = self.local.transport.write self.local.dataReceived = self.write def dataReceived(self, data): self.buf += data def closed(self): if self.local: self.local.loseConnection() self.local = None class SSHAgentForwardingLocal(protocol.Protocol): pass __pycache__/knownhosts.cpython-310.pyc 0000644 00000047302 15030320052 0013666 0 ustar 00 o �b�M � @ sP d Z ddlZddlZddlmZmZmZ ddlm Z ddl m Z ddl m Z ddlmZmZmZ ddlmZ dd lmZmZmZ dd lmZ dd lmZ dd l m Z dd l m Z ddl m!Z! e� Z"dd� Z#dd� Z$G dd� d�Z%e e�G dd� de%��Z&e e�G dd� d��Z'dd� Z(e e�G dd � d e%e!��Z)G d d � d �Z*G d d � d �Z+dS )!zE An implementation of the OpenSSH known_hosts database. @since: 8.2 � N)�Error� a2b_base64� b2a_base64)�closing)�sha1)� implementer)�HostKeyChanged� InvalidEntry�UserRejectedKey)�IKnownHostEntry)� BadKeyError�FingerprintFormats�Key)�defer)�Logger)� nativeString)� secureRandom)� FancyEqMixinc C s t | ��� S )z� Encode a binary string as base64 with no trailing newline. @param s: The string to encode. @type s: L{bytes} @return: The base64-encoded string. @rtype: L{bytes} )r �strip)�s� r �A/usr/lib/python3/dist-packages/twisted/conch/client/knownhosts.py� _b64encode s r c C sz | � dd�}t|�dkrt� �|\}}}|� dd�}t|�dkr*|\}}|�d�}n|d }d}t�t|��}||||fS )a� Extract common elements of base64 keys from an entry in a hosts file. @param string: A known hosts file entry (a single line). @type string: L{bytes} @return: a 4-tuple of hostname data (L{bytes}), ssh key type (L{bytes}), key (L{Key}), and comment (L{bytes} or L{None}). The hostname data is simply the beginning of the line up to the first occurrence of whitespace. @rtype: L{tuple} N� � � � r )�split�lenr �rstripr � fromStringr ) �string�elements� hostnames�keyType� keyAndComment�splitkey� keyString�comment�keyr r r �_extractCommon- s r* c @ s e Zd ZdZdd� Zdd� ZdS )� _BaseEntrya� Abstract base of both hashed and non-hashed entry objects, since they represent keys and key types the same way. @ivar keyType: The type of the key; either ssh-dss or ssh-rsa. @type keyType: L{bytes} @ivar publicKey: The server public key indicated by this line. @type publicKey: L{twisted.conch.ssh.keys.Key} @ivar comment: Trailing garbage after the key line. @type comment: L{bytes} c C s || _ || _|| _d S �N)r$ � publicKeyr( )�selfr$ r- r( r r r �__init__X s z_BaseEntry.__init__c C s | j |kS )a Check to see if this entry matches a given key object. @param keyObject: A public key object to check. @type keyObject: L{Key} @return: C{True} if this entry's key matches C{keyObject}, C{False} otherwise. @rtype: L{bool} )r- )r. � keyObjectr r r � matchesKey] s z_BaseEntry.matchesKeyN)�__name__� __module__� __qualname__�__doc__r/ r1 r r r r r+ I s r+ c s< e Zd ZdZ� fdd�Zedd� �Zdd� Zdd � Z� Z S ) � PlainEntryz� A L{PlainEntry} is a representation of a plain-text entry in a known_hosts file. @ivar _hostnames: the list of all host-names associated with this entry. @type _hostnames: L{list} of L{bytes} c s || _ t� �|||� d S r, )� _hostnames�superr/ )r. r# r$ r- r( �� __class__r r r/ u s zPlainEntry.__init__c C s( t |�\}}}}| |�d�|||�}|S )a� Parse a plain-text entry in a known_hosts file, and return a corresponding L{PlainEntry}. @param string: a space-separated string formatted like "hostname key-type base64-key-data comment". @type string: L{bytes} @raise DecodeError: if the key is not valid encoded as valid base64. @raise InvalidEntry: if the entry does not have the right number of elements and is therefore invalid. @raise BadKeyError: if the key, once decoded from base64, is not actually an SSH key. @return: an IKnownHostEntry representing the hostname and key in the input line. @rtype: L{PlainEntry} � ,)r* r )�clsr! r# r$ r) r( r. r r r r y s zPlainEntry.fromStringc C s t |t�r |�d�}|| jv S )aT Check to see if this entry matches a given hostname. @param hostname: A hostname or IP address literal to check against this entry. @type hostname: L{bytes} @return: C{True} if this entry is for the given hostname or IP address, C{False} otherwise. @rtype: L{bool} �utf-8)� isinstance�str�encoder7 �r. �hostnamer r r � matchesHost� s zPlainEntry.matchesHostc C s> d� | j�| jt| j�� �g}| jdur|�| j� d� |�S )a Implement L{IKnownHostEntry.toString} by recording the comma-separated hostnames, key type, and base-64 encoded key. @return: The string representation of this entry, with unhashed hostname information. @rtype: L{bytes} r; N� )�joinr7 r$ r r- �blobr( �append�r. �fieldsr r r �toString� s � zPlainEntry.toString) r2 r3 r4 r5 r/ � classmethodr rC rJ � __classcell__r r r9 r r6 k s r6 c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd S ) � UnparsedEntryz� L{UnparsedEntry} is an entry in a L{KnownHostsFile} which can't actually be parsed; therefore it matches no keys and no hosts. c C � || _ dS )zv Create an unparsed entry from a line in a known_hosts file which cannot otherwise be parsed. N)�_string)r. r! r r r r/ � s zUnparsedEntry.__init__c C � dS �z' Always returns False. Fr rA r r r rC � � zUnparsedEntry.matchesHostc C rP rQ r )r. r) r r r r1 � rR zUnparsedEntry.matchesKeyc C s | j �d�S )a Returns the input line, without its newline if one was given. @return: The string representation of this entry, almost exactly as was used to initialize this entry but without a trailing newline. @rtype: L{bytes} r )rO r �r. r r r rJ � s zUnparsedEntry.toStringN)r2 r3 r4 r5 r/ rC r1 rJ r r r r rM � s rM c C s4 t j| td�}t|t�r|�d�}|�|� |�� S )z� Return the SHA-1 HMAC hash of the given key and string. @param key: The HMAC key. @type key: L{bytes} @param string: The string to be hashed. @type string: L{bytes} @return: The keyed hash value. @rtype: L{bytes} )� digestmodr= )�hmac�HMACr r> r? r@ �update�digest)r) r! �hashr r r � _hmacedString� s rZ c sD e Zd ZdZdZdZ� fdd�Zedd� �Zdd � Z d d � Z � Z S ) � HashedEntrya� A L{HashedEntry} is a representation of an entry in a known_hosts file where the hostname has been hashed and salted. @ivar _hostSalt: the salt to combine with a hostname for hashing. @ivar _hostHash: the hashed representation of the hostname. @cvar MAGIC: the 'hash magic' string used to identify a hashed line in a known_hosts file as opposed to a plaintext one. s |1|)� _hostSalt� _hostHashr$ r- r( c s || _ || _t� �|||� d S r, )r\ r] r8 r/ )r. �hostSalt�hostHashr$ r- r( r9 r r r/ s zHashedEntry.__init__c C s^ t |�\}}}}|t| j�d� �d�}t|�dkr t� �|\}}| t|�t|�|||�} | S )a# Load a hashed entry from a string representing a line in a known_hosts file. @param string: A complete single line from a I{known_hosts} file, formatted as defined by OpenSSH. @type string: L{bytes} @raise DecodeError: if the key, the hostname, or the is not valid encoded as valid base64 @raise InvalidEntry: if the entry does not have the right number of elements and is therefore invalid, or the host/hash portion contains more items than just the host and hash. @raise BadKeyError: if the key, once decoded from base64, is not actually an SSH key. @return: The newly created L{HashedEntry} instance, initialized with the information from C{string}. N� |r )r* r �MAGICr r r ) r< r! �stuffr$ r) r( � saltAndHashr^ r_ r. r r r r s zHashedEntry.fromStringc C s t �t| j|�| j�S )a� Implement L{IKnownHostEntry.matchesHost} to compare the hash of the input to the stored hash. @param hostname: A hostname or IP address literal to check against this entry. @type hostname: L{bytes} @return: C{True} if this entry is for the given hostname or IP address, C{False} otherwise. @rtype: L{bool} )rU �compare_digestrZ r\ r] rA r r r rC ' s �zHashedEntry.matchesHostc C sR | j d�t| j�t| j�g� | jt| j�� �g}| jdur$|� | j� d�|�S )z� Implement L{IKnownHostEntry.toString} by base64-encoding the salt, host hash, and key. @return: The string representation of this entry, with the hostname part hashed. @rtype: L{bytes} r` NrD ) ra rE r r\ r] r$ r- rF r( rG rH r r r rJ 8 s � � zHashedEntry.toString) r2 r3 r4 r5 ra �compareAttributesr/ rK r rC rJ rL r r r9 r r[ � s r[ c @ sX e Zd ZdZdd� Zedd� �Zdd� Zdd � Zd d � Z d d � Z dd� Z e dd� �Z dS )�KnownHostsFileaz A structured representation of an OpenSSH-format ~/.ssh/known_hosts file. @ivar _added: A list of L{IKnownHostEntry} providers which have been added to this instance in memory but not yet saved. @ivar _clobber: A flag indicating whether the current contents of the save path will be disregarded and potentially overwritten or not. If C{True}, this will be done. If C{False}, entries in the save path will be read and new entries will be saved by appending rather than overwriting. @type _clobber: L{bool} @ivar _savePath: See C{savePath} parameter of L{__init__}. c C s g | _ || _d| _dS )a$ Create a new, empty KnownHostsFile. Unless you want to erase the current contents of C{savePath}, you want to use L{KnownHostsFile.fromPath} instead. @param savePath: The L{FilePath} to which to save new entries. @type savePath: L{FilePath} TN)�_added� _savePath�_clobber)r. �savePathr r r r/ ] s zKnownHostsFile.__init__c C s | j S )z< @see: C{savePath} parameter of L{__init__} )rh rS r r r rj k s zKnownHostsFile.savePathc c s� � | j D ]}|V q| jrdS z| j�� }W n ty Y dS w |�5 |D ])}z|�tj�r5t�|�}nt �|�}W n t t t fyK t |�}Y nw |V q&W d � dS 1 s[w Y dS )aK Iterate over the host entries in this file. @return: An iterable the elements of which provide L{IKnownHostEntry}. There is an element for each entry in the file as well as an element for each added but not yet saved entry. @rtype: iterable of L{IKnownHostEntry} providers N)rg ri rh �open�OSError� startswithr[ ra r r6 � DecodeErrorr r rM )r. �entry�fp�liner r r � iterentriesr s. � � � ��"�zKnownHostsFile.iterentriesc C sx t | �� t| j� �D ].\}}|�|�r9|j|�� kr9|�|�r# dS |dk r,d}d}n|d }| j}t |||��q dS )a Check for an entry with matching hostname and key. @param hostname: A hostname or IP address literal to check for. @type hostname: L{bytes} @param key: The public key to check for. @type key: L{Key} @return: C{True} if the given hostname and key are present in this file, C{False} if they are not. @rtype: L{bool} @raise HostKeyChanged: if the host key found for the given hostname does not match the given key. Tr Nr F) � enumeraterr r rg rC r$ �sshTyper1 rh r )r. rB r) �lineidxro rq �pathr r r � hasHostKey� s �zKnownHostsFile.hasHostKeyc s. t ��j� ��}� ����fdd�}|�|�S )a� Verify the given host key for the given IP and host, asking for confirmation from, and notifying, the given UI about changes to this file. @param ui: The user interface to request an IP address from. @param hostname: The hostname that the user requested to connect to. @param ip: The string representation of the IP address that is actually being connected to. @param key: The public key of the server. @return: a L{Deferred} that fires with True when the key has been verified, or fires with an errback when the key either cannot be verified or has changed. @rtype: L{Deferred} c s� | r!�� ���s ��d��� t��f � ����� ��� | S � ���fdd�}��� }|dkr4d}dt� �t��|�jtjd�f }�� |� t � � ��}|� |�S )NzZWarning: Permanently added the %s host key for IP address '%s' to the list of known hosts.c s. | r�� � �� �� ��� ��� | S t� �r, )� addHostKey�saver )�response)rB �ipr) r. r r �promptResponse� s zGKnownHostsFile.verifyHostKey.<locals>.gotHasKey.<locals>.promptResponse�EC�ECDSAz�The authenticity of host '%s (%s)' can't be established. %s key fingerprint is SHA256:%s. Are you sure you want to continue connecting (yes/no)? )�format)rw �warn�typer rx ry � fingerprintr � SHA256_BASE64�promptr@ �sys�getdefaultencoding� addCallback)�resultr| �keytyper� �proceed�rB r{ r) r. �uir r � gotHasKey� s0 �� ��� z/KnownHostsFile.verifyHostKey.<locals>.gotHasKey)r � maybeDeferredrw r� )r. r� rB r{ r) �hhkr� r r� r � verifyHostKey� s *z KnownHostsFile.verifyHostKeyc C s6 t d�}|�� }t|t||�||d�}| j�|� |S )a� Add a new L{HashedEntry} to the key database. Note that you still need to call L{KnownHostsFile.save} if you wish these changes to be persisted. @param hostname: A hostname or IP address literal to associate with the new entry. @type hostname: L{bytes} @param key: The public key to associate with the new entry. @type key: L{Key} @return: The L{HashedEntry} that was added. @rtype: L{HashedEntry} � N)r rt r[ rZ rg rG )r. rB r) �saltr$ ro r r r rx � s zKnownHostsFile.addHostKeyc C s� | j �� }|�� s |�� | jrd}nd}| j �|�� }| jr2|�d�dd� | jD ��d � g | _W d � n1 s<w Y d| _dS )zM Save this L{KnownHostsFile} to the path it was loaded from. �wb�abr c S s g | ]}|� � �qS r )rJ )�.0ro r r r � <listcomp> s z'KnownHostsFile.save.<locals>.<listcomp>NF) rh �parent�isdir�makedirsri rk rg �writerE )r. �p�mode� hostsFileObjr r r ry s �� � zKnownHostsFile.savec C s | |�}d|_ |S )a� Create a new L{KnownHostsFile}, potentially reading existing known hosts information from the given file. @param path: A path object to use for both reading contents from and later saving to. If no file exists at this path, it is not an error; a L{KnownHostsFile} with no entries is returned. @type path: L{FilePath} @return: A L{KnownHostsFile} initialized with entries from C{path}. @rtype: L{KnownHostsFile} F)ri )r< rv � knownHostsr r r �fromPath s zKnownHostsFile.fromPathN)r2 r3 r4 r5 r/ �propertyrj rr rw r� rx ry rK r� r r r r rf L s !Brf c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) � ConsoleUIz� A UI object that can ask true/false questions and post notifications on the console, to be used during key verification. c C rN )aA @param opener: A no-argument callable which should open a console binary-mode file-like object to be used for reading and writing. This initializes the C{opener} attribute. @type opener: callable taking no arguments and returning a read/write file-like object N)�opener)r. r� r r r r/ 9 s zConsoleUI.__init__c s" t �d�}� �fdd�}|�|�S )a� Write the given text as a prompt to the console output, then read a result from the console input. @param text: Something to present to a user to solicit a yes or no response. @type text: L{bytes} @return: a L{Deferred} which fires with L{True} when the user answers 'yes' and L{False} when the user answers 'no'. It may errback if there were any I/O errors. Nc s~ t � �� ��/}|��� |�� �� �� }|dkr" W d � dS |dkr/ W d � dS |�d� q 1 s8w Y d S )NTs yess noFs Please type 'yes' or 'no': )r r� r� �readliner �lower)�ignored�f�answer�r. �textr r �bodyR s �� ��z ConsoleUI.prompt.<locals>.body)r �succeedr� )r. r� �dr� r r� r r� C s zConsoleUI.promptc C s` z t | �� ��}|�|� W d � W dS 1 sw Y W dS ty/ t�d� Y dS w )z� Notify the user (non-interactively) of the provided text, by writing it to the console. @param text: Some information the user is to be made aware of. @type text: L{bytes} NzFailed to write to console)r r� r� � Exception�log�failure)r. r� r� r r r r� ` s &� �zConsoleUI.warnN)r2 r3 r4 r5 r/ r� r� r r r r r� 3 s r� ),r5 rU r� �binasciir rn r r � contextlibr �hashlibr �zope.interfacer �twisted.conch.errorr r r �twisted.conch.interfacesr �twisted.conch.ssh.keysr r r �twisted.internetr �twisted.loggerr �twisted.python.compatr �twisted.python.randbytesr �twisted.python.utilr r� r r* r+ r6 rM rZ r[ rf r� r r r r �<module> s: "L$Z h __pycache__/__init__.cpython-310.pyc 0000644 00000000341 15030320052 0013200 0 ustar 00 o �b� � @ s d Z dS )z9 Client support code for Conch. Maintainer: Paul Swartz N)�__doc__� r r �?/usr/lib/python3/dist-packages/twisted/conch/client/__init__.py�<module> s __pycache__/connect.cpython-310.pyc 0000644 00000001264 15030320052 0013077 0 ustar 00 o �b� � @ s* d dl mZ dejiZdd� Zdd� ZdS )� )�directr c C s dg}t d || ||||�S )Nr )� _ebConnect)�host�port�options� verifyHostKey�userAuthObject� useConnects� r �>/usr/lib/python3/dist-packages/twisted/conch/client/connect.py�connect s �r c C sD |s| S |� d�}t| } | |||||�}|�t||||||� |S )Nr )�pop� connectTypes� addErrbackr ) �fr r r r �vhk�uao� connectType�dr r r r s r N)�twisted.conch.clientr r r r r r r r �<module> s __pycache__/agent.cpython-310.pyc 0000644 00000005761 15030320052 0012552 0 ustar 00 o �b� � @ sv d Z ddlZddlmZmZmZ ddlmZmZ ddl m Z G dd� dej �Z G dd � d ej �Z G d d � d ej�ZdS ) zJ Accesses the key agent for user authentication. Maintainer: Paul Swartz � N)�agent�channel�keys)�protocol�reactor)�Loggerc @ s2 e Zd Ze� Zdd� Zdd� Zdd� Zdd� Zd S ) �SSHAgentClientc C s t j�| � g | _d S �N)r r �__init__�blobs��self� r �</usr/lib/python3/dist-packages/twisted/conch/client/agent.pyr s zSSHAgentClient.__init__c C s | � � �| j�S r )�requestIdentities� addCallback� _cbPublicKeysr r r r � getPublicKeys s z SSHAgentClient.getPublicKeysc C s( | j jdt|�d� dd� |D �| _d S )Nzgot {num_keys} public keys)�num_keysc S s g | ]}|d �qS )r r )�.0�xr r r � <listcomp> s z0SSHAgentClient._cbPublicKeys.<locals>.<listcomp>)�_log�debug�lenr )r �blobcommr r r r s z SSHAgentClient._cbPublicKeysc C s | j r tj�| j �d��S dS )zj Return a L{Key} from the first blob in C{self.blobs}, if any, or return L{None}. r N)r r �Key� fromString�popr r r r � getPublicKey s zSSHAgentClient.getPublicKeyN) �__name__� __module__� __qualname__r r r r r r r r r r r s r c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S ) �SSHAgentForwardingChannelc sD t �tt�}|�tjd �}|�� j� |� � fdd�� d� _ d S )N� SSH_AUTH_SOCKc s � � � S r )�loseConnection)r r r r �<lambda>/ s z7SSHAgentForwardingChannel.channelOpen.<locals>.<lambda>� ) r � ClientCreatorr �SSHAgentForwardingLocal� connectUNIX�os�environr � _cbGotLocal� addErrback�buf)r � specificData�cc�dr r r � channelOpen+ s z%SSHAgentForwardingChannel.channelOpenc C s || _ | j jj| _| j| j _d S r )�local� transport�write� dataReceived)r r4 r r r r- 2 s z%SSHAgentForwardingChannel._cbGotLocalc C s | j |7 _ d S r )r/ )r �datar r r r7 7 s z&SSHAgentForwardingChannel.dataReceivedc C s | j r | j �� d | _ d S d S r )r4 r% r r r r �closed: s �z SSHAgentForwardingChannel.closedN)r r! r" r3 r- r7 r9 r r r r r# * s r# c @ s e Zd ZdS )r) N)r r! r" r r r r r) @ s r) )�__doc__r+ �twisted.conch.sshr r r �twisted.internetr r �twisted.loggerr r � SSHChannelr# �Protocolr) r r r r �<module> s __pycache__/options.cpython-310.pyc 0000644 00000007555 15030320052 0013152 0 ustar 00 o �b� � @ sN d dl Z d dlmZmZmZ d dlmZmZ d dlm Z G dd� de j �Z dS )� N)�List�Optional�Union)� SSHCiphers�SSHClientTransport)�usagec @ sN e Zd ZU g d�g d�g d�g d�g d�g d�g d�g d�g d �g d �g Zeeeeeef e d < g d �g d �g d�g d�g d�g d�g d�gZ e j dge � � e jdd� ej�� D �dd�e jdd� ej�� D �dd�e jdd� ejD �dd�d �e �� e jd d�e jd d d �gd!�Zd"d#� Zd$d%� Zd&d'� Zd(d)� Zd*d+� Z d,d-� Z d.S )/� ConchOptions)�user�lNz Log in using this user name.)�identity�iN)�ciphers�cN)�macs�mN)�port�pNz7Connect to this port. Server must be on the same port.)�option�oNzIgnored OpenSSH options)�host-key-algorithms� N)z known-hostsr NzFile to check for host keys)�user-authenticationsr Nz%Types of user authentications to use.)�logfiler Nz File to log to, or - for stdout� optParameters)�version�Vz Display version number only.)�compress�CzEnable compression.)�log�vz#Enable logging (defaults to stderr))�nox11�xz+Disable X11 connection forwarding (default))�agent�Az&Enable authentication agent forwarding)�noagent�az1Disable authentication agent forwarding (default))� reconnect�rz2Reconnect to the server if the connection is lost.)r" r$ c C � g | ]}|� � �qS � ��decode��.0r r) r) �>/usr/lib/python3/dist-packages/twisted/conch/client/options.py� <listcomp>* � zConchOptions.<listcomp>zciphers to choose from)�descrc C r( r) r* r, r) r) r. r/ . r0 zmacs to choose fromc C r( r) r* r, r) r) r. r/ 2 r0 z"host key algorithms to choose from)r r r r �command�argumentT)r1 �repeat)�mutuallyExclusive� optActions� extraActionsc O s* t jj| g|�R i |�� g | _d | _d S )N)r �Options�__init__� identitys�conns)�self�args�kwr) r) r. r9 ? s zConchOptions.__init__c C s | j �|� dS )z&Identity for public-key authenticationN)r: �append)r< r r) r) r. � opt_identityD s zConchOptions.opt_identityc C s8 |� d�}|D ]}|tjvrt�d| � q|| d< dS )z Select encryption algorithms�,zUnknown cipher type '%s'r N)�splitr � cipherMap�sys�exit)r< r �cipherr) r) r. � opt_ciphersH s � zConchOptions.opt_ciphersc C �L t |t�r |�d�}|�d�}|D ]}|tjvr t�d| � q|| d< dS )zSpecify MAC algorithms�utf-8� ,zUnknown mac type '%r'r N)� isinstance�str�encoderB r �macMaprD rE )r< r �macr) r) r. �opt_macsP � � zConchOptions.opt_macsc C rH )zSelect host key algorithmsrI rJ zUnknown host key type '%r'r N)rK rL rM rB r �supportedPublicKeysrD rE )r< �hkas�hkar) r) r. �opt_host_key_algorithmsZ rQ z$ConchOptions.opt_host_key_algorithmsc C s& t |t�r |�d�}|�d�| d< dS )z/Choose how to authenticate to the remote serverrI rJ r N)rK rL rM rB )r< �uasr) r) r. �opt_user_authenticationsd s z%ConchOptions.opt_user_authenticationsN) �__name__� __module__� __qualname__r r r r rL �int�__annotations__�optFlagsr � Completions�CompleteUsernames�CompleteMultiListr rC �keysrN r rR �CompleteUserAtHost� Completer�compDatar9 r@ rG rP rU rW r) r) r) r. r s^ �� ���� �� r ) rD �typingr r r �twisted.conch.ssh.transportr r �twisted.pythonr r8 r r) r) r) r. �<module> s __pycache__/direct.cpython-310.pyc 0000644 00000007760 15030320052 0012727 0 ustar 00 o �b� � @ s\ d dl mZ d dlmZ d dlmZmZmZ G dd� dej�Z G dd� dej �Z dd � Z d S ) � )�error)� transport)�defer�protocol�reactorc @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S ) �SSHClientFactoryc C s || _ || _|| _|| _d S �N)�d�options� verifyHostKey�userAuthObject)�selfr r r r � r �=/usr/lib/python3/dist-packages/twisted/conch/client/direct.py�__init__ s zSSHClientFactory.__init__c C s | j d r |�� d S d S )N� reconnect)r �connect)r � connector�reasonr r r �clientConnectionLost s �z%SSHClientFactory.clientConnectionLostc C s* | j d u rd S | j d }| _ |�|� d S r )r �errback)r r r r r r r �clientConnectionFailed s z'SSHClientFactory.clientConnectionFailedc C sh t | �}| jd r| jd |_| jd r| jd |_| jd r'dg|jdd�< | jd r2| jd |_|S )N�ciphers�macs�compress�zlibr � zhost-key-algorithms)�SSHClientTransportr �supportedCiphers� supportedMACs�supportedCompressions�supportedPublicKeys)r �addr�transr r r � buildProtocol s z SSHClientFactory.buildProtocolN)�__name__� __module__� __qualname__r r r r$ r r r r r s r c @ sL e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Zd d� Z dd� Z dS )r c C s || _ d | _d S r )�factory� unixServer)r r( r r r r ) s zSSHClientTransport.__init__c s: �j r �j �� }d �_ nt�d �}|�� �fdd�� d S )Nc s t j��� �S r )r r �connectionLost)�x�r r r r �<lambda>4 s z3SSHClientTransport.connectionLost.<locals>.<lambda>)r) � stopListeningr �succeed� addCallback)r r r r r, r r* - s �z!SSHClientTransport.connectionLostc C s8 | j jd u rd S | j jd }| j _|�t�||�� d S r )r( r r r � ConchError)r �code�descr r r r � receiveError7 s z SSHClientTransport.receiveErrorc C sH | j jd u rd S | j jd }| j _tj�| ||� |�t�||�� d S r )r( r r r �sendDisconnectr r r1 )r r2 r r r r r r5 = s z!SSHClientTransport.sendDisconnectc C s( | j jd|||d� |rt|� d S d S )Nz!Received Debug Message: {message})�message� alwaysDisplay�lang)�_log�debug�print)r r7 r6 r8 r r r � receiveDebugD s � �z SSHClientTransport.receiveDebugc C s | j �| | j�� j||�S r )r( r r �getPeer�host)r �pubKey� fingerprintr r r r N s �z SSHClientTransport.verifyHostKeyc C s\ | j jd|d� tj�| |� |jdkr*| jjd ur,| jjd }| j_|�d � d S d S d S )Nz"setting client server to {service})�servicez ssh-userauth) r9 �infor r � setService�namer( r �callback)r rA r r r r rC S s �z SSHClientTransport.setServicec C s | � | jj� d S r )�requestServicer( r )r r r r �connectionSecureZ s z#SSHClientTransport.connectionSecureN) r% r& r' r r* r4 r5 r< r rC rG r r r r r ( s r c C s( t �� }t||||�}t�| ||� |S r )r �Deferredr r � connectTCP)r> �portr r r r r( r r r r ^ s r N) � twisted.conchr �twisted.conch.sshr �twisted.internetr r r � ClientFactoryr r r r r r r �<module> s 6 __pycache__/default.cpython-310.pyc 0000644 00000025624 15030320052 0013100 0 ustar 00 o �b�. � @ s� d Z ddlZddlZddlZddlZddlZddlmZ ddlm Z ddl m Z m Z ddl mZ ddlmZmZmZ ddlmZmZmZ dd lmZ dd lmZ d Ze Z e Z d d � Z dd� Z!dd� Z"G dd� dej#�Z#dS )aN Various classes and functions for implementing user-interaction in the command-line conch client. You probably shouldn't use anything in this module directly, since it assumes you are sitting at an interactive terminal. For example, to programmatically interact with a known_hosts database, use L{twisted.conch.client.knownhosts}. � N)� decodebytes)�agent)� ConsoleUI�KnownHostsFile)� ConchError)�common�keys�userauth)�defer�protocol�reactor)� nativeString)�FilePathz~/.ssh/known_hostsc C sV | j jd }tj�|�}t�t| j jd ptj � t ���}t dd� �}|� ||||�S )a% Verify a host's key. This function is a gross vestige of some bad factoring in the client internals. The actual implementation, and a better signature of this logic is in L{KnownHostsFile.verifyHostKey}. This function is not deprecated yet because the callers have not yet been rehabilitated, but they should eventually be changed to call that method instead. However, this function does perform two functions not implemented by L{KnownHostsFile.verifyHostKey}. It determines the path to the user's known_hosts file based on the options (which should really be the options object's job), and it provides an opener to L{ConsoleUI} which opens '/dev/tty' so that the user will be prompted on the tty of the process even if the input and output of the process has been redirected. This latter part is, somewhat obviously, not portable, but I don't know of a portable equivalent that could be used. @param host: Due to a bug in L{SSHClientTransport.verifyHostKey}, this is always the dotted-quad IP address of the host being connected to. @type host: L{str} @param transport: the client transport which is attempting to connect to the given host. @type transport: L{SSHClientTransport} @param fingerprint: the fingerprint of the given public key, in xx:xx:xx:... format. This is ignored in favor of getting the fingerprint from the key itself. @type fingerprint: L{str} @param pubKey: The public key of the server being connected to. @type pubKey: L{str} @return: a L{Deferred} which fires with C{1} if the key was successfully verified, or fails if the key could not be successfully verified. Failure types may include L{HostKeyChanged}, L{UserRejectedKey}, L{IOError} or L{KeyboardInterrupt}. �host� known-hostsc S s t dddd�S )N�/dev/ttyzr+br )� buffering)�_open� r r �>/usr/lib/python3/dist-packages/twisted/conch/client/default.py�<lambda>V s z verifyHostKey.<locals>.<lambda>)�factory�optionsr �Key� fromStringr �fromPathr �os�path� expanduser� _KNOWN_HOSTSr � verifyHostKey)� transportr �pubKey� fingerprint� actualHost� actualKey�kh�uir r r r ' s ( �� r c C s8 t �|�d }d}|d s$tj�tj�d��s$td� t�tj�d�� |d p)t}z t tj�|�d�}W n t y? Y dS w |�P |� � D ]B}|� � }t |�dk rTqG|dd� \} } } | | � d �vreqG| |krjqGzt| �} W n tyy Y qGw | |kr� W d � d S d }qGW d � |S 1 s�w Y |S ) z� Checks to see if host is in the known_hosts file for the user. @return: 0 if it isn't, 1 if it is and is the same, 2 if it's changed. @rtype: L{int} r r z~/.ssh/z Creating ~/.ssh directory...z~/.ssh�rb� N� ,� � )r �getNSr r �existsr �print�mkdirr �open�OSError� readlines�split�lenr � BaseException) r r" r �keyType�retVal�kh_file� known_hosts�liner4 �hosts� hostKeyType� encodedKey� decodedKeyr r r �isInKnownHostsZ sF � ��� ��r@ c C sX t �t|d p tj�t���}g }|�� D ]}|�| �r'|j |vr'|� |j � q|p+dS )a� Look in known_hosts for a key corresponding to C{host}. This can be used to change the order of supported key types in the KEXINIT packet. @type host: L{str} @param host: the host to check in known_hosts @type options: L{twisted.conch.client.options.ConchOptions} @param options: options passed to client @return: L{list} of L{str} representing key types or L{None}. r N) r r r r r r r � iterentries� matchesHostr7 �append)r r � knownHosts�keyTypes�entryr r r �getHostKeyAlgorithms� s � �rG c @ s� e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Zd dd�Z dd� Z dd� Z dd� Z dd� Z edd� �Zeejdd� ��Zd S ) �SSHUserAuthClientc G s@ t jj| |g|�R � d | _|| _g | _|js ddg|_d S d S )Nz ~/.ssh/id_rsaz ~/.ssh/id_dsa)r rH �__init__�keyAgentr � usedFiles� identitys)�self�userr �argsr r r rI � s �zSSHUserAuthClient.__init__c C st dt jv r2| jd s2| jjdt jd d� t�ttj �}|� t jd �}|� | j � |� | j� d S tj�| � d S )N� SSH_AUTH_SOCK�noagentz using SSH agent {authSock!r})�authSock)r �environr �_log�debugr � ClientCreatorr r �SSHAgentClient� connectUNIX� addCallback� _setAgent� addErrback� _ebSetAgentr rH �serviceStarted)rM �cc�dr r r r] � s � z SSHUserAuthClient.serviceStartedc C s | j r| j j�� d | _ d S d S �N)rJ r! �loseConnection�rM r r r �serviceStopped� s �z SSHUserAuthClient.serviceStoppedc C s || _ | j �� }|�| j� |S r` )rJ � getPublicKeys�addBothr\ )rM �ar_ r r r rZ � s zSSHUserAuthClient._setAgentc C s t j�| � d S r` )r rH r] )rM �fr r r r\ � s z SSHUserAuthClient._ebSetAgentc C s^ | � � �! zt�|�}|W W d � S ttfy$ t� td��w 1 s(w Y dS )z� Prompt for a password using L{getpass.getpass}. @param prompt: Written on tty to ask for the input. @type prompt: L{str} @return: The input. @rtype: L{str} N�PEBKAC)�_replaceStdoutStdin�getpass�KeyboardInterruptr2 r/ r �rM �prompt�pr r r � _getPassword� s ���z SSHUserAuthClient._getPasswordNc C sf |rt |�}nd�t | j�| jj�� j�}z| �|��t� � �}t � |�W S t y2 t � � Y S w )Nz{}@{}'s password: )r �formatrN r! �getPeerr ro �encode�sys�getdefaultencodingr �succeedr �failrl r r r � getPassword� s � �z SSHUserAuthClient.getPasswordc s� � j r� j �� }|dur|S � fdd�� jjD �}� jjd� jj|d� |s(dS |d }� j�|� tj � |�}|d7 }tj � |�sF� �� S zt j �|�W S t jy[ � �� Y S w )z� Get a public key from the key agent if possible, otherwise look in the next configured identity file for one. Nc s g | ] }|� j vr|�qS r )rK )�.0�xrb r r � <listcomp>� s z2SSHUserAuthClient.getPublicKey.<locals>.<listcomp>z+public key identities: {identities} {files})� identities�filesr z.pub)rJ � getPublicKeyr rL rT rU rK rC r r r r. r r �fromFile� BadKeyError)rM �keyr| �filer rb r r} � s. � �z SSHUserAuthClient.getPublicKeyc C s( | j s | j�|�� |�S tj�| ||�S )z� Extend the base signing behavior by using an SSH agent to sign the data, if one is available. @type publicKey: L{Key} @type signData: L{bytes} )rK rJ �signData�blobr rH )rM � publicKeyr� r r r r� � s zSSHUserAuthClient.signDatac C s� t j�| jd �}t j�|�sdS z t�tj� |��W S tj yc t d�D ]:}d| jd }z| � |�� t�� �}t�tjj ||d��W Y S tjtfyV Y nw t�td�� Y S � tys t� t�� Y dS w )z� Try to load the private key from the last used file identified by C{getPublicKey}, potentially asking for the passphrase if the key is encrypted. ���Nr) z Enter passphrase for key '%s': )� passphrasez bad password)r r r rK r. r ru r r r~ �EncryptedKeyError�rangero rr rs �getfilesystemencodingr r rv rk r/ r �stop)rM r� �irm rn r r r � getPrivateKey s( � �z SSHUserAuthClient.getPrivateKeyc C s� g }| � � �8 |rt|�d�� |rt|�d�� |D ]\}}|�d�}|r.|�t|�� q|�t�|�� qW d � n1 sAw Y t�|�S )Nzutf-8)ri r/ �decoderC �_inputrj r ru )rM �name� instruction�prompts� responsesrm �echor r r �getGenericAnswers s � � z#SSHUserAuthClient.getGenericAnswersc C s( t �tdd��}t �tdd��}||fS )a0 Open /dev/tty as two streams one in read, one in write mode, and return them. @return: File objects for reading and writing to /dev/tty, corresponding to standard input and standard output. @rtype: A L{tuple} of L{io.TextIOWrapper} on Python 3. r r( �wb)�io� TextIOWrapperr1 )�cls�stdin�stdoutr r r �_openTty/ s zSSHUserAuthClient._openTtyc c st � t jt j}}| �� \t _t _zdV W t j�� t j�� ||t _t _dS t j�� t j�� ||t _t _w )zv Contextmanager that replaces stdout and stdin with /dev/tty and resets them when it is done. N)rs r� r� r� �close)r� �oldout�oldinr r r ri = s � � z%SSHUserAuthClient._replaceStdoutStdinr` )�__name__� __module__� __qualname__rI r] rc rZ r\ ro rw r} r� r� r� � classmethodr� � contextlib�contextmanagerri r r r r rH � s" rH )$�__doc__r� rj r� r rs �base64r �twisted.conch.clientr � twisted.conch.client.knownhostsr r �twisted.conch.errorr �twisted.conch.sshr r r �twisted.internetr r r �twisted.python.compatr �twisted.python.filepathr r r1 r �inputr� r r@ rG rH r r r r �<module> s* 3' default.py 0000644 00000027227 15030320052 0006542 0 ustar 00 # -*- test-case-name: twisted.conch.test.test_knownhosts,twisted.conch.test.test_default -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Various classes and functions for implementing user-interaction in the command-line conch client. You probably shouldn't use anything in this module directly, since it assumes you are sitting at an interactive terminal. For example, to programmatically interact with a known_hosts database, use L{twisted.conch.client.knownhosts}. """ import contextlib import getpass import io import os import sys from base64 import decodebytes from twisted.conch.client import agent from twisted.conch.client.knownhosts import ConsoleUI, KnownHostsFile from twisted.conch.error import ConchError from twisted.conch.ssh import common, keys, userauth from twisted.internet import defer, protocol, reactor from twisted.python.compat import nativeString from twisted.python.filepath import FilePath # The default location of the known hosts file (probably should be parsed out # of an ssh config file someday). _KNOWN_HOSTS = "~/.ssh/known_hosts" # This name is bound so that the unit tests can use 'patch' to override it. _open = open _input = input def verifyHostKey(transport, host, pubKey, fingerprint): """ Verify a host's key. This function is a gross vestige of some bad factoring in the client internals. The actual implementation, and a better signature of this logic is in L{KnownHostsFile.verifyHostKey}. This function is not deprecated yet because the callers have not yet been rehabilitated, but they should eventually be changed to call that method instead. However, this function does perform two functions not implemented by L{KnownHostsFile.verifyHostKey}. It determines the path to the user's known_hosts file based on the options (which should really be the options object's job), and it provides an opener to L{ConsoleUI} which opens '/dev/tty' so that the user will be prompted on the tty of the process even if the input and output of the process has been redirected. This latter part is, somewhat obviously, not portable, but I don't know of a portable equivalent that could be used. @param host: Due to a bug in L{SSHClientTransport.verifyHostKey}, this is always the dotted-quad IP address of the host being connected to. @type host: L{str} @param transport: the client transport which is attempting to connect to the given host. @type transport: L{SSHClientTransport} @param fingerprint: the fingerprint of the given public key, in xx:xx:xx:... format. This is ignored in favor of getting the fingerprint from the key itself. @type fingerprint: L{str} @param pubKey: The public key of the server being connected to. @type pubKey: L{str} @return: a L{Deferred} which fires with C{1} if the key was successfully verified, or fails if the key could not be successfully verified. Failure types may include L{HostKeyChanged}, L{UserRejectedKey}, L{IOError} or L{KeyboardInterrupt}. """ actualHost = transport.factory.options["host"] actualKey = keys.Key.fromString(pubKey) kh = KnownHostsFile.fromPath( FilePath( transport.factory.options["known-hosts"] or os.path.expanduser(_KNOWN_HOSTS) ) ) ui = ConsoleUI(lambda: _open("/dev/tty", "r+b", buffering=0)) return kh.verifyHostKey(ui, actualHost, host, actualKey) def isInKnownHosts(host, pubKey, options): """ Checks to see if host is in the known_hosts file for the user. @return: 0 if it isn't, 1 if it is and is the same, 2 if it's changed. @rtype: L{int} """ keyType = common.getNS(pubKey)[0] retVal = 0 if not options["known-hosts"] and not os.path.exists(os.path.expanduser("~/.ssh/")): print("Creating ~/.ssh directory...") os.mkdir(os.path.expanduser("~/.ssh")) kh_file = options["known-hosts"] or _KNOWN_HOSTS try: known_hosts = open(os.path.expanduser(kh_file), "rb") except OSError: return 0 with known_hosts: for line in known_hosts.readlines(): split = line.split() if len(split) < 3: continue hosts, hostKeyType, encodedKey = split[:3] if host not in hosts.split(b","): # incorrect host continue if hostKeyType != keyType: # incorrect type of key continue try: decodedKey = decodebytes(encodedKey) except BaseException: continue if decodedKey == pubKey: return 1 else: retVal = 2 return retVal def getHostKeyAlgorithms(host, options): """ Look in known_hosts for a key corresponding to C{host}. This can be used to change the order of supported key types in the KEXINIT packet. @type host: L{str} @param host: the host to check in known_hosts @type options: L{twisted.conch.client.options.ConchOptions} @param options: options passed to client @return: L{list} of L{str} representing key types or L{None}. """ knownHosts = KnownHostsFile.fromPath( FilePath(options["known-hosts"] or os.path.expanduser(_KNOWN_HOSTS)) ) keyTypes = [] for entry in knownHosts.iterentries(): if entry.matchesHost(host): if entry.keyType not in keyTypes: keyTypes.append(entry.keyType) return keyTypes or None class SSHUserAuthClient(userauth.SSHUserAuthClient): def __init__(self, user, options, *args): userauth.SSHUserAuthClient.__init__(self, user, *args) self.keyAgent = None self.options = options self.usedFiles = [] if not options.identitys: options.identitys = ["~/.ssh/id_rsa", "~/.ssh/id_dsa"] def serviceStarted(self): if "SSH_AUTH_SOCK" in os.environ and not self.options["noagent"]: self._log.debug( "using SSH agent {authSock!r}", authSock=os.environ["SSH_AUTH_SOCK"] ) cc = protocol.ClientCreator(reactor, agent.SSHAgentClient) d = cc.connectUNIX(os.environ["SSH_AUTH_SOCK"]) d.addCallback(self._setAgent) d.addErrback(self._ebSetAgent) else: userauth.SSHUserAuthClient.serviceStarted(self) def serviceStopped(self): if self.keyAgent: self.keyAgent.transport.loseConnection() self.keyAgent = None def _setAgent(self, a): self.keyAgent = a d = self.keyAgent.getPublicKeys() d.addBoth(self._ebSetAgent) return d def _ebSetAgent(self, f): userauth.SSHUserAuthClient.serviceStarted(self) def _getPassword(self, prompt): """ Prompt for a password using L{getpass.getpass}. @param prompt: Written on tty to ask for the input. @type prompt: L{str} @return: The input. @rtype: L{str} """ with self._replaceStdoutStdin(): try: p = getpass.getpass(prompt) return p except (KeyboardInterrupt, OSError): print() raise ConchError("PEBKAC") def getPassword(self, prompt=None): if prompt: prompt = nativeString(prompt) else: prompt = "{}@{}'s password: ".format( nativeString(self.user), self.transport.transport.getPeer().host, ) try: # We don't know the encoding the other side is using, # signaling that is not part of the SSH protocol. But # using our defaultencoding is better than just going for # ASCII. p = self._getPassword(prompt).encode(sys.getdefaultencoding()) return defer.succeed(p) except ConchError: return defer.fail() def getPublicKey(self): """ Get a public key from the key agent if possible, otherwise look in the next configured identity file for one. """ if self.keyAgent: key = self.keyAgent.getPublicKey() if key is not None: return key files = [x for x in self.options.identitys if x not in self.usedFiles] self._log.debug( "public key identities: {identities}\n{files}", identities=self.options.identitys, files=files, ) if not files: return None file = files[0] self.usedFiles.append(file) file = os.path.expanduser(file) file += ".pub" if not os.path.exists(file): return self.getPublicKey() # try again try: return keys.Key.fromFile(file) except keys.BadKeyError: return self.getPublicKey() # try again def signData(self, publicKey, signData): """ Extend the base signing behavior by using an SSH agent to sign the data, if one is available. @type publicKey: L{Key} @type signData: L{bytes} """ if not self.usedFiles: # agent key return self.keyAgent.signData(publicKey.blob(), signData) else: return userauth.SSHUserAuthClient.signData(self, publicKey, signData) def getPrivateKey(self): """ Try to load the private key from the last used file identified by C{getPublicKey}, potentially asking for the passphrase if the key is encrypted. """ file = os.path.expanduser(self.usedFiles[-1]) if not os.path.exists(file): return None try: return defer.succeed(keys.Key.fromFile(file)) except keys.EncryptedKeyError: for i in range(3): prompt = "Enter passphrase for key '%s': " % self.usedFiles[-1] try: p = self._getPassword(prompt).encode(sys.getfilesystemencoding()) return defer.succeed(keys.Key.fromFile(file, passphrase=p)) except (keys.BadKeyError, ConchError): pass return defer.fail(ConchError("bad password")) raise except KeyboardInterrupt: print() reactor.stop() def getGenericAnswers(self, name, instruction, prompts): responses = [] with self._replaceStdoutStdin(): if name: print(name.decode("utf-8")) if instruction: print(instruction.decode("utf-8")) for prompt, echo in prompts: prompt = prompt.decode("utf-8") if echo: responses.append(_input(prompt)) else: responses.append(getpass.getpass(prompt)) return defer.succeed(responses) @classmethod def _openTty(cls): """ Open /dev/tty as two streams one in read, one in write mode, and return them. @return: File objects for reading and writing to /dev/tty, corresponding to standard input and standard output. @rtype: A L{tuple} of L{io.TextIOWrapper} on Python 3. """ stdin = io.TextIOWrapper(open("/dev/tty", "rb")) stdout = io.TextIOWrapper(open("/dev/tty", "wb")) return stdin, stdout @classmethod @contextlib.contextmanager def _replaceStdoutStdin(cls): """ Contextmanager that replaces stdout and stdin with /dev/tty and resets them when it is done. """ oldout, oldin = sys.stdout, sys.stdin sys.stdin, sys.stdout = cls._openTty() try: yield finally: sys.stdout.close() sys.stdin.close() sys.stdout, sys.stdin = oldout, oldin direct.py 0000644 00000006360 15030320052 0006363 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. from twisted.conch import error from twisted.conch.ssh import transport from twisted.internet import defer, protocol, reactor class SSHClientFactory(protocol.ClientFactory): def __init__(self, d, options, verifyHostKey, userAuthObject): self.d = d self.options = options self.verifyHostKey = verifyHostKey self.userAuthObject = userAuthObject def clientConnectionLost(self, connector, reason): if self.options["reconnect"]: connector.connect() def clientConnectionFailed(self, connector, reason): if self.d is None: return d, self.d = self.d, None d.errback(reason) def buildProtocol(self, addr): trans = SSHClientTransport(self) if self.options["ciphers"]: trans.supportedCiphers = self.options["ciphers"] if self.options["macs"]: trans.supportedMACs = self.options["macs"] if self.options["compress"]: trans.supportedCompressions[0:1] = ["zlib"] if self.options["host-key-algorithms"]: trans.supportedPublicKeys = self.options["host-key-algorithms"] return trans class SSHClientTransport(transport.SSHClientTransport): def __init__(self, factory): self.factory = factory self.unixServer = None def connectionLost(self, reason): if self.unixServer: d = self.unixServer.stopListening() self.unixServer = None else: d = defer.succeed(None) d.addCallback( lambda x: transport.SSHClientTransport.connectionLost(self, reason) ) def receiveError(self, code, desc): if self.factory.d is None: return d, self.factory.d = self.factory.d, None d.errback(error.ConchError(desc, code)) def sendDisconnect(self, code, reason): if self.factory.d is None: return d, self.factory.d = self.factory.d, None transport.SSHClientTransport.sendDisconnect(self, code, reason) d.errback(error.ConchError(reason, code)) def receiveDebug(self, alwaysDisplay, message, lang): self._log.debug( "Received Debug Message: {message}", message=message, alwaysDisplay=alwaysDisplay, lang=lang, ) if alwaysDisplay: # XXX what should happen here? print(message) def verifyHostKey(self, pubKey, fingerprint): return self.factory.verifyHostKey( self, self.transport.getPeer().host, pubKey, fingerprint ) def setService(self, service): self._log.info("setting client server to {service}", service=service) transport.SSHClientTransport.setService(self, service) if service.name != "ssh-userauth" and self.factory.d is not None: d, self.factory.d = self.factory.d, None d.callback(None) def connectionSecure(self): self.requestService(self.factory.userAuthObject) def connect(host, port, options, verifyHostKey, userAuthObject): d = defer.Deferred() factory = SSHClientFactory(d, options, verifyHostKey, userAuthObject) reactor.connectTCP(host, port, factory) return d knownhosts.py 0000644 00000046706 15030320052 0007336 0 ustar 00 # -*- test-case-name: twisted.conch.test.test_knownhosts -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ An implementation of the OpenSSH known_hosts database. @since: 8.2 """ import hmac import sys from binascii import Error as DecodeError, a2b_base64, b2a_base64 from contextlib import closing from hashlib import sha1 from zope.interface import implementer from twisted.conch.error import HostKeyChanged, InvalidEntry, UserRejectedKey from twisted.conch.interfaces import IKnownHostEntry from twisted.conch.ssh.keys import BadKeyError, FingerprintFormats, Key from twisted.internet import defer from twisted.logger import Logger from twisted.python.compat import nativeString from twisted.python.randbytes import secureRandom from twisted.python.util import FancyEqMixin log = Logger() def _b64encode(s): """ Encode a binary string as base64 with no trailing newline. @param s: The string to encode. @type s: L{bytes} @return: The base64-encoded string. @rtype: L{bytes} """ return b2a_base64(s).strip() def _extractCommon(string): """ Extract common elements of base64 keys from an entry in a hosts file. @param string: A known hosts file entry (a single line). @type string: L{bytes} @return: a 4-tuple of hostname data (L{bytes}), ssh key type (L{bytes}), key (L{Key}), and comment (L{bytes} or L{None}). The hostname data is simply the beginning of the line up to the first occurrence of whitespace. @rtype: L{tuple} """ elements = string.split(None, 2) if len(elements) != 3: raise InvalidEntry() hostnames, keyType, keyAndComment = elements splitkey = keyAndComment.split(None, 1) if len(splitkey) == 2: keyString, comment = splitkey comment = comment.rstrip(b"\n") else: keyString = splitkey[0] comment = None key = Key.fromString(a2b_base64(keyString)) return hostnames, keyType, key, comment class _BaseEntry: """ Abstract base of both hashed and non-hashed entry objects, since they represent keys and key types the same way. @ivar keyType: The type of the key; either ssh-dss or ssh-rsa. @type keyType: L{bytes} @ivar publicKey: The server public key indicated by this line. @type publicKey: L{twisted.conch.ssh.keys.Key} @ivar comment: Trailing garbage after the key line. @type comment: L{bytes} """ def __init__(self, keyType, publicKey, comment): self.keyType = keyType self.publicKey = publicKey self.comment = comment def matchesKey(self, keyObject): """ Check to see if this entry matches a given key object. @param keyObject: A public key object to check. @type keyObject: L{Key} @return: C{True} if this entry's key matches C{keyObject}, C{False} otherwise. @rtype: L{bool} """ return self.publicKey == keyObject @implementer(IKnownHostEntry) class PlainEntry(_BaseEntry): """ A L{PlainEntry} is a representation of a plain-text entry in a known_hosts file. @ivar _hostnames: the list of all host-names associated with this entry. @type _hostnames: L{list} of L{bytes} """ def __init__(self, hostnames, keyType, publicKey, comment): self._hostnames = hostnames super().__init__(keyType, publicKey, comment) @classmethod def fromString(cls, string): """ Parse a plain-text entry in a known_hosts file, and return a corresponding L{PlainEntry}. @param string: a space-separated string formatted like "hostname key-type base64-key-data comment". @type string: L{bytes} @raise DecodeError: if the key is not valid encoded as valid base64. @raise InvalidEntry: if the entry does not have the right number of elements and is therefore invalid. @raise BadKeyError: if the key, once decoded from base64, is not actually an SSH key. @return: an IKnownHostEntry representing the hostname and key in the input line. @rtype: L{PlainEntry} """ hostnames, keyType, key, comment = _extractCommon(string) self = cls(hostnames.split(b","), keyType, key, comment) return self def matchesHost(self, hostname): """ Check to see if this entry matches a given hostname. @param hostname: A hostname or IP address literal to check against this entry. @type hostname: L{bytes} @return: C{True} if this entry is for the given hostname or IP address, C{False} otherwise. @rtype: L{bool} """ if isinstance(hostname, str): hostname = hostname.encode("utf-8") return hostname in self._hostnames def toString(self): """ Implement L{IKnownHostEntry.toString} by recording the comma-separated hostnames, key type, and base-64 encoded key. @return: The string representation of this entry, with unhashed hostname information. @rtype: L{bytes} """ fields = [ b",".join(self._hostnames), self.keyType, _b64encode(self.publicKey.blob()), ] if self.comment is not None: fields.append(self.comment) return b" ".join(fields) @implementer(IKnownHostEntry) class UnparsedEntry: """ L{UnparsedEntry} is an entry in a L{KnownHostsFile} which can't actually be parsed; therefore it matches no keys and no hosts. """ def __init__(self, string): """ Create an unparsed entry from a line in a known_hosts file which cannot otherwise be parsed. """ self._string = string def matchesHost(self, hostname): """ Always returns False. """ return False def matchesKey(self, key): """ Always returns False. """ return False def toString(self): """ Returns the input line, without its newline if one was given. @return: The string representation of this entry, almost exactly as was used to initialize this entry but without a trailing newline. @rtype: L{bytes} """ return self._string.rstrip(b"\n") def _hmacedString(key, string): """ Return the SHA-1 HMAC hash of the given key and string. @param key: The HMAC key. @type key: L{bytes} @param string: The string to be hashed. @type string: L{bytes} @return: The keyed hash value. @rtype: L{bytes} """ hash = hmac.HMAC(key, digestmod=sha1) if isinstance(string, str): string = string.encode("utf-8") hash.update(string) return hash.digest() @implementer(IKnownHostEntry) class HashedEntry(_BaseEntry, FancyEqMixin): """ A L{HashedEntry} is a representation of an entry in a known_hosts file where the hostname has been hashed and salted. @ivar _hostSalt: the salt to combine with a hostname for hashing. @ivar _hostHash: the hashed representation of the hostname. @cvar MAGIC: the 'hash magic' string used to identify a hashed line in a known_hosts file as opposed to a plaintext one. """ MAGIC = b"|1|" compareAttributes = ("_hostSalt", "_hostHash", "keyType", "publicKey", "comment") def __init__(self, hostSalt, hostHash, keyType, publicKey, comment): self._hostSalt = hostSalt self._hostHash = hostHash super().__init__(keyType, publicKey, comment) @classmethod def fromString(cls, string): """ Load a hashed entry from a string representing a line in a known_hosts file. @param string: A complete single line from a I{known_hosts} file, formatted as defined by OpenSSH. @type string: L{bytes} @raise DecodeError: if the key, the hostname, or the is not valid encoded as valid base64 @raise InvalidEntry: if the entry does not have the right number of elements and is therefore invalid, or the host/hash portion contains more items than just the host and hash. @raise BadKeyError: if the key, once decoded from base64, is not actually an SSH key. @return: The newly created L{HashedEntry} instance, initialized with the information from C{string}. """ stuff, keyType, key, comment = _extractCommon(string) saltAndHash = stuff[len(cls.MAGIC) :].split(b"|") if len(saltAndHash) != 2: raise InvalidEntry() hostSalt, hostHash = saltAndHash self = cls(a2b_base64(hostSalt), a2b_base64(hostHash), keyType, key, comment) return self def matchesHost(self, hostname): """ Implement L{IKnownHostEntry.matchesHost} to compare the hash of the input to the stored hash. @param hostname: A hostname or IP address literal to check against this entry. @type hostname: L{bytes} @return: C{True} if this entry is for the given hostname or IP address, C{False} otherwise. @rtype: L{bool} """ return hmac.compare_digest( _hmacedString(self._hostSalt, hostname), self._hostHash ) def toString(self): """ Implement L{IKnownHostEntry.toString} by base64-encoding the salt, host hash, and key. @return: The string representation of this entry, with the hostname part hashed. @rtype: L{bytes} """ fields = [ self.MAGIC + b"|".join([_b64encode(self._hostSalt), _b64encode(self._hostHash)]), self.keyType, _b64encode(self.publicKey.blob()), ] if self.comment is not None: fields.append(self.comment) return b" ".join(fields) class KnownHostsFile: """ A structured representation of an OpenSSH-format ~/.ssh/known_hosts file. @ivar _added: A list of L{IKnownHostEntry} providers which have been added to this instance in memory but not yet saved. @ivar _clobber: A flag indicating whether the current contents of the save path will be disregarded and potentially overwritten or not. If C{True}, this will be done. If C{False}, entries in the save path will be read and new entries will be saved by appending rather than overwriting. @type _clobber: L{bool} @ivar _savePath: See C{savePath} parameter of L{__init__}. """ def __init__(self, savePath): """ Create a new, empty KnownHostsFile. Unless you want to erase the current contents of C{savePath}, you want to use L{KnownHostsFile.fromPath} instead. @param savePath: The L{FilePath} to which to save new entries. @type savePath: L{FilePath} """ self._added = [] self._savePath = savePath self._clobber = True @property def savePath(self): """ @see: C{savePath} parameter of L{__init__} """ return self._savePath def iterentries(self): """ Iterate over the host entries in this file. @return: An iterable the elements of which provide L{IKnownHostEntry}. There is an element for each entry in the file as well as an element for each added but not yet saved entry. @rtype: iterable of L{IKnownHostEntry} providers """ for entry in self._added: yield entry if self._clobber: return try: fp = self._savePath.open() except OSError: return with fp: for line in fp: try: if line.startswith(HashedEntry.MAGIC): entry = HashedEntry.fromString(line) else: entry = PlainEntry.fromString(line) except (DecodeError, InvalidEntry, BadKeyError): entry = UnparsedEntry(line) yield entry def hasHostKey(self, hostname, key): """ Check for an entry with matching hostname and key. @param hostname: A hostname or IP address literal to check for. @type hostname: L{bytes} @param key: The public key to check for. @type key: L{Key} @return: C{True} if the given hostname and key are present in this file, C{False} if they are not. @rtype: L{bool} @raise HostKeyChanged: if the host key found for the given hostname does not match the given key. """ for lineidx, entry in enumerate(self.iterentries(), -len(self._added)): if entry.matchesHost(hostname) and entry.keyType == key.sshType(): if entry.matchesKey(key): return True else: # Notice that lineidx is 0-based but HostKeyChanged.lineno # is 1-based. if lineidx < 0: line = None path = None else: line = lineidx + 1 path = self._savePath raise HostKeyChanged(entry, path, line) return False def verifyHostKey(self, ui, hostname, ip, key): """ Verify the given host key for the given IP and host, asking for confirmation from, and notifying, the given UI about changes to this file. @param ui: The user interface to request an IP address from. @param hostname: The hostname that the user requested to connect to. @param ip: The string representation of the IP address that is actually being connected to. @param key: The public key of the server. @return: a L{Deferred} that fires with True when the key has been verified, or fires with an errback when the key either cannot be verified or has changed. @rtype: L{Deferred} """ hhk = defer.maybeDeferred(self.hasHostKey, hostname, key) def gotHasKey(result): if result: if not self.hasHostKey(ip, key): ui.warn( "Warning: Permanently added the %s host key for " "IP address '%s' to the list of known hosts." % (key.type(), nativeString(ip)) ) self.addHostKey(ip, key) self.save() return result else: def promptResponse(response): if response: self.addHostKey(hostname, key) self.addHostKey(ip, key) self.save() return response else: raise UserRejectedKey() keytype = key.type() if keytype == "EC": keytype = "ECDSA" prompt = ( "The authenticity of host '%s (%s)' " "can't be established.\n" "%s key fingerprint is SHA256:%s.\n" "Are you sure you want to continue connecting (yes/no)? " % ( nativeString(hostname), nativeString(ip), keytype, key.fingerprint(format=FingerprintFormats.SHA256_BASE64), ) ) proceed = ui.prompt(prompt.encode(sys.getdefaultencoding())) return proceed.addCallback(promptResponse) return hhk.addCallback(gotHasKey) def addHostKey(self, hostname, key): """ Add a new L{HashedEntry} to the key database. Note that you still need to call L{KnownHostsFile.save} if you wish these changes to be persisted. @param hostname: A hostname or IP address literal to associate with the new entry. @type hostname: L{bytes} @param key: The public key to associate with the new entry. @type key: L{Key} @return: The L{HashedEntry} that was added. @rtype: L{HashedEntry} """ salt = secureRandom(20) keyType = key.sshType() entry = HashedEntry(salt, _hmacedString(salt, hostname), keyType, key, None) self._added.append(entry) return entry def save(self): """ Save this L{KnownHostsFile} to the path it was loaded from. """ p = self._savePath.parent() if not p.isdir(): p.makedirs() if self._clobber: mode = "wb" else: mode = "ab" with self._savePath.open(mode) as hostsFileObj: if self._added: hostsFileObj.write( b"\n".join([entry.toString() for entry in self._added]) + b"\n" ) self._added = [] self._clobber = False @classmethod def fromPath(cls, path): """ Create a new L{KnownHostsFile}, potentially reading existing known hosts information from the given file. @param path: A path object to use for both reading contents from and later saving to. If no file exists at this path, it is not an error; a L{KnownHostsFile} with no entries is returned. @type path: L{FilePath} @return: A L{KnownHostsFile} initialized with entries from C{path}. @rtype: L{KnownHostsFile} """ knownHosts = cls(path) knownHosts._clobber = False return knownHosts class ConsoleUI: """ A UI object that can ask true/false questions and post notifications on the console, to be used during key verification. """ def __init__(self, opener): """ @param opener: A no-argument callable which should open a console binary-mode file-like object to be used for reading and writing. This initializes the C{opener} attribute. @type opener: callable taking no arguments and returning a read/write file-like object """ self.opener = opener def prompt(self, text): """ Write the given text as a prompt to the console output, then read a result from the console input. @param text: Something to present to a user to solicit a yes or no response. @type text: L{bytes} @return: a L{Deferred} which fires with L{True} when the user answers 'yes' and L{False} when the user answers 'no'. It may errback if there were any I/O errors. """ d = defer.succeed(None) def body(ignored): with closing(self.opener()) as f: f.write(text) while True: answer = f.readline().strip().lower() if answer == b"yes": return True elif answer == b"no": return False else: f.write(b"Please type 'yes' or 'no': ") return d.addCallback(body) def warn(self, text): """ Notify the user (non-interactively) of the provided text, by writing it to the console. @param text: Some information the user is to be made aware of. @type text: L{bytes} """ try: with closing(self.opener()) as f: f.write(text) except Exception: log.failure("Failed to write to console") connect.py 0000644 00000001231 15030320052 0006532 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. # from twisted.conch.client import direct connectTypes = {"direct": direct.connect} def connect(host, port, options, verifyHostKey, userAuthObject): useConnects = ["direct"] return _ebConnect( None, useConnects, host, port, options, verifyHostKey, userAuthObject ) def _ebConnect(f, useConnects, host, port, options, vhk, uao): if not useConnects: return f connectType = useConnects.pop(0) f = connectTypes[connectType] d = f(host, port, options, vhk, uao) d.addErrback(_ebConnect, useConnects, host, port, options, vhk, uao) return d options.py 0000644 00000007742 15030320052 0006611 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. import sys from typing import List, Optional, Union # from twisted.conch.ssh.transport import SSHCiphers, SSHClientTransport from twisted.python import usage class ConchOptions(usage.Options): optParameters: List[List[Optional[Union[str, int]]]] = [ ["user", "l", None, "Log in using this user name."], ["identity", "i", None], ["ciphers", "c", None], ["macs", "m", None], ["port", "p", None, "Connect to this port. Server must be on the same port."], ["option", "o", None, "Ignored OpenSSH options"], ["host-key-algorithms", "", None], ["known-hosts", "", None, "File to check for host keys"], ["user-authentications", "", None, "Types of user authentications to use."], ["logfile", "", None, "File to log to, or - for stdout"], ] optFlags = [ ["version", "V", "Display version number only."], ["compress", "C", "Enable compression."], ["log", "v", "Enable logging (defaults to stderr)"], ["nox11", "x", "Disable X11 connection forwarding (default)"], ["agent", "A", "Enable authentication agent forwarding"], ["noagent", "a", "Disable authentication agent forwarding (default)"], ["reconnect", "r", "Reconnect to the server if the connection is lost."], ] compData = usage.Completions( mutuallyExclusive=[("agent", "noagent")], optActions={ "user": usage.CompleteUsernames(), "ciphers": usage.CompleteMultiList( [v.decode() for v in SSHCiphers.cipherMap.keys()], descr="ciphers to choose from", ), "macs": usage.CompleteMultiList( [v.decode() for v in SSHCiphers.macMap.keys()], descr="macs to choose from", ), "host-key-algorithms": usage.CompleteMultiList( [v.decode() for v in SSHClientTransport.supportedPublicKeys], descr="host key algorithms to choose from", ), # "user-authentications": usage.CompleteMultiList(? # descr='user authentication types' ), }, extraActions=[ usage.CompleteUserAtHost(), usage.Completer(descr="command"), usage.Completer(descr="argument", repeat=True), ], ) def __init__(self, *args, **kw): usage.Options.__init__(self, *args, **kw) self.identitys = [] self.conns = None def opt_identity(self, i): """Identity for public-key authentication""" self.identitys.append(i) def opt_ciphers(self, ciphers): "Select encryption algorithms" ciphers = ciphers.split(",") for cipher in ciphers: if cipher not in SSHCiphers.cipherMap: sys.exit("Unknown cipher type '%s'" % cipher) self["ciphers"] = ciphers def opt_macs(self, macs): "Specify MAC algorithms" if isinstance(macs, str): macs = macs.encode("utf-8") macs = macs.split(b",") for mac in macs: if mac not in SSHCiphers.macMap: sys.exit("Unknown mac type '%r'" % mac) self["macs"] = macs def opt_host_key_algorithms(self, hkas): "Select host key algorithms" if isinstance(hkas, str): hkas = hkas.encode("utf-8") hkas = hkas.split(b",") for hka in hkas: if hka not in SSHClientTransport.supportedPublicKeys: sys.exit("Unknown host key type '%r'" % hka) self["host-key-algorithms"] = hkas def opt_user_authentications(self, uas): "Choose how to authenticate to the remote server" if isinstance(uas, str): uas = uas.encode("utf-8") self["user-authentications"] = uas.split(b",") # def opt_compress(self): # "Enable compression" # self.enableCompression = 1 # SSHClientTransport.supportedCompressions[0:1] = ['zlib'] __init__.py 0000644 00000000213 15030320052 0006637 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. # """ Client support code for Conch. Maintainer: Paul Swartz """
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings