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/config.py (27362B)

# Copyright (C) 2006-2007  Robey Pointer <robeypointer@gmail.com>
# Copyright (C) 2012  Olle Lundberg <geek@nerd.sh>
#
# This file is part of paramiko.
#
# Paramiko is free software; you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2.1 of the License, or (at your option)
# any later version.
#
# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.

"""
Configuration file (aka ``ssh_config``) support.
"""

import fnmatch
import getpass
import os
import re
import shlex
import socket
from hashlib import sha1
from io import StringIO
from functools import partial

invoke, invoke_import_error = None, None
try:
    import invoke
except ImportError as e:
    invoke_import_error = e

from .ssh_exception import CouldNotCanonicalize, ConfigParseError

SSH_PORT = 22

class SSHConfig:
    """
    Representation of config information as stored in the format used by
    OpenSSH. Queries can be made via `lookup`. The format is described in
    OpenSSH's ``ssh_config`` man page. This class is provided primarily as a
    convenience to posix users (since the OpenSSH format is a de-facto
    standard on posix) but should work fine on Windows too.

.. versionadded:: 1.6
    """

SETTINGS_REGEX = re.compile(r"(\w+)(?:\s*=\s*|\s+)(.+)")

# TODO: do a full scan of ssh.c & friends to make sure we're fully
    # compatible across the board, e.g. OpenSSH 8.1 added %n to ProxyCommand.
    TOKENS_BY_CONFIG_KEY = {
        "controlpath": ["%C", "%h", "%l", "%L", "%n", "%p", "%r", "%u"],
        "hostname": ["%h"],
        "identityfile": ["%C", "~", "%d", "%h", "%l", "%u", "%r"],
        "proxycommand": ["~", "%h", "%p", "%r"],
        "proxyjump": ["%h", "%p", "%r"],
        # Doesn't seem worth making this 'special' for now, it will fit well
        # enough (no actual match-exec config key to be confused with).
        "match-exec": ["%C", "%d", "%h", "%L", "%l", "%n", "%p", "%r", "%u"],
    }

def __init__(self):
        """
        Create a new OpenSSH config object.

Note: the newer alternate constructors `from_path`, `from_file` and
        `from_text` are simpler to use, as they parse on instantiation. For
        example, instead of::

config = SSHConfig()
            config.parse(open("some-path.config")

you could::

config = SSHConfig.from_file(open("some-path.config"))
            # Or more directly:
            config = SSHConfig.from_path("some-path.config")
            # Or if you have arbitrary ssh_config text from some other source:
            config = SSHConfig.from_text("Host foo\\n\\tUser bar")
        """
        self._config = []

@classmethod
    def from_text(cls, text):
        """
        Create a new, parsed `SSHConfig` from ``text`` string.

.. versionadded:: 2.7
        """
        return cls.from_file(StringIO(text))

@classmethod
    def from_path(cls, path):
        """
        Create a new, parsed `SSHConfig` from the file found at ``path``.

.. versionadded:: 2.7
        """
        with open(path) as flo:
            return cls.from_file(flo)

@classmethod
    def from_file(cls, flo):
        """
        Create a new, parsed `SSHConfig` from file-like object ``flo``.

.. versionadded:: 2.7
        """
        obj = cls()
        obj.parse(flo)
        return obj

def parse(self, file_obj):
        """
        Read an OpenSSH config from the given file object.

:param file_obj: a file-like object to read the config file from
        """
        # Start out w/ implicit/anonymous global host-like block to hold
        # anything not contained by an explicit one.
        context = {"host": ["*"], "config": {}}
        for line in file_obj:
            # Strip any leading or trailing whitespace from the line.
            # Refer to https://github.com/paramiko/paramiko/issues/499
            line = line.strip()
            # Skip blanks, comments
            if not line or line.startswith("#"):
                continue

# Parse line into key, value
            match = re.match(self.SETTINGS_REGEX, line)
            if not match:
                raise ConfigParseError("Unparsable line {}".format(line))
            key = match.group(1).lower()
            value = match.group(2)

