FM / paramiko

/opt/imh-python/lib/python3.9/site-packages/paramiko/

/opt/imh-python/lib/python3.9/site-packages/paramiko

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
agent.py	15877	0644	edit dl rm
auth_handler.py	42964	0644	edit dl rm
auth_handler.py.orig	43006	0644	edit dl rm
auth_strategy.py	11437	0644	edit dl rm
ber.py	4369	0644	edit dl rm
buffered_pipe.py	7225	0644	edit dl rm
channel.py	49222	0644	edit dl rm
client.py	34492	0644	edit dl rm
common.py	7756	0644	edit dl rm
compress.py	1282	0644	edit dl rm
config.py	27362	0644	edit dl rm
dsskey.py	8248	0644	edit dl rm
ecdsakey.py	11653	0644	edit dl rm
ed25519key.py	7457	0644	edit dl rm
file.py	19063	0644	edit dl rm
hostkeys.py	13208	0644	edit dl rm
kex_curve25519.py	4436	0644	edit dl rm
kex_ecdh_nist.py	5012	0644	edit dl rm
kex_gex.py	10320	0644	edit dl rm
kex_group1.py	5740	0644	edit dl rm
kex_group14.py	1833	0644	edit dl rm
kex_group16.py	2288	0644	edit dl rm
kex_gss.py	24562	0644	edit dl rm
message.py	9349	0644	edit dl rm
packet.py	24314	0644	edit dl rm
pipe.py	3902	0644	edit dl rm
pkey.py	36851	0644	edit dl rm
primes.py	5107	0644	edit dl rm
proxy.py	4648	0644	edit dl rm
rsakey.py	7546	0644	edit dl rm
server.py	30457	0644	edit dl rm
sftp.py	6471	0644	edit dl rm
sftp_attr.py	8258	0644	edit dl rm
sftp_client.py	35855	0644	edit dl rm
sftp_file.py	21820	0644	edit dl rm
sftp_handle.py	7424	0644	edit dl rm
sftp_server.py	19492	0644	edit dl rm
sftp_si.py	12544	0644	edit dl rm
ssh_exception.py	7494	0644	edit dl rm
ssh_gss.py	28887	0644	edit dl rm
transport.py	135632	0644	edit dl rm
util.py	9550	0644	edit dl rm
win_openssh.py	1918	0644	edit dl rm
win_pageant.py	4177	0644	edit dl rm
_version.py	80	0644	edit dl rm
_winapi.py	11204	0644	edit dl rm
__init__.py	4446	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/paramiko/auth_strategy.py (11437B)

"""
Modern, adaptable authentication machinery.

Replaces certain parts of `.SSHClient`. For a concrete implementation, see the
``OpenSSHAuthStrategy`` class in `Fabric <https://fabfile.org>`_.
"""

from collections import namedtuple

from .agent import AgentKey
from .util import get_logger
from .ssh_exception import AuthenticationException

class AuthSource:
    """
    Some SSH authentication source, such as a password, private key, or agent.

See subclasses in this module for concrete implementations.

All implementations must accept at least a ``username`` (``str``) kwarg.
    """

def __init__(self, username):
        self.username = username

def _repr(self, **kwargs):
        # TODO: are there any good libs for this? maybe some helper from
        # structlog?
        pairs = [f"{k}={v!r}" for k, v in kwargs.items()]
        joined = ", ".join(pairs)
        return f"{self.__class__.__name__}({joined})"

def __repr__(self):
        return self._repr()

def authenticate(self, transport):
        """
        Perform authentication.
        """
        raise NotImplementedError

class NoneAuth(AuthSource):
    """
    Auth type "none", ie https://www.rfc-editor.org/rfc/rfc4252#section-5.2 .
    """

def authenticate(self, transport):
        return transport.auth_none(self.username)

class Password(AuthSource):
    """
    Password authentication.

:param callable password_getter:
        A lazy callable that should return a `str` password value at
        authentication time, such as a `functools.partial` wrapping
        `getpass.getpass`, an API call to a secrets store, or similar.

If you already know the password at instantiation time, you should
        simply use something like ``lambda: "my literal"`` (for a literal, but
        also, shame on you!) or ``lambda: variable_name`` (for something stored
        in a variable).
    """

