Python paramiko 模块，Agent() 实例源码

def agent_auth(transport, username):
    """
    Attempt to authenticate to the given transport using any of the private
    keys available from an SSH agent.
    """

    agent = paramiko.Agent()
    agent_keys = agent.get_keys()
    if len(agent_keys) == 0:
        return

    for key in agent_keys:
        print('Trying ssh-agent key %s' % hexlify(key.get_fingerprint()))
        try:
            transport.auth_publickey(username, key)
            print('... success!')
            return
        except paramiko.SSHException:
            print('... nope.')

def agent_auth(self):
        """
        ????,????????,?????????
        :return:
        """
        agent = paramiko.Agent()
        agent_keys = agent.get_keys()
        if len(agent_keys) == 0:
            return False
        for key in agent_keys:
            print('Trying ssh-agent key %s' % hexlify(key.get_fingerprint()))
            try:
                self.transport.auth_publickey(self.username, key)
                print('... success!')
            except paramiko.SSHException:
                print('... nope.')

def agent_auth(transport, username):
    """
    Attempt to authenticate to the given transport using any of the private
    keys available from an SSH agent.
    """

    agent = paramiko.Agent()
    agent_keys = agent.get_keys()
    if len(agent_keys) == 0:
        return

    for key in agent_keys:
        print('Trying ssh-agent key %s' % hexlify(key.get_fingerprint()))
        try:
            transport.auth_publickey(username, key)
            print('... success!')
            return
        except paramiko.SSHException:
            print('... nope.')

def encrypt(**a):
    """Encrypt contents of INPUT file to OUTPUT file.

    To read from STDIN or write to STDOUT, specify '-' as the INPUT or OUTPUT
    file respectively.
    """

    key = None
    if a['key'] is not None:
        key = a['key'].replace(':','').lower()
        if len(key) != 32:
            raise click.ClickException('Invalid key specified')

    try:
        if key is None:
            keys = paramiko.Agent().get_keys()
            if not keys:
                raise AgentKeyError(AgentKeyError.E_NO_KEYS)
            click.echo('Key not specified.  Please select from the following...')
            for i, k in enumerate(keys):
                click.echo('[%s] %s %s' % (i+1, k.get_name(), to_hex(k.get_fingerprint())))
            i = 0
            while i > len(keys) or i < 1:
                i = click.prompt('Selection (1..%s):' % len(keys), default=1)
            key = to_hex(keys[i - 1].get_fingerprint())

        if a['input'] == '-':
            click.echo('Reading from STDIN...\n')
            text = sys.stdin.read()
        else:
            with open(a['input'], 'r') as f:
                text = f.read()

        text = encrypt_string(text, key)

        if a['output'] == '-':
            sys.stdout.write(text)
        else:
            _check_output_file(a)
            with open(a['output'], 'wb') as f:
                f.write(text)

    except AgentKeyError as e:
        raise click.ClickException(str(e))

def sign_via_agent(data, fingerprint=None):
    """Attempt to sign 'data' via ssh-agent.

    Args:
        data (str):
            The data to sign

    Kwargs:
        fingerprint (str, optional):
            The fingerprint of an SSH public key associated with the private key
            to be used for signing data.

    Returns:
        A dict containing the following keys:
            key_fingerprint:
                The SSH public key fingerprint associated with the private key
                used for signing 'data'.
            key_type: The SSH key type used for signing.
            signature: The data signature returned from ssh-agent.
    Raises:
        AgentKeyError: An error occured while signing.
    """

    agent = paramiko.Agent()
    keys = agent.get_keys()
    sign_key = None
    key_fp = None

    if not keys:
        raise AgentKeyError(AgentKeyError.E_NO_KEYS)

    if fingerprint is not None:
        for key in keys:
            key_fp = key.get_fingerprint()
            if fingerprint == key_fp:
                sign_key = key
                break
        if sign_key is None:
            raise AgentKeyError(AgentKeyError.E_MISSING_KEY, fingerprint=to_hex(fingerprint))
    else:
        sign_key = keys[0]
        key_fp = sign_key.get_fingerprint()

    if PARAMIKO_VER >= (1, 14, 0):
        sig = sign_key.sign_ssh_data(data)
    else:
        sig = sign_key.sign_ssh_data(None, data)

    sig = paramiko.message.Message(sig)

    return {
        'key_fingerprint': key_fp,
        'key_type': sig.get_string(),
        'signature': sig.get_string()
    }

def _load_from_agent(self):
        agent = paramiko.Agent()
        return agent.get_keys()