# Host keyword triggers switch to new block/context
            if key in ("host", "match"):
                self._config.append(context)
                context = {"config": {}}
                if key == "host":
                    # TODO 4.0: make these real objects or at least name this
                    # "hosts" to acknowledge it's an iterable. (Doing so prior
                    # to 3.0, despite it being a private API, feels bad -
                    # surely such an old codebase has folks actually relying on
                    # these keys.)
                    context["host"] = self._get_hosts(value)
                else:
                    context["matches"] = self._get_matches(value)
            # Special-case for noop ProxyCommands
            elif key == "proxycommand" and value.lower() == "none":
                # Store 'none' as None - not as a string implying that the
                # proxycommand is the literal shell command "none"!
                context["config"][key] = None
            # All other keywords get stored, directly or via append
            else:
                if value.startswith('"') and value.endswith('"'):
                    value = value[1:-1]

# identityfile, localforward, remoteforward keys are special
                # cases, since they are allowed to be specified multiple times
                # and they should be tried in order of specification.
                if key in ["identityfile", "localforward", "remoteforward"]:
                    if key in context["config"]:
                        context["config"][key].append(value)
                    else:
                        context["config"][key] = [value]
                elif key not in context["config"]:
                    context["config"][key] = value
        # Store last 'open' block and we're done
        self._config.append(context)

def lookup(self, hostname):
        """
        Return a dict (`SSHConfigDict`) of config options for a given hostname.

The host-matching rules of OpenSSH's ``ssh_config`` man page are used:
        For each parameter, the first obtained value will be used.  The
        configuration files contain sections separated by ``Host`` and/or
        ``Match`` specifications, and that section is only applied for hosts
        which match the given patterns or keywords

Since the first obtained value for each parameter is used, more host-
        specific declarations should be given near the beginning of the file,
        and general defaults at the end.

The keys in the returned dict are all normalized to lowercase (look for
        ``"port"``, not ``"Port"``. The values are processed according to the
        rules for substitution variable expansion in ``ssh_config``.

Finally, please see the docs for `SSHConfigDict` for deeper info on
        features such as optional type conversion methods, e.g.::

conf = my_config.lookup('myhost')
            assert conf['passwordauthentication'] == 'yes'
            assert conf.as_bool('passwordauthentication') is True

.. note::
            If there is no explicitly configured ``HostName`` value, it will be
            set to the being-looked-up hostname, which is as close as we can
            get to OpenSSH's behavior around that particular option.

:param str hostname: the hostname to lookup

.. versionchanged:: 2.5
            Returns `SSHConfigDict` objects instead of dict literals.
        .. versionchanged:: 2.7
            Added canonicalization support.
        .. versionchanged:: 2.7
            Added ``Match`` support.
        .. versionchanged:: 3.3
            Added ``Match final`` support.
        """
        # First pass
        options = self._lookup(hostname=hostname)
        # Inject HostName if it was not set (this used to be done incidentally
        # during tokenization, for some reason).
        if "hostname" not in options:
            options["hostname"] = hostname
        # Handle canonicalization
        canon = options.get("canonicalizehostname", None) in ("yes", "always")
        maxdots = int(options.get("canonicalizemaxdots", 1))
        if canon and hostname.count(".") <= maxdots:
            # NOTE: OpenSSH manpage does not explicitly state this, but its
            # implementation for CanonicalDomains is 'split on any whitespace'.
            domains = options["canonicaldomains"].split()
            hostname = self.canonicalize(hostname, options, domains)
            # Overwrite HostName again here (this is also what OpenSSH does)
            options["hostname"] = hostname
            options = self._lookup(
                hostname, options, canonical=True, final=True
            )
        else:
            options = self._lookup(
                hostname, options, canonical=False, final=True
            )
        return options

def _lookup(self, hostname, options=None, canonical=False, final=False):
        # Init
        if options is None:
            options = SSHConfigDict()
        # Iterate all stanzas, applying any that match, in turn (so that things
        # like Match can reference currently understood state)
        for context in self._config:
            if not (
                self._pattern_matches(context.get("host", []), hostname)
                or self._does_match(
                    context.get("matches", []),
                    hostname,
                    canonical,
                    final,
                    options,
                )
            ):
                continue
            for key, value in context["config"].items():
                if key not in options:
                    # Create a copy of the original value,
                    # else it will reference the original list
                    # in self._config and update that value too
                    # when the extend() is being called.
                    options[key] = value[:] if value is not None else value
                elif key == "identityfile":
                    options[key].extend(
                        x for x in value if x not in options[key]
                    )
        if final:
            # Expand variables in resulting values
            # (besides 'Match exec' which was already handled above)
            options = self._expand_variables(options, hostname)
        return options