def __init__(self, username, password_getter):
        super().__init__(username=username)
        self.password_getter = password_getter

def __repr__(self):
        # Password auth is marginally more 'username-caring' than pkeys, so may
        # as well log that info here.
        return super()._repr(user=self.username)

def authenticate(self, transport):
        # Lazily get the password, in case it's prompting a user
        # TODO: be nice to log source _of_ the password?
        password = self.password_getter()
        return transport.auth_password(self.username, password)

# TODO 4.0: twiddle this, or PKey, or both, so they're more obviously distinct.
# TODO 4.0: the obvious is to make this more wordy (PrivateKeyAuth), the
# minimalist approach might be to rename PKey to just Key (esp given all the
# subclasses are WhateverKey and not WhateverPKey)
class PrivateKey(AuthSource):
    """
    Essentially a mixin for private keys.

Knows how to auth, but leaves key material discovery/loading/decryption to
    subclasses.

Subclasses **must** ensure that they've set ``self.pkey`` to a decrypted
    `.PKey` instance before calling ``super().authenticate``; typically
    either in their ``__init__``, or in an overridden ``authenticate`` prior to
    its `super` call.
    """

def authenticate(self, transport):
        return transport.auth_publickey(self.username, self.pkey)

class InMemoryPrivateKey(PrivateKey):
    """
    An in-memory, decrypted `.PKey` object.
    """

def __init__(self, username, pkey):
        super().__init__(username=username)
        # No decryption (presumably) necessary!
        self.pkey = pkey

def __repr__(self):
        # NOTE: most of interesting repr-bits for private keys is in PKey.
        # TODO: tacking on agent-ness like this is a bit awkward, but, eh?
        rep = super()._repr(pkey=self.pkey)
        if isinstance(self.pkey, AgentKey):
            rep += " [agent]"
        return rep

class OnDiskPrivateKey(PrivateKey):
    """
    Some on-disk private key that needs opening and possibly decrypting.

:param str source:
        String tracking where this key's path was specified; should be one of
        ``"ssh-config"``, ``"python-config"``, or ``"implicit-home"``.
    :param Path path:
        The filesystem path this key was loaded from.
    :param PKey pkey:
        The `PKey` object this auth source uses/represents.
    """

def __init__(self, username, source, path, pkey):
        super().__init__(username=username)
        self.source = source
        allowed = ("ssh-config", "python-config", "implicit-home")
        if source not in allowed:
            raise ValueError(f"source argument must be one of: {allowed!r}")
        self.path = path
        # Superclass wants .pkey, other two are mostly for display/debugging.
        self.pkey = pkey

def __repr__(self):
        return self._repr(
            key=self.pkey, source=self.source, path=str(self.path)
        )

# TODO re sources: is there anything in an OpenSSH config file that doesn't fit
# into what Paramiko already had kwargs for?

SourceResult = namedtuple("SourceResult", ["source", "result"])

# TODO: tempting to make this an OrderedDict, except the keys essentially want
# to be rich objects (AuthSources) which do not make for useful user indexing?
# TODO: members being vanilla tuples is pretty old-school/expedient; they
# "really" want to be something that's type friendlier (unless the tuple's 2nd
# member being a Union of two types is "fine"?), which I assume means yet more
# classes, eg an abstract SourceResult with concrete AuthSuccess and
# AuthFailure children?
# TODO: arguably we want __init__ typechecking of the members (or to leverage
# mypy by classifying this literally as list-of-AuthSource?)
class AuthResult(list):
    """
    Represents a partial or complete SSH authentication attempt.

This class conceptually extends `AuthStrategy` by pairing the former's
    authentication **sources** with the **results** of trying to authenticate
    with them.

`AuthResult` is a (subclass of) `list` of `namedtuple`, which are of the
    form ``namedtuple('SourceResult', 'source', 'result')`` (where the
    ``source`` member is an `AuthSource` and the ``result`` member is either a
    return value from the relevant `.Transport` method, or an exception
    object).

.. note::
        Transport auth method results are always themselves a ``list`` of "next
        allowable authentication methods".

In the simple case of "you just authenticated successfully", it's an
        empty list; if your auth was rejected but you're allowed to try again,
        it will be a list of string method names like ``pubkey`` or
        ``password``.

The ``__str__`` of this class represents the empty-list scenario as the
        word ``success``, which should make reading the result of an
        authentication session more obvious to humans.

Instances also have a `strategy` attribute referencing the `AuthStrategy`
    which was attempted.
    """

