summaryrefslogtreecommitdiffstats
path: root/tests/test_config.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_config.py')
-rw-r--r--tests/test_config.py1048
1 files changed, 1048 insertions, 0 deletions
diff --git a/tests/test_config.py b/tests/test_config.py
new file mode 100644
index 0000000..2e49aa3
--- /dev/null
+++ b/tests/test_config.py
@@ -0,0 +1,1048 @@
+# This file is part of Paramiko and subject to the license in /LICENSE in this
+# repository
+
+from os.path import expanduser
+from socket import gaierror
+
+try:
+ from invoke import Result
+except ImportError:
+ Result = None
+
+from unittest.mock import patch
+from pytest import raises, mark, fixture
+
+from paramiko import (
+ SSHConfig,
+ SSHConfigDict,
+ CouldNotCanonicalize,
+ ConfigParseError,
+)
+
+from ._util import _config
+
+
+@fixture
+def socket():
+ """
+ Patch all of socket.* in our config module to prevent eg real DNS lookups.
+
+ Also forces getaddrinfo (used in our addressfamily lookup stuff) to always
+ fail by default to mimic usual lack of AddressFamily related crap.
+
+ Callers who want to mock DNS lookups can then safely assume gethostbyname()
+ will be in use.
+ """
+ with patch("paramiko.config.socket") as mocket:
+ # Reinstate gaierror as an actual exception and not a sub-mock.
+ # (Presumably this would work with any exception, but why not use the
+ # real one?)
+ mocket.gaierror = gaierror
+ # Patch out getaddrinfo, used to detect family-specific IP lookup -
+ # only useful for a few specific tests.
+ mocket.getaddrinfo.side_effect = mocket.gaierror
+ # Patch out getfqdn to return some real string for when it gets called;
+ # some code (eg tokenization) gets mad w/ MagicMocks
+ mocket.getfqdn.return_value = "some.fake.fqdn"
+ mocket.gethostname.return_value = "local.fake.fqdn"
+ yield mocket
+
+
+def load_config(name):
+ return SSHConfig.from_path(_config(name))
+
+
+class TestSSHConfig:
+ def setup(self):
+ self.config = load_config("robey")
+
+ def test_init(self):
+ # No args!
+ with raises(TypeError):
+ SSHConfig("uh oh!")
+ # No args.
+ assert not SSHConfig()._config
+
+ def test_from_text(self):
+ config = SSHConfig.from_text("User foo")
+ assert config.lookup("foo.example.com")["user"] == "foo"
+
+ def test_from_file(self):
+ with open(_config("robey")) as flo:
+ config = SSHConfig.from_file(flo)
+ assert config.lookup("whatever")["user"] == "robey"
+
+ def test_from_path(self):
+ # NOTE: DO NOT replace with use of load_config() :D
+ config = SSHConfig.from_path(_config("robey"))
+ assert config.lookup("meh.example.com")["port"] == "3333"
+
+ def test_parse_config(self):
+ expected = [
+ {"host": ["*"], "config": {}},
+ {
+ "host": ["*"],
+ "config": {"identityfile": ["~/.ssh/id_rsa"], "user": "robey"},
+ },
+ {
+ "host": ["*.example.com"],
+ "config": {"user": "bjork", "port": "3333"},
+ },
+ {"host": ["*"], "config": {"crazy": "something dumb"}},
+ {
+ "host": ["spoo.example.com"],
+ "config": {"crazy": "something else"},
+ },
+ ]
+ assert self.config._config == expected
+
+ @mark.parametrize(
+ "host,values",
+ (
+ (
+ "irc.danger.com",
+ {
+ "crazy": "something dumb",
+ "hostname": "irc.danger.com",
+ "user": "robey",
+ },
+ ),
+ (
+ "irc.example.com",
+ {
+ "crazy": "something dumb",
+ "hostname": "irc.example.com",
+ "user": "robey",
+ "port": "3333",
+ },
+ ),
+ (
+ "spoo.example.com",
+ {
+ "crazy": "something dumb",
+ "hostname": "spoo.example.com",
+ "user": "robey",
+ "port": "3333",
+ },
+ ),
+ ),
+ )
+ def test_host_config(self, host, values):
+ expected = dict(
+ values, hostname=host, identityfile=[expanduser("~/.ssh/id_rsa")]
+ )
+ assert self.config.lookup(host) == expected
+
+ def test_fabric_issue_33(self):
+ config = SSHConfig.from_text(
+ """
+Host www13.*
+ Port 22
+
+Host *.example.com
+ Port 2222
+
+Host *
+ Port 3333
+"""
+ )
+ host = "www13.example.com"
+ expected = {"hostname": host, "port": "22"}
+ assert config.lookup(host) == expected
+
+ def test_proxycommand_config_equals_parsing(self):
+ """
+ ProxyCommand should not split on equals signs within the value.
+ """
+ config = SSHConfig.from_text(
+ """
+Host space-delimited
+ ProxyCommand foo bar=biz baz
+
+Host equals-delimited
+ ProxyCommand=foo bar=biz baz
+"""
+ )
+ for host in ("space-delimited", "equals-delimited"):
+ value = config.lookup(host)["proxycommand"]
+ assert value == "foo bar=biz baz"
+
+ def test_proxycommand_interpolation(self):
+ """
+ ProxyCommand should perform interpolation on the value
+ """
+ config = SSHConfig.from_text(
+ """
+Host specific
+ Port 37
+ ProxyCommand host %h port %p lol
+
+Host portonly
+ Port 155
+
+Host *
+ Port 25
+ ProxyCommand host %h port %p
+"""
+ )
+ for host, val in (
+ ("foo.com", "host foo.com port 25"),
+ ("specific", "host specific port 37 lol"),
+ ("portonly", "host portonly port 155"),
+ ):
+ assert config.lookup(host)["proxycommand"] == val
+
+ def test_proxycommand_tilde_expansion(self):
+ """
+ Tilde (~) should be expanded inside ProxyCommand
+ """
+ config = SSHConfig.from_text(
+ """
+Host test
+ ProxyCommand ssh -F ~/.ssh/test_config bastion nc %h %p
+"""
+ )
+ expected = "ssh -F {}/.ssh/test_config bastion nc test 22".format(
+ expanduser("~")
+ )
+ got = config.lookup("test")["proxycommand"]
+ assert got == expected
+
+ @patch("paramiko.config.getpass")
+ def test_proxyjump_token_expansion(self, getpass):
+ getpass.getuser.return_value = "gandalf"
+ config = SSHConfig.from_text(
+ """
+Host justhost
+ ProxyJump jumpuser@%h
+Host userhost
+ ProxyJump %r@%h:222
+Host allcustom
+ ProxyJump %r@%h:%p
+"""
+ )
+ assert config.lookup("justhost")["proxyjump"] == "jumpuser@justhost"
+ assert config.lookup("userhost")["proxyjump"] == "gandalf@userhost:222"
+ assert (
+ config.lookup("allcustom")["proxyjump"] == "gandalf@allcustom:22"
+ )
+
+ @patch("paramiko.config.getpass")
+ def test_controlpath_token_expansion(self, getpass, socket):
+ getpass.getuser.return_value = "gandalf"
+ config = SSHConfig.from_text(
+ """
+Host explicit_user
+ User root
+ ControlPath user %u remoteuser %r
+
+Host explicit_host
+ HostName ohai
+ ControlPath remoteuser %r host %h orighost %n
+
+Host hashbrowns
+ ControlPath %C
+ """
+ )
+ result = config.lookup("explicit_user")["controlpath"]
+ # Remote user is User val, local user is User val
+ assert result == "user gandalf remoteuser root"
+ result = config.lookup("explicit_host")["controlpath"]
+ # Remote user falls back to local user; host and orighost may differ
+ assert result == "remoteuser gandalf host ohai orighost explicit_host"
+ # Supports %C
+ result = config.lookup("hashbrowns")["controlpath"]
+ assert result == "a438e7dbf5308b923aba9db8fe2ca63447ac8688"
+
+ def test_negation(self):
+ config = SSHConfig.from_text(
+ """
+Host www13.* !*.example.com
+ Port 22
+
+Host *.example.com !www13.*
+ Port 2222
+
+Host www13.*
+ Port 8080
+
+Host *
+ Port 3333
+"""
+ )
+ host = "www13.example.com"
+ expected = {"hostname": host, "port": "8080"}
+ assert config.lookup(host) == expected
+
+ def test_proxycommand(self):
+ config = SSHConfig.from_text(
+ """
+Host proxy-with-equal-divisor-and-space
+ProxyCommand = foo=bar
+
+Host proxy-with-equal-divisor-and-no-space
+ProxyCommand=foo=bar
+
+Host proxy-without-equal-divisor
+ProxyCommand foo=bar:%h-%p
+"""
+ )
+ for host, values in {
+ "proxy-with-equal-divisor-and-space": {
+ "hostname": "proxy-with-equal-divisor-and-space",
+ "proxycommand": "foo=bar",
+ },
+ "proxy-with-equal-divisor-and-no-space": {
+ "hostname": "proxy-with-equal-divisor-and-no-space",
+ "proxycommand": "foo=bar",
+ },
+ "proxy-without-equal-divisor": {
+ "hostname": "proxy-without-equal-divisor",
+ "proxycommand": "foo=bar:proxy-without-equal-divisor-22",
+ },
+ }.items():
+
+ assert config.lookup(host) == values
+
+ @patch("paramiko.config.getpass")
+ def test_identityfile(self, getpass, socket):
+ getpass.getuser.return_value = "gandalf"
+ config = SSHConfig.from_text(
+ """
+IdentityFile id_dsa0
+
+Host *
+IdentityFile id_dsa1
+
+Host dsa2
+IdentityFile id_dsa2
+
+Host dsa2*
+IdentityFile id_dsa22
+
+Host hashbrowns
+IdentityFile %C
+"""
+ )
+ for host, values in {
+ "foo": {"hostname": "foo", "identityfile": ["id_dsa0", "id_dsa1"]},
+ "dsa2": {
+ "hostname": "dsa2",
+ "identityfile": ["id_dsa0", "id_dsa1", "id_dsa2", "id_dsa22"],
+ },
+ "dsa22": {
+ "hostname": "dsa22",
+ "identityfile": ["id_dsa0", "id_dsa1", "id_dsa22"],
+ },
+ "hashbrowns": {
+ "hostname": "hashbrowns",
+ "identityfile": [
+ "id_dsa0",
+ "id_dsa1",
+ "a438e7dbf5308b923aba9db8fe2ca63447ac8688",
+ ],
+ },
+ }.items():
+ assert config.lookup(host) == values
+
+ def test_config_addressfamily_and_lazy_fqdn(self):
+ """
+ Ensure the code path honoring non-'all' AddressFamily doesn't asplode
+ """
+ config = SSHConfig.from_text(
+ """
+AddressFamily inet
+IdentityFile something_%l_using_fqdn
+"""
+ )
+ assert config.lookup(
+ "meh"
+ ) # will die during lookup() if bug regresses
+
+ def test_config_dos_crlf_succeeds(self):
+ config = SSHConfig.from_text(
+ """
+Host abcqwerty\r\nHostName 127.0.0.1\r\n
+"""
+ )
+ assert config.lookup("abcqwerty")["hostname"] == "127.0.0.1"
+
+ def test_get_hostnames(self):
+ expected = {"*", "*.example.com", "spoo.example.com"}
+ assert self.config.get_hostnames() == expected
+
+ def test_quoted_host_names(self):
+ config = SSHConfig.from_text(
+ """
+Host "param pam" param "pam"
+ Port 1111
+
+Host "param2"
+ Port 2222
+
+Host param3 parara
+ Port 3333
+
+Host param4 "p a r" "p" "par" para
+ Port 4444
+"""
+ )
+ res = {
+ "param pam": {"hostname": "param pam", "port": "1111"},
+ "param": {"hostname": "param", "port": "1111"},
+ "pam": {"hostname": "pam", "port": "1111"},
+ "param2": {"hostname": "param2", "port": "2222"},
+ "param3": {"hostname": "param3", "port": "3333"},
+ "parara": {"hostname": "parara", "port": "3333"},
+ "param4": {"hostname": "param4", "port": "4444"},
+ "p a r": {"hostname": "p a r", "port": "4444"},
+ "p": {"hostname": "p", "port": "4444"},
+ "par": {"hostname": "par", "port": "4444"},
+ "para": {"hostname": "para", "port": "4444"},
+ }
+ for host, values in res.items():
+ assert config.lookup(host) == values
+
+ def test_quoted_params_in_config(self):
+ config = SSHConfig.from_text(
+ """
+Host "param pam" param "pam"
+ IdentityFile id_rsa
+
+Host "param2"
+ IdentityFile "test rsa key"
+
+Host param3 parara
+ IdentityFile id_rsa
+ IdentityFile "test rsa key"
+"""
+ )
+ res = {
+ "param pam": {"hostname": "param pam", "identityfile": ["id_rsa"]},
+ "param": {"hostname": "param", "identityfile": ["id_rsa"]},
+ "pam": {"hostname": "pam", "identityfile": ["id_rsa"]},
+ "param2": {"hostname": "param2", "identityfile": ["test rsa key"]},
+ "param3": {
+ "hostname": "param3",
+ "identityfile": ["id_rsa", "test rsa key"],
+ },
+ "parara": {
+ "hostname": "parara",
+ "identityfile": ["id_rsa", "test rsa key"],
+ },
+ }
+ for host, values in res.items():
+ assert config.lookup(host) == values
+
+ def test_quoted_host_in_config(self):
+ conf = SSHConfig()
+ correct_data = {
+ "param": ["param"],
+ '"param"': ["param"],
+ "param pam": ["param", "pam"],
+ '"param" "pam"': ["param", "pam"],
+ '"param" pam': ["param", "pam"],
+ 'param "pam"': ["param", "pam"],
+ 'param "pam" p': ["param", "pam", "p"],
+ '"param" pam "p"': ["param", "pam", "p"],
+ '"pa ram"': ["pa ram"],
+ '"pa ram" pam': ["pa ram", "pam"],
+ 'param "p a m"': ["param", "p a m"],
+ }
+ incorrect_data = ['param"', '"param', 'param "pam', 'param "pam" "p a']
+ for host, values in correct_data.items():
+ assert conf._get_hosts(host) == values
+ for host in incorrect_data:
+ with raises(ConfigParseError):
+ conf._get_hosts(host)
+
+ def test_invalid_line_format_excepts(self):
+ with raises(ConfigParseError):
+ load_config("invalid")
+
+ def test_proxycommand_none_issue_415(self):
+ config = SSHConfig.from_text(
+ """
+Host proxycommand-standard-none
+ ProxyCommand None
+
+Host proxycommand-with-equals-none
+ ProxyCommand=None
+"""
+ )
+ for host, values in {
+ "proxycommand-standard-none": {
+ "hostname": "proxycommand-standard-none",
+ "proxycommand": None,
+ },
+ "proxycommand-with-equals-none": {
+ "hostname": "proxycommand-with-equals-none",
+ "proxycommand": None,
+ },
+ }.items():
+
+ assert config.lookup(host) == values
+
+ def test_proxycommand_none_masking(self):
+ # Re: https://github.com/paramiko/paramiko/issues/670
+ config = SSHConfig.from_text(
+ """
+Host specific-host
+ ProxyCommand none
+
+Host other-host
+ ProxyCommand other-proxy
+
+Host *
+ ProxyCommand default-proxy
+"""
+ )
+ # In versions <3.0, 'None' ProxyCommands got deleted, and this itself
+ # caused bugs. In 3.0, we more cleanly map "none" to None. This test
+ # has been altered accordingly but left around to ensure no
+ # regressions.
+ assert config.lookup("specific-host")["proxycommand"] is None
+ assert config.lookup("other-host")["proxycommand"] == "other-proxy"
+ cmd = config.lookup("some-random-host")["proxycommand"]
+ assert cmd == "default-proxy"
+
+ def test_hostname_tokenization(self):
+ result = load_config("hostname-tokenized").lookup("whatever")
+ assert result["hostname"] == "prefix.whatever"
+
+
+class TestSSHConfigDict:
+ def test_SSHConfigDict_construct_empty(self):
+ assert not SSHConfigDict()
+
+ def test_SSHConfigDict_construct_from_list(self):
+ assert SSHConfigDict([(1, 2)])[1] == 2
+
+ def test_SSHConfigDict_construct_from_dict(self):
+ assert SSHConfigDict({1: 2})[1] == 2
+
+ @mark.parametrize("true_ish", ("yes", "YES", "Yes", True))
+ def test_SSHConfigDict_as_bool_true_ish(self, true_ish):
+ assert SSHConfigDict({"key": true_ish}).as_bool("key") is True
+
+ @mark.parametrize("false_ish", ("no", "NO", "No", False))
+ def test_SSHConfigDict_as_bool(self, false_ish):
+ assert SSHConfigDict({"key": false_ish}).as_bool("key") is False
+
+ @mark.parametrize("int_val", ("42", 42))
+ def test_SSHConfigDict_as_int(self, int_val):
+ assert SSHConfigDict({"key": int_val}).as_int("key") == 42
+
+ @mark.parametrize("non_int", ("not an int", None, object()))
+ def test_SSHConfigDict_as_int_failures(self, non_int):
+ conf = SSHConfigDict({"key": non_int})
+
+ try:
+ int(non_int)
+ except Exception as e:
+ exception_type = type(e)
+
+ with raises(exception_type):
+ conf.as_int("key")
+
+ def test_SSHConfig_host_dicts_are_SSHConfigDict_instances(self):
+ config = SSHConfig.from_text(
+ """
+Host *.example.com
+ Port 2222
+
+Host *
+ Port 3333
+"""
+ )
+ assert config.lookup("foo.example.com").as_int("port") == 2222
+
+ def test_SSHConfig_wildcard_host_dicts_are_SSHConfigDict_instances(self):
+ config = SSHConfig.from_text(
+ """
+Host *.example.com
+ Port 2222
+
+Host *
+ Port 3333
+"""
+ )
+ assert config.lookup("anything-else").as_int("port") == 3333
+
+
+class TestHostnameCanonicalization:
+ # NOTE: this class uses on-disk configs, and ones with real (at time of
+ # writing) DNS names, so that one can easily test OpenSSH's behavior using
+ # "ssh -F path/to/file.config -G <target>".
+
+ def test_off_by_default(self, socket):
+ result = load_config("basic").lookup("www")
+ assert result["hostname"] == "www"
+ assert "user" not in result
+ assert not socket.gethostbyname.called
+
+ def test_explicit_no_same_as_default(self, socket):
+ result = load_config("no-canon").lookup("www")
+ assert result["hostname"] == "www"
+ assert "user" not in result
+ assert not socket.gethostbyname.called
+
+ @mark.parametrize(
+ "config_name",
+ ("canon", "canon-always", "canon-local", "canon-local-always"),
+ )
+ def test_canonicalization_base_cases(self, socket, config_name):
+ result = load_config(config_name).lookup("www")
+ assert result["hostname"] == "www.paramiko.org"
+ assert result["user"] == "rando"
+ socket.gethostbyname.assert_called_once_with("www.paramiko.org")
+
+ def test_uses_getaddrinfo_when_AddressFamily_given(self, socket):
+ # Undo default 'always fails' mock
+ socket.getaddrinfo.side_effect = None
+ socket.getaddrinfo.return_value = [True] # just need 1st value truthy
+ result = load_config("canon-ipv4").lookup("www")
+ assert result["hostname"] == "www.paramiko.org"
+ assert result["user"] == "rando"
+ assert not socket.gethostbyname.called
+ gai_args = socket.getaddrinfo.call_args[0]
+ assert gai_args[0] == "www.paramiko.org"
+ assert gai_args[2] is socket.AF_INET # Mocked, but, still useful
+
+ @mark.skip
+ def test_empty_CanonicalDomains_canonicalizes_despite_noop(self, socket):
+ # Confirmed this is how OpenSSH behaves as well. Bit silly, but.
+ # TODO: this requires modifying SETTINGS_REGEX, which is a mite scary
+ # (honestly I'd prefer to move to a real parser lib anyhow) and since
+ # this is a very dumb corner case, it's marked skip for now.
+ result = load_config("empty-canon").lookup("www")
+ assert result["hostname"] == "www" # no paramiko.org
+ assert "user" not in result # did not discover canonicalized block
+
+ def test_CanonicalDomains_may_be_set_to_space_separated_list(self, socket):
+ # Test config has a bogus domain, followed by paramiko.org
+ socket.gethostbyname.side_effect = [socket.gaierror, True]
+ result = load_config("multi-canon-domains").lookup("www")
+ assert result["hostname"] == "www.paramiko.org"
+ assert result["user"] == "rando"
+ assert [x[0][0] for x in socket.gethostbyname.call_args_list] == [
+ "www.not-a-real-tld",
+ "www.paramiko.org",
+ ]
+
+ def test_canonicalization_applies_to_single_dot_by_default(self, socket):
+ result = load_config("deep-canon").lookup("sub.www")
+ assert result["hostname"] == "sub.www.paramiko.org"
+ assert result["user"] == "deep"
+
+ def test_canonicalization_not_applied_to_two_dots_by_default(self, socket):
+ result = load_config("deep-canon").lookup("subber.sub.www")
+ assert result["hostname"] == "subber.sub.www"
+ assert "user" not in result
+
+ def test_hostname_depth_controllable_with_max_dots_directive(self, socket):
+ # This config sets MaxDots of 2, so now canonicalization occurs
+ result = load_config("deep-canon-maxdots").lookup("subber.sub.www")
+ assert result["hostname"] == "subber.sub.www.paramiko.org"
+ assert result["user"] == "deeper"
+
+ def test_max_dots_may_be_zero(self, socket):
+ result = load_config("zero-maxdots").lookup("sub.www")
+ assert result["hostname"] == "sub.www"
+ assert "user" not in result
+
+ def test_fallback_yes_does_not_canonicalize_or_error(self, socket):
+ socket.gethostbyname.side_effect = socket.gaierror
+ result = load_config("fallback-yes").lookup("www")
+ assert result["hostname"] == "www"
+ assert "user" not in result
+
+ def test_fallback_no_causes_errors_for_unresolvable_names(self, socket):
+ socket.gethostbyname.side_effect = socket.gaierror
+ with raises(CouldNotCanonicalize) as info:
+ load_config("fallback-no").lookup("doesnotexist")
+ assert str(info.value) == "doesnotexist"
+
+ def test_identityfile_continues_being_appended_to(self, socket):
+ result = load_config("canon").lookup("www")
+ assert result["identityfile"] == ["base.key", "canonicalized.key"]
+
+
+@mark.skip
+class TestCanonicalizationOfCNAMEs:
+ def test_permitted_cnames_may_be_one_to_one_mapping(self):
+ # CanonicalizePermittedCNAMEs *.foo.com:*.bar.com
+ pass
+
+ def test_permitted_cnames_may_be_one_to_many_mapping(self):
+ # CanonicalizePermittedCNAMEs *.foo.com:*.bar.com,*.biz.com
+ pass
+
+ def test_permitted_cnames_may_be_many_to_one_mapping(self):
+ # CanonicalizePermittedCNAMEs *.foo.com,*.bar.com:*.biz.com
+ pass
+
+ def test_permitted_cnames_may_be_many_to_many_mapping(self):
+ # CanonicalizePermittedCNAMEs *.foo.com,*.bar.com:*.biz.com,*.baz.com
+ pass
+
+ def test_permitted_cnames_may_be_multiple_mappings(self):
+ # CanonicalizePermittedCNAMEs *.foo.com,*.bar.com *.biz.com:*.baz.com
+ pass
+
+ def test_permitted_cnames_may_be_multiple_complex_mappings(self):
+ # Same as prev but with multiple patterns on both ends in both args
+ pass
+
+
+class TestMatchAll:
+ def test_always_matches(self):
+ result = load_config("match-all").lookup("general")
+ assert result["user"] == "awesome"
+
+ def test_may_not_mix_with_non_canonical_keywords(self):
+ for config in ("match-all-and-more", "match-all-and-more-before"):
+ with raises(ConfigParseError):
+ load_config(config).lookup("whatever")
+
+ def test_may_come_after_canonical(self, socket):
+ result = load_config("match-all-after-canonical").lookup("www")
+ assert result["user"] == "awesome"
+
+ def test_may_not_come_before_canonical(self, socket):
+ with raises(ConfigParseError):
+ load_config("match-all-before-canonical")
+
+ def test_after_canonical_not_loaded_when_non_canonicalized(self, socket):
+ result = load_config("match-canonical-no").lookup("a-host")
+ assert "user" not in result
+
+
+def _expect(success_on):
+ """
+ Returns a side_effect-friendly Invoke success result for given command(s).
+
+ Ensures that any other commands fail; this is useful for testing 'Match
+ exec' because it means all other such clauses under test act like no-ops.
+
+ :param success_on:
+ Single string or list of strings, noting commands that should appear to
+ succeed.
+ """
+ if isinstance(success_on, str):
+ success_on = [success_on]
+
+ def inner(command, *args, **kwargs):
+ # Sanity checking - we always expect that invoke.run is called with
+ # these.
+ assert kwargs.get("hide", None) == "stdout"
+ assert kwargs.get("warn", None) is True
+ # Fake exit
+ exit = 0 if command in success_on else 1
+ return Result(exited=exit)
+
+ return inner
+
+
+@mark.skipif(Result is None, reason="requires invoke package")
+class TestMatchExec:
+ @patch("paramiko.config.invoke", new=None)
+ @patch("paramiko.config.invoke_import_error", new=ImportError("meh"))
+ def test_raises_invoke_ImportErrors_at_runtime(self):
+ # Not an ideal test, but I don't know of a non-bad way to fake out
+ # module-time ImportErrors. So we mock the symptoms. Meh!
+ with raises(ImportError) as info:
+ load_config("match-exec").lookup("oh-noes")
+ assert str(info.value) == "meh"
+
+ @patch("paramiko.config.invoke.run")
+ @mark.parametrize(
+ "cmd,user",
+ [
+ ("unquoted", "rando"),
+ ("quoted", "benjamin"),
+ ("quoted spaced", "neil"),
+ ],
+ )
+ def test_accepts_single_possibly_quoted_argument(self, run, cmd, user):
+ run.side_effect = _expect(cmd)
+ result = load_config("match-exec").lookup("whatever")
+ assert result["user"] == user
+
+ @patch("paramiko.config.invoke.run")
+ def test_does_not_match_nonzero_exit_codes(self, run):
+ # Nothing will succeed -> no User ever gets loaded
+ run.return_value = Result(exited=1)
+ result = load_config("match-exec").lookup("whatever")
+ assert "user" not in result
+
+ @patch("paramiko.config.getpass")
+ @patch("paramiko.config.invoke.run")
+ def test_tokenizes_argument(self, run, getpass, socket):
+ getpass.getuser.return_value = "gandalf"
+ # Actual exec value is "%C %d %h %L %l %n %p %r %u"
+ parts = (
+ "bf5ba06778434a9384ee4217e462f64888bd0cd2",
+ expanduser("~"),
+ "configured",
+ "local",
+ "some.fake.fqdn",
+ "target",
+ "22",
+ "intermediate",
+ "gandalf",
+ )
+ run.side_effect = _expect(" ".join(parts))
+ result = load_config("match-exec").lookup("target")
+ assert result["port"] == "1337"
+
+ @patch("paramiko.config.invoke.run")
+ def test_works_with_canonical(self, run, socket):
+ # Ensure both stanzas' exec components appear to match
+ run.side_effect = _expect(["uncanonicalized", "canonicalized"])
+ result = load_config("match-exec-canonical").lookup("who-cares")
+ # Prove both config values got loaded up, across the two passes
+ assert result["user"] == "defenseless"
+ assert result["port"] == "8007"
+
+ @patch("paramiko.config.invoke.run")
+ def test_may_be_negated(self, run):
+ run.side_effect = _expect("this succeeds")
+ result = load_config("match-exec-negation").lookup("so-confusing")
+ # If negation did not work, the first of the two Match exec directives
+ # would have set User to 'nope' (and/or the second would have NOT set
+ # User to 'yup')
+ assert result["user"] == "yup"
+
+ def test_requires_an_argument(self):
+ with raises(ConfigParseError):
+ load_config("match-exec-no-arg")
+
+ @patch("paramiko.config.invoke.run")
+ def test_works_with_tokenized_hostname(self, run):
+ run.side_effect = _expect("ping target")
+ result = load_config("hostname-exec-tokenized").lookup("target")
+ assert result["hostname"] == "pingable.target"
+
+
+class TestMatchHost:
+ def test_matches_target_name_when_no_hostname(self):
+ result = load_config("match-host").lookup("target")
+ assert result["user"] == "rand"
+
+ def test_matches_hostname_from_global_setting(self):
+ # Also works for ones set in regular Host stanzas
+ result = load_config("match-host-name").lookup("anything")
+ assert result["user"] == "silly"
+
+ def test_matches_hostname_from_earlier_match(self):
+ # Corner case: one Match matches original host, sets HostName,
+ # subsequent Match matches the latter.
+ result = load_config("match-host-from-match").lookup("original-host")
+ assert result["user"] == "inner"
+
+ def test_may_be_globbed(self):
+ result = load_config("match-host-glob-list").lookup("whatever")
+ assert result["user"] == "matrim"
+
+ def test_may_be_comma_separated_list(self):
+ for target in ("somehost", "someotherhost"):
+ result = load_config("match-host-glob-list").lookup(target)
+ assert result["user"] == "thom"
+
+ def test_comma_separated_list_may_have_internal_negation(self):
+ conf = load_config("match-host-glob-list")
+ assert conf.lookup("good")["user"] == "perrin"
+ assert "user" not in conf.lookup("goof")
+
+ def test_matches_canonicalized_name(self, socket):
+ # Without 'canonical' explicitly declared, mind.
+ result = load_config("match-host-canonicalized").lookup("www")
+ assert result["user"] == "rand"
+
+ def test_works_with_canonical_keyword(self, socket):
+ # NOTE: distinct from 'happens to be canonicalized' above
+ result = load_config("match-host-canonicalized").lookup("docs")
+ assert result["user"] == "eric"
+
+ def test_may_be_negated(self):
+ conf = load_config("match-host-negated")
+ assert conf.lookup("docs")["user"] == "jeff"
+ assert "user" not in conf.lookup("www")
+
+ def test_requires_an_argument(self):
+ with raises(ConfigParseError):
+ load_config("match-host-no-arg")
+
+
+class TestMatchOriginalHost:
+ def test_matches_target_host_not_hostname(self):
+ result = load_config("match-orighost").lookup("target")
+ assert result["hostname"] == "bogus"
+ assert result["user"] == "tuon"
+
+ def test_matches_target_host_not_canonicalized_name(self, socket):
+ result = load_config("match-orighost-canonical").lookup("www")
+ assert result["hostname"] == "www.paramiko.org"
+ assert result["user"] == "tuon"
+
+ def test_may_be_globbed(self):
+ result = load_config("match-orighost").lookup("whatever")
+ assert result["user"] == "matrim"
+
+ def test_may_be_comma_separated_list(self):
+ for target in ("comma", "separated"):
+ result = load_config("match-orighost").lookup(target)
+ assert result["user"] == "chameleon"
+
+ def test_comma_separated_list_may_have_internal_negation(self):
+ result = load_config("match-orighost").lookup("nope")
+ assert "user" not in result
+
+ def test_may_be_negated(self):
+ result = load_config("match-orighost").lookup("docs")
+ assert result["user"] == "thom"
+
+ def test_requires_an_argument(self):
+ with raises(ConfigParseError):
+ load_config("match-orighost-no-arg")
+
+
+class TestMatchUser:
+ def test_matches_configured_username(self):
+ result = load_config("match-user-explicit").lookup("anything")
+ assert result["hostname"] == "dumb"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_matches_local_username_by_default(self, getuser):
+ getuser.return_value = "gandalf"
+ result = load_config("match-user").lookup("anything")
+ assert result["hostname"] == "gondor"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_may_be_globbed(self, getuser):
+ for user in ("bilbo", "bombadil"):
+ getuser.return_value = user
+ result = load_config("match-user").lookup("anything")
+ assert result["hostname"] == "shire"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_may_be_comma_separated_list(self, getuser):
+ for user in ("aragorn", "frodo"):
+ getuser.return_value = user
+ result = load_config("match-user").lookup("anything")
+ assert result["hostname"] == "moria"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_comma_separated_list_may_have_internal_negation(self, getuser):
+ getuser.return_value = "legolas"
+ result = load_config("match-user").lookup("anything")
+ assert "port" not in result
+ getuser.return_value = "gimli"
+ result = load_config("match-user").lookup("anything")
+ assert result["port"] == "7373"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_may_be_negated(self, getuser):
+ getuser.return_value = "saruman"
+ result = load_config("match-user").lookup("anything")
+ assert result["hostname"] == "mordor"
+
+ def test_requires_an_argument(self):
+ with raises(ConfigParseError):
+ load_config("match-user-no-arg")
+
+
+# NOTE: highly derivative of previous suite due to the former's use of
+# localuser fallback. Doesn't seem worth conflating/refactoring right now.
+class TestMatchLocalUser:
+ @patch("paramiko.config.getpass.getuser")
+ def test_matches_local_username(self, getuser):
+ getuser.return_value = "gandalf"
+ result = load_config("match-localuser").lookup("anything")
+ assert result["hostname"] == "gondor"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_may_be_globbed(self, getuser):
+ for user in ("bilbo", "bombadil"):
+ getuser.return_value = user
+ result = load_config("match-localuser").lookup("anything")
+ assert result["hostname"] == "shire"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_may_be_comma_separated_list(self, getuser):
+ for user in ("aragorn", "frodo"):
+ getuser.return_value = user
+ result = load_config("match-localuser").lookup("anything")
+ assert result["hostname"] == "moria"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_comma_separated_list_may_have_internal_negation(self, getuser):
+ getuser.return_value = "legolas"
+ result = load_config("match-localuser").lookup("anything")
+ assert "port" not in result
+ getuser.return_value = "gimli"
+ result = load_config("match-localuser").lookup("anything")
+ assert result["port"] == "7373"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_may_be_negated(self, getuser):
+ getuser.return_value = "saruman"
+ result = load_config("match-localuser").lookup("anything")
+ assert result["hostname"] == "mordor"
+
+ def test_requires_an_argument(self):
+ with raises(ConfigParseError):
+ load_config("match-localuser-no-arg")
+
+
+class TestComplexMatching:
+ # NOTE: this is still a cherry-pick of a few levels of complexity, there's
+ # no point testing literally all possible combinations.
+
+ def test_originalhost_host(self):
+ result = load_config("match-complex").lookup("target")
+ assert result["hostname"] == "bogus"
+ assert result["user"] == "rand"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_originalhost_localuser(self, getuser):
+ getuser.return_value = "rando"
+ result = load_config("match-complex").lookup("remote")
+ assert result["user"] == "calrissian"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_everything_but_all(self, getuser):
+ getuser.return_value = "rando"
+ result = load_config("match-complex").lookup("www")
+ assert result["port"] == "7777"
+
+ @patch("paramiko.config.getpass.getuser")
+ def test_everything_but_all_with_some_negated(self, getuser):
+ getuser.return_value = "rando"
+ result = load_config("match-complex").lookup("docs")
+ assert result["port"] == "1234"
+
+ def test_negated_canonical(self, socket):
+ # !canonical in a config that is not canonicalized - does match
+ result = load_config("match-canonical-no").lookup("specific")
+ assert result["user"] == "overload"
+ # !canonical in a config that is canonicalized - does NOT match
+ result = load_config("match-canonical-yes").lookup("www")
+ assert result["user"] == "hidden"
+
+
+class TestFinalMatching(object):
+ def test_finally(self):
+ result = load_config("match-final").lookup("finally")
+ assert result["proxyjump"] == "jump"
+ assert result["port"] == "1001"
+
+ def test_default_port(self):
+ result = load_config("match-final").lookup("default-port")
+ assert result["proxyjump"] == "jump"
+ assert result["port"] == "1002"
+
+ def test_negated(self):
+ result = load_config("match-final").lookup("jump")
+ assert result["port"] == "1003"