def canonicalize(self, hostname, options, domains):
        """
        Return canonicalized version of ``hostname``.

:param str hostname: Target hostname.
        :param options: An `SSHConfigDict` from a previous lookup pass.
        :param domains: List of domains (e.g. ``["paramiko.org"]``).

:returns: A canonicalized hostname if one was found, else ``None``.

.. versionadded:: 2.7
        """
        found = False
        for domain in domains:
            candidate = "{}.{}".format(hostname, domain)
            family_specific = _addressfamily_host_lookup(candidate, options)
            if family_specific is not None:
                # TODO: would we want to dig deeper into other results? e.g. to
                # find something that satisfies PermittedCNAMEs when that is
                # implemented?
                found = family_specific[0]
            else:
                # TODO: what does ssh use here and is there a reason to use
                # that instead of gethostbyname?
                try:
                    found = socket.gethostbyname(candidate)
                except socket.gaierror:
                    pass
            if found:
                # TODO: follow CNAME (implied by found != candidate?) if
                # CanonicalizePermittedCNAMEs allows it
                return candidate
        # If we got here, it means canonicalization failed.
        # When CanonicalizeFallbackLocal is undefined or 'yes', we just spit
        # back the original hostname.
        if options.get("canonicalizefallbacklocal", "yes") == "yes":
            return hostname
        # And here, we failed AND fallback was set to a non-yes value, so we
        # need to get mad.
        raise CouldNotCanonicalize(hostname)

def get_hostnames(self):
        """
        Return the set of literal hostnames defined in the SSH config (both
        explicit hostnames and wildcard entries).
        """
        hosts = set()
        for entry in self._config:
            hosts.update(entry["host"])
        return hosts

def _pattern_matches(self, patterns, target):
        # Convenience auto-splitter if not already a list
        if hasattr(patterns, "split"):
            patterns = patterns.split(",")
        match = False
        for pattern in patterns:
            # Short-circuit if target matches a negated pattern
            if pattern.startswith("!") and fnmatch.fnmatch(
                target, pattern[1:]
            ):
                return False
            # Flag a match, but continue (in case of later negation) if regular
            # match occurs
            elif fnmatch.fnmatch(target, pattern):
                match = True
        return match

def _does_match(
        self, match_list, target_hostname, canonical, final, options
    ):
        matched = []
        candidates = match_list[:]
        local_username = getpass.getuser()
        while candidates:
            candidate = candidates.pop(0)
            passed = None
            # Obtain latest host/user value every loop, so later Match may
            # reference values assigned within a prior Match.
            configured_host = options.get("hostname", None)
            configured_user = options.get("user", None)
            type_, param = candidate["type"], candidate["param"]
            # Canonical is a hard pass/fail based on whether this is a
            # canonicalized re-lookup.
            if type_ == "canonical":
                if self._should_fail(canonical, candidate):
                    return False
            if type_ == "final":
                passed = final
            # The parse step ensures we only see this by itself or after
            # canonical, so it's also an easy hard pass. (No negation here as
            # that would be uh, pretty weird?)
            elif type_ == "all":
                return True
            # From here, we are testing various non-hard criteria,
            # short-circuiting only on fail
            elif type_ == "host":
                hostval = configured_host or target_hostname
                passed = self._pattern_matches(param, hostval)
            elif type_ == "originalhost":
                passed = self._pattern_matches(param, target_hostname)
            elif type_ == "user":
                user = configured_user or local_username
                passed = self._pattern_matches(param, user)
            elif type_ == "localuser":
                passed = self._pattern_matches(param, local_username)
            elif type_ == "exec":
                exec_cmd = self._tokenize(
                    options, target_hostname, "match-exec", param
                )
                # This is the laziest spot in which we can get mad about an
                # inability to import Invoke.
                if invoke is None:
                    raise invoke_import_error
                # Like OpenSSH, we 'redirect' stdout but let stderr bubble up
                passed = invoke.run(exec_cmd, hide="stdout", warn=True).ok
            # Tackle any 'passed, but was negated' results from above
            if passed is not None and self._should_fail(passed, candidate):
                return False
            # Made it all the way here? Everything matched!
            matched.append(candidate)
        # Did anything match? (To be treated as bool, usually.)
        return matched

def _should_fail(self, would_pass, candidate):
        return would_pass if candidate["negate"] else not would_pass

