697 lines
27 KiB
Python
697 lines
27 KiB
Python
|
# 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])
|