# Copyright (C) 2006-2007 Robey Pointer <[email protected]>
# Copyright (C) 2012 Olle Lundberg <[email protected]>
#
# 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])