def _tokenize(self, config, target_hostname, key, value):
        """
        Tokenize a string based on current config/hostname data.

:param config: Current config data.
        :param target_hostname: Original target connection hostname.
        :param key: Config key being tokenized (used to filter token list).
        :param value: Config value being tokenized.

:returns: The tokenized version of the input ``value`` string.
        """
        allowed_tokens = self._allowed_tokens(key)
        # Short-circuit if no tokenization possible
        if not allowed_tokens:
            return value
        # Obtain potentially configured hostname, for use with %h.
        # Special-case where we are tokenizing the hostname itself, to avoid
        # replacing %h with a %h-bearing value, etc.
        configured_hostname = target_hostname
        if key != "hostname":
            configured_hostname = config.get("hostname", configured_hostname)
        # Ditto the rest of the source values
        if "port" in config:
            port = config["port"]
        else:
            port = SSH_PORT
        user = getpass.getuser()
        if "user" in config:
            remoteuser = config["user"]
        else:
            remoteuser = user
        local_hostname = socket.gethostname().split(".")[0]
        local_fqdn = LazyFqdn(config, local_hostname)
        homedir = os.path.expanduser("~")
        tohash = local_hostname + target_hostname + repr(port) + remoteuser
        # The actual tokens!
        replacements = {
            # TODO: %%???
            "%C": sha1(tohash.encode()).hexdigest(),
            "%d": homedir,
            "%h": configured_hostname,
            # TODO: %i?
            "%L": local_hostname,
            "%l": local_fqdn,
            # also this is pseudo buggy when not in Match exec mode so document
            # that. also WHY is that the case?? don't we do all of this late?
            "%n": target_hostname,
            "%p": port,
            "%r": remoteuser,
            # TODO: %T? don't believe this is possible however
            "%u": user,
            "~": homedir,
        }
        # Do the thing with the stuff
        tokenized = value
        for find, replace in replacements.items():
            if find not in allowed_tokens:
                continue
            tokenized = tokenized.replace(find, str(replace))
        # TODO: log? eg that value -> tokenized
        return tokenized

def _allowed_tokens(self, key):
        """
        Given config ``key``, return list of token strings to tokenize.

.. note::
            This feels like it wants to eventually go away, but is used to
            preserve as-strict-as-possible compatibility with OpenSSH, which
            for whatever reason only applies some tokens to some config keys.
        """
        return self.TOKENS_BY_CONFIG_KEY.get(key, [])

def _expand_variables(self, config, target_hostname):
        """
        Return a dict of config options with expanded substitutions
        for a given original & current target hostname.

Please refer to :doc:`/api/config` for details.

:param dict config: the currently parsed config
        :param str hostname: the hostname whose config is being looked up
        """
        for k in config:
            if config[k] is None:
                continue
            tokenizer = partial(self._tokenize, config, target_hostname, k)
            if isinstance(config[k], list):
                for i, value in enumerate(config[k]):
                    config[k][i] = tokenizer(value)
            else:
                config[k] = tokenizer(config[k])
        return config

def _get_hosts(self, host):
        """
        Return a list of host_names from host value.
        """
        try:
            return shlex.split(host)
        except ValueError:
            raise ConfigParseError("Unparsable host {}".format(host))

def _get_matches(self, match):
        """
        Parse a specific Match config line into a list-of-dicts for its values.

Performs some parse-time validation as well.
        """
        matches = []
        tokens = shlex.split(match)
        while tokens:
            match = {"type": None, "param": None, "negate": False}
            type_ = tokens.pop(0)
            # Handle per-keyword negation
            if type_.startswith("!"):
                match["negate"] = True
                type_ = type_[1:]
            match["type"] = type_
            # all/canonical have no params (everything else does)
            if type_ in ("all", "canonical", "final"):
                matches.append(match)
                continue
            if not tokens:
                raise ConfigParseError(
                    "Missing parameter to Match '{}' keyword".format(type_)
                )
            match["param"] = tokens.pop(0)
            matches.append(match)
        # Perform some (easier to do now than in the middle) validation that is
        # better handled here than at lookup time.
        keywords = [x["type"] for x in matches]
        if "all" in keywords:
            allowable = ("all", "canonical")
            ok, bad = (
                list(filter(lambda x: x in allowable, keywords)),
                list(filter(lambda x: x not in allowable, keywords)),
            )
            err = None
            if any(bad):
                err = "Match does not allow 'all' mixed with anything but 'canonical'"  # noqa
            elif "canonical" in ok and ok.index("canonical") > ok.index("all"):
                err = "Match does not allow 'all' before 'canonical'"
            if err is not None:
                raise ConfigParseError(err)
        return matches