def __init__(self, strategy, *args, **kwargs):
        self.strategy = strategy
        super().__init__(*args, **kwargs)

def __str__(self):
        # NOTE: meaningfully distinct from __repr__, which still wants to use
        # superclass' implementation.
        # TODO: go hog wild, use rich.Table? how is that on degraded term's?
        # TODO: test this lol
        return "\n".join(
            f"{x.source} -> {x.result or 'success'}" for x in self
        )

# TODO 4.0: descend from SSHException or even just Exception
class AuthFailure(AuthenticationException):
    """
    Basic exception wrapping an `AuthResult` indicating overall auth failure.

Note that `AuthFailure` descends from `AuthenticationException` but is
    generally "higher level"; the latter is now only raised by individual
    `AuthSource` attempts and should typically only be seen by users when
    encapsulated in this class. It subclasses `AuthenticationException`
    primarily for backwards compatibility reasons.
    """

def __init__(self, result):
        self.result = result

def __str__(self):
        return "\n" + str(self.result)

class AuthStrategy:
    """
    This class represents one or more attempts to auth with an SSH server.

By default, subclasses must at least accept an ``ssh_config``
    (`.SSHConfig`) keyword argument, but may opt to accept more as needed for
    their particular strategy.
    """

def __init__(
        self,
        ssh_config,
    ):
        self.ssh_config = ssh_config
        self.log = get_logger(__name__)

def get_sources(self):
        """
        Generator yielding `AuthSource` instances, in the order to try.

This is the primary override point for subclasses: you figure out what
        sources you need, and ``yield`` them.

Subclasses _of_ subclasses may find themselves wanting to do things
        like filtering or discarding around a call to `super`.
        """
        raise NotImplementedError

def authenticate(self, transport):
        """
        Handles attempting `AuthSource` instances yielded from `get_sources`.

You *normally* won't need to override this, but it's an option for
        advanced users.
        """
        succeeded = False
        overall_result = AuthResult(strategy=self)
        # TODO: arguably we could fit in a "send none auth, record allowed auth
        # types sent back" thing here as OpenSSH-client does, but that likely
        # wants to live in fabric.OpenSSHAuthStrategy as not all target servers
        # will implement it!
        # TODO: needs better "server told us too many attempts" checking!
        for source in self.get_sources():
            self.log.debug(f"Trying {source}")
            try:  # NOTE: this really wants to _only_ wrap the authenticate()!
                result = source.authenticate(transport)
                succeeded = True
            # TODO: 'except PartialAuthentication' is needed for 2FA and
            # similar, as per old SSHClient.connect - it is the only way
            # AuthHandler supplies access to the 'name-list' field from
            # MSG_USERAUTH_FAILURE, at present.
            except Exception as e:
                result = e
                # TODO: look at what this could possibly raise, we don't really
                # want Exception here, right? just SSHException subclasses? or
                # do we truly want to capture anything at all with assumption
                # it's easy enough for users to look afterwards?
                # NOTE: showing type, not message, for tersity & also most of
                # the time it's basically just "Authentication failed."
                source_class = e.__class__.__name__
                self.log.info(
                    f"Authentication via {source} failed with {source_class}"
                )
            overall_result.append(SourceResult(source, result))
            if succeeded:
                break
        # Gotta die here if nothing worked, otherwise Transport's main loop
        # just kinda hangs out until something times out!
        if not succeeded:
            raise AuthFailure(result=overall_result)
        # Success: give back what was done, in case they care.
        return overall_result

# TODO: is there anything OpenSSH client does which _can't_ cleanly map to
    # iterating a generator?

cmd: