path: root/tests/test_config.py

# 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

from 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
        yield mocket


def load_config(name):
    return SSHConfig.from_path(_config(name))


class TestSSHConfig(object):
    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

    def test_host_config(self):
        for host, values in {
            "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",
            },
        }.items():
            values = dict(
                values,
                hostname=host,
                identityfile=[expanduser("~/.ssh/id_rsa")],
            )
            assert self.config.lookup(host) == values

    def test_host_config_expose_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_controlpath_token_expansion(self, getpass):
        getpass.getuser.return_value = "gandalf"
        config = SSHConfig.from_text("""
Host explicit
    User root
    ControlPath user %u remoteuser %r

Host fallback
    ControlPath user %u remoteuser %r
        """)
        result = config.lookup("explicit")["controlpath"]
        assert result == "user gandalf remoteuser root"
        result = config.lookup("fallback")["controlpath"]
        assert result == "user gandalf remoteuser gandalf"

    def test_host_config_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_host_config_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

    def test_host_config_test_identityfile(self):
        config = SSHConfig.from_text(
            """

IdentityFile id_dsa0

Host *
IdentityFile id_dsa1

Host dsa2
IdentityFile id_dsa2

Host dsa2*
IdentityFile id_dsa22
"""
        )
        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"],
            },
        }.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_418(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-with-equals-none": {
                "hostname": "proxycommand-with-equals-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
"""
        )
        # When bug is present, the full stripping-out of specific-host's
        # ProxyCommand means it actually appears to pick up the default
        # ProxyCommand value instead, due to cascading. It should (for
        # backwards compatibility reasons in 1.x/2.x) appear completely blank,
        # as if the host had no ProxyCommand whatsoever.
        # Threw another unrelated host in there just for sanity reasons.
        assert "proxycommand" not in config.lookup("specific-host")
        assert config.lookup("other-host")["proxycommand"] == "other-proxy"
        cmd = config.lookup("some-random-host")["proxycommand"]
        assert cmd == "default-proxy"


class TestSSHConfigDict(object):
    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(object):
    # 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(object):
    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(object):
    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):
        with raises(ConfigParseError):
            load_config("match-all-and-more")

    def test_may_come_after_canonical(self):
        result = load_config("match-all-after-canonical").lookup("anything")
        assert result["user"] == "awesome"

    def test_may_not_come_before_canonical(self):
        with raises(ConfigParseError):
            load_config("match-all-before-canonical")

    def test_after_canonical_not_loaded_when_non_canonicalized(self):
        result = load_config("match-canonical-no").lookup("a-host")
        assert "user" not in result