def _addressfamily_host_lookup(hostname, options):
    """
    Try looking up ``hostname`` in an IPv4 or IPv6 specific manner.

This is an odd duck due to needing use in two divergent use cases. It looks
    up ``AddressFamily`` in ``options`` and if it is ``inet`` or ``inet6``,
    this function uses `socket.getaddrinfo` to perform a family-specific
    lookup, returning the result if successful.

In any other situation -- lookup failure, or ``AddressFamily`` being
    unspecified or ``any`` -- ``None`` is returned instead and the caller is
    expected to do something situation-appropriate like calling
    `socket.gethostbyname`.

:param str hostname: Hostname to look up.
    :param options: `SSHConfigDict` instance w/ parsed options.
    :returns: ``getaddrinfo``-style tuples, or ``None``, depending.
    """
    address_family = options.get("addressfamily", "any").lower()
    if address_family == "any":
        return
    try:
        family = socket.AF_INET6
        if address_family == "inet":
            family = socket.AF_INET
        return socket.getaddrinfo(
            hostname,
            None,
            family,
            socket.SOCK_DGRAM,
            socket.IPPROTO_IP,
            socket.AI_CANONNAME,
        )
    except socket.gaierror:
        pass

class LazyFqdn:
    """
    Returns the host's fqdn on request as string.
    """

def __init__(self, config, host=None):
        self.fqdn = None
        self.config = config
        self.host = host

def __str__(self):
        if self.fqdn is None:
            #
            # If the SSH config contains AddressFamily, use that when
            # determining  the local host's FQDN. Using socket.getfqdn() from
            # the standard library is the most general solution, but can
            # result in noticeable delays on some platforms when IPv6 is
            # misconfigured or not available, as it calls getaddrinfo with no
            # address family specified, so both IPv4 and IPv6 are checked.
            #

# Handle specific option
            fqdn = None
            results = _addressfamily_host_lookup(self.host, self.config)
            if results is not None:
                for res in results:
                    af, socktype, proto, canonname, sa = res
                    if canonname and "." in canonname:
                        fqdn = canonname
                        break
            # Handle 'any' / unspecified / lookup failure
            if fqdn is None:
                fqdn = socket.getfqdn()
            # Cache
            self.fqdn = fqdn
        return self.fqdn

class SSHConfigDict(dict):
    """
    A dictionary wrapper/subclass for per-host configuration structures.

This class introduces some usage niceties for consumers of `SSHConfig`,
    specifically around the issue of variable type conversions: normal value
    access yields strings, but there are now methods such as `as_bool` and
    `as_int` that yield casted values instead.

For example, given the following ``ssh_config`` file snippet::

Host foo.example.com
            PasswordAuthentication no
            Compression yes
            ServerAliveInterval 60

the following code highlights how you can access the raw strings as well as
    usefully Python type-casted versions (recalling that keys are all
    normalized to lowercase first)::

my_config = SSHConfig()
        my_config.parse(open('~/.ssh/config'))
        conf = my_config.lookup('foo.example.com')

assert conf['passwordauthentication'] == 'no'
        assert conf.as_bool('passwordauthentication') is False
        assert conf['compression'] == 'yes'
        assert conf.as_bool('compression') is True
        assert conf['serveraliveinterval'] == '60'
        assert conf.as_int('serveraliveinterval') == 60

.. versionadded:: 2.5
    """

def as_bool(self, key):
        """
        Express given key's value as a boolean type.

Typically, this is used for ``ssh_config``'s pseudo-boolean values
        which are either ``"yes"`` or ``"no"``. In such cases, ``"yes"`` yields
        ``True`` and any other value becomes ``False``.

.. note::
            If (for whatever reason) the stored value is already boolean in
            nature, it's simply returned.

.. versionadded:: 2.5
        """
        val = self[key]
        if isinstance(val, bool):
            return val
        return val.lower() == "yes"

def as_int(self, key):
        """
        Express given key's value as an integer, if possible.

This method will raise ``ValueError`` or similar if the value is not
        int-appropriate, same as the builtin `int` type.

.. versionadded:: 2.5
        """
        return int(self[key])

cmd: