From bcd04e88601035e586a106409e453502538199a2 Mon Sep 17 00:00:00 2001
From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com>
Date: Tue, 22 Sep 2026 16:19:45 +0000
Subject: [PATCH 1/3] fix: confine keys to rootdir; verify host keys by default

- Keys that are absolute or whose `..` segments leave rootdir now raise
  KeyError in getitem/setitem/delitem/mkdir (and `in` returns False).
  Keyword-only `allow_escape=True` restores the old behaviour.
- Host keys are checked against known_hosts (system, user, and the ssh
  config's UserKnownHostsFile). Unknown keys are refused unless the host's
  ssh config sets StrictHostKeyChecking no/accept-new, or the caller passes
  keyword-only `missing_host_key_policy` (e.g. paramiko.AutoAddPolicy).

Closes #7

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 sshdol/base.py                   | 118 +++++++++++++++++++++++++++++--
 sshdol/tests/test_confinement.py |  71 +++++++++++++++++++
 2 files changed, 182 insertions(+), 7 deletions(-)
 create mode 100644 sshdol/tests/test_confinement.py

diff --git a/sshdol/base.py b/sshdol/base.py
index c20d491..cb2d2a2 100644
--- a/sshdol/base.py
+++ b/sshdol/base.py
@@ -29,6 +29,7 @@
 import sys
 import paramiko
 from functools import lru_cache
+import posixpath
 import shlex
 
 # Make a DFLT_RECYCLE_BIN
@@ -86,6 +87,75 @@ def normalize_path(path: str) -> str:
     return path.replace("\\", "/")
 
 
+def escapes_root(path: str) -> bool:
+    """Whether a (normalized, ``/``-separated) key path points outside the root.
+
+    >>> escapes_root('a/b.txt'), escapes_root('a/../b'), escapes_root('')
+    (False, False, False)
+    >>> escapes_root('..'), escapes_root('a/../../b'), escapes_root('/etc/x')
+    (True, True, True)
+    """
+    if not path:
+        return False
+    if path.startswith("/"):
+        return True
+    normalized = posixpath.normpath(path)
+    return normalized == ".." or normalized.startswith("../")
+
+
+class RejectUnknownHostKey(paramiko.RejectPolicy):
+    """Refuse hosts whose key is not in known_hosts, with a message saying what to do."""
+
+    def missing_host_key(self, client, hostname, key):
+        raise paramiko.SSHException(
+            f"The host key of {hostname!r} is not in known_hosts, so the connection "
+            "was refused. Verify and record it by connecting once with `ssh`, set "
+            "StrictHostKeyChecking in your ssh config for this host, or pass "
+            "missing_host_key_policy=paramiko.AutoAddPolicy to accept unknown keys."
+        )
+
+
+_SSH_CONFIG_AUTO_ADD_VALUES = {"no", "off", "accept-new"}
+_NULL_KNOWN_HOSTS_FILES = {"/dev/null", "none"}
+
+
+def resolve_host_key_policy(missing_host_key_policy=None, ssh_config=None):
+    """The paramiko policy for hosts whose key is not in ``known_hosts``.
+
+    An explicit ``missing_host_key_policy`` (a paramiko policy class or instance)
+    wins. Otherwise the host's ssh config decides: ``StrictHostKeyChecking`` set
+    to ``no``/``off``/``accept-new`` accepts (and remembers for the session) an
+    unknown key; anything else rejects it, as a non-interactive ``ssh`` would.
+    A key that *changed* is always refused by paramiko.
+
+    >>> type(resolve_host_key_policy()).__name__
+    'RejectUnknownHostKey'
+    >>> type(resolve_host_key_policy(ssh_config={'stricthostkeychecking': 'accept-new'})).__name__
+    'AutoAddPolicy'
+    >>> type(resolve_host_key_policy(paramiko.AutoAddPolicy)).__name__
+    'AutoAddPolicy'
+    """
+    if missing_host_key_policy is not None:
+        if isinstance(missing_host_key_policy, type):
+            missing_host_key_policy = missing_host_key_policy()
+        return missing_host_key_policy
+    strict = str((ssh_config or {}).get("stricthostkeychecking", "")).lower()
+    if strict in _SSH_CONFIG_AUTO_ADD_VALUES:
+        return paramiko.AutoAddPolicy()
+    return RejectUnknownHostKey()
+
+
+def _known_hosts_files(ssh_config=None):
+    files = [
+        "/etc/ssh/ssh_known_hosts",
+        os.path.expanduser("~/.ssh/known_hosts"),
+    ]
+    for f in (ssh_config or {}).get("userknownhostsfile", "").split():
+        if f.lower() not in _NULL_KNOWN_HOSTS_FILES:
+            files.append(os.path.expanduser(f))
+    return [f for f in files if os.path.isfile(f)]
+
+
 def split_path(path: str) -> tuple[str, str]:
     """
     Split a path into directory and file parts.
@@ -145,6 +215,8 @@ def __init__(
         max_levels=0,
         create_dirs=False,  # Only relevant for writable stores
         strict_contains=False,  # Whether to raise KeyError or return False for deep paths in __contains__
+        allow_escape=False,  # Whether keys may point outside rootdir (absolute or ``..``)
+        missing_host_key_policy=None,  # paramiko policy for unknown host keys (see below)
     ):
         """
         Initialize an SSH connection with read-only file access.
@@ -171,6 +243,16 @@ def __init__(
             strict_contains: If True, __contains__ will raise KeyError for paths beyond max_levels
                             If False (default), it will return False for such paths
                             Design notes: https://github.com/i2mint/sshdol/issues/1#issuecomment-2714508482
+            allow_escape: If False (default), keys are confined to ``rootdir``:
+                       absolute keys and keys whose ``..`` segments leave it raise
+                       ``KeyError`` (and ``in`` returns False). Symbolic links on
+                       the server are not resolved by this check.
+            missing_host_key_policy: What to do when the server's host key is not
+                       in ``known_hosts`` (system, user, or the config's
+                       ``UserKnownHostsFile``). Default: follow the host's ssh
+                       config ``StrictHostKeyChecking`` (``no``/``accept-new`` accept
+                       the key), else refuse to connect. Pass e.g.
+                       ``paramiko.AutoAddPolicy`` to accept unknown keys.
         """
         # Store initialization parameters
         self._init_params = {
@@ -187,6 +269,8 @@ def __init__(
             "max_levels": max_levels,
             "create_dirs": create_dirs,
             "strict_contains": strict_contains,
+            "allow_escape": allow_escape,
+            "missing_host_key_policy": missing_host_key_policy,
         }
 
         # Store configuration options
@@ -197,18 +281,26 @@ def __init__(
         self._include_hidden = include_hidden
         self._include_directories = include_directories  # Store the new parameter
         self._dir_access = dir_access  # Store the new parameter
+        self._allow_escape = allow_escape
 
         assert self._encoding is None or isinstance(self._encoding, str), (
             "Encoding must be a string"
         )
 
         # Initialize the SSH connection
+        ssh_config = get_ssh_config_for_host(host) if host else {}
         self._ssh = paramiko.SSHClient()
-        self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        for known_hosts_file in _known_hosts_files(ssh_config):
+            try:
+                self._ssh.load_system_host_keys(known_hosts_file)
+            except (OSError, paramiko.SSHException):
+                pass  # an unreadable known_hosts file just contributes no keys
+        self._ssh.set_missing_host_key_policy(
+            resolve_host_key_policy(missing_host_key_policy, ssh_config)
+        )
 
         # If a host alias is provided, try to get config from SSH config file
         if host and not all([user, url]):
-            ssh_config = get_ssh_config_for_host(host)
 
             # Use values from config if not explicitly provided
             user = user or ssh_config.get("user")
@@ -253,6 +345,15 @@ def __init__(
                 # If directory doesn't exist, don't error - it will be handled by operations
                 pass
 
+    def _key_to_path(self, k):
+        """Normalize key ``k`` to a path, refusing (``KeyError``) one that leaves rootdir."""
+        path = normalize_path(k)
+        if not getattr(self, "_allow_escape", False) and escapes_root(path):
+            raise KeyError(
+                f"Key points outside rootdir (pass allow_escape=True to allow): {k}"
+            )
+        return path
+
     def _is_dir(self, path):
         """Check if a path is a directory"""
         try:
@@ -383,7 +484,7 @@ def __getitem__(self, k):
         Supports path-based keys with slashes for nested files.
         Respects max_levels constraint for reading.
         """
-        path = normalize_path(k)
+        path = self._key_to_path(k)
 
         # Check if path exceeds allowed depth
         self._check_path_depth(path)
@@ -532,7 +633,10 @@ def __contains__(self, k):
 
         See https://github.com/i2mint/sshdol/issues/1#issuecomment-2714508482
         """
-        path = normalize_path(k)
+        try:
+            path = self._key_to_path(k)
+        except KeyError:
+            return False
 
         # Check if path exceeds allowed depth
         if self._max_levels is not None and path.count("/") > self._max_levels:
@@ -627,7 +731,7 @@ def __setitem__(self, k, v):
         Respects max_levels constraint for writing.
         """
         # Check if path exceeds allowed depth
-        path = normalize_path(k)
+        path = self._key_to_path(k)
         self._check_path_depth(path)
 
         # Handle encoding based on the _encoding attribute
@@ -661,7 +765,7 @@ def __delitem__(self, k):
         """
         Delete a file on the SSH server.
         """
-        path = normalize_path(k)
+        path = self._key_to_path(k)
 
         if not self._path_exists(path):
             raise KeyError(k)
@@ -695,7 +799,7 @@ def mkdir(self, path, exist_ok=False):
         Raises:
             KeyError: If directory cannot be created
         """
-        path = normalize_path(path)
+        path = self._key_to_path(path)
 
         # Check if directory already exists
         if self._path_exists(path):
diff --git a/sshdol/tests/test_confinement.py b/sshdol/tests/test_confinement.py
new file mode 100644
index 0000000..bad1f91
--- /dev/null
+++ b/sshdol/tests/test_confinement.py
@@ -0,0 +1,71 @@
+"""Key confinement and host-key policy (no SSH server needed)."""
+
+import paramiko
+import pytest
+
+from sshdol.base import (
+    RejectUnknownHostKey,
+    SshFiles,
+    escapes_root,
+    resolve_host_key_policy,
+)
+
+
+def _store(**attrs):
+    # Bypass __init__ (which connects); only key handling is exercised.
+    s = SshFiles.__new__(SshFiles)
+    s._allow_escape = False
+    s.__dict__.update(attrs)
+    return s
+
+
+@pytest.mark.parametrize(
+    "key", ["..", "../x", "a/../../x", "/etc/hosts", "..\\x", "a/b/../../../x"]
+)
+def test_escaping_keys_are_refused(key):
+    s = _store()
+    for op in (
+        lambda: s[key],
+        lambda: s.__setitem__(key, b"x"),
+        lambda: s.__delitem__(key),
+        lambda: s.mkdir(key),
+    ):
+        with pytest.raises(KeyError, match="outside rootdir"):
+            op()
+    assert key not in s
+
+
+@pytest.mark.parametrize("key", ["a", "a/b.txt", "a/../b", "./a", "a/"])
+def test_keys_inside_root_pass(key):
+    assert not escapes_root(_store()._key_to_path(key))
+
+
+def test_allow_escape_opt_in():
+    assert _store(_allow_escape=True)._key_to_path("../x") == "../x"
+
+
+def test_host_key_policy_defaults_to_reject():
+    assert isinstance(resolve_host_key_policy(), RejectUnknownHostKey)
+    assert isinstance(
+        resolve_host_key_policy(ssh_config={"stricthostkeychecking": "yes"}),
+        RejectUnknownHostKey,
+    )
+    with pytest.raises(paramiko.SSHException, match="not in known_hosts"):
+        RejectUnknownHostKey().missing_host_key(None, "example.invalid", None)
+
+
+@pytest.mark.parametrize("value", ["no", "accept-new", "OFF"])
+def test_host_key_policy_follows_ssh_config(value):
+    policy = resolve_host_key_policy(ssh_config={"stricthostkeychecking": value})
+    assert isinstance(policy, paramiko.AutoAddPolicy)
+
+
+def test_explicit_host_key_policy_wins():
+    assert isinstance(
+        resolve_host_key_policy(
+            paramiko.AutoAddPolicy, {"stricthostkeychecking": "yes"}
+        ),
+        paramiko.AutoAddPolicy,
+    )
+    policy = paramiko.WarningPolicy()
+    assert resolve_host_key_policy(policy) is policy

From f2ab02b15978a01d2ebe6c79f8b55528c1543c0a Mon Sep 17 00:00:00 2001
From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com>
Date: Tue, 22 Sep 2026 16:29:25 +0000
Subject: [PATCH 2/3] review: lenient known_hosts parsing, OpenSSH file
 selection, key pinning

- known_hosts is parsed line by line; marker (@cert-authority/@revoked),
  unsupported, malformed or undecodable lines are skipped instead of
  aborting (or crashing) the connection.
- GlobalKnownHostsFile/UserKnownHostsFile replace the defaults, as in
  OpenSSH (/dev/null or none = no file).
- Keys recorded under HostKeyAlias or the lower-cased host name are matched.
- Subdirectory instances (new connections) are pinned to the host key their
  parent connected with.
- Keys containing NUL are refused; clearer refusal message.
- Integration tests fail instead of skipping when SSH_TEST_HOST is set
  explicitly (as in CI) and the connection fails.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 sshdol/base.py                   | 141 +++++++++++++++++++++++++++----
 sshdol/tests/test_base.py        |   5 ++
 sshdol/tests/test_confinement.py |  78 ++++++++++++++++-
 3 files changed, 204 insertions(+), 20 deletions(-)

diff --git a/sshdol/base.py b/sshdol/base.py
index cb2d2a2..4aa7b82 100644
--- a/sshdol/base.py
+++ b/sshdol/base.py
@@ -110,8 +110,8 @@ def missing_host_key(self, client, hostname, key):
         raise paramiko.SSHException(
             f"The host key of {hostname!r} is not in known_hosts, so the connection "
             "was refused. Verify and record it by connecting once with `ssh`, set "
-            "StrictHostKeyChecking in your ssh config for this host, or pass "
-            "missing_host_key_policy=paramiko.AutoAddPolicy to accept unknown keys."
+            "`StrictHostKeyChecking accept-new` in your ssh config for this host, or "
+            "pass missing_host_key_policy=paramiko.AutoAddPolicy to accept unknown keys."
         )
 
 
@@ -145,17 +145,92 @@ def resolve_host_key_policy(missing_host_key_policy=None, ssh_config=None):
     return RejectUnknownHostKey()
 
 
-def _known_hosts_files(ssh_config=None):
+_DFLT_GLOBAL_KNOWN_HOSTS = "/etc/ssh/ssh_known_hosts /etc/ssh/ssh_known_hosts2"
+_DFLT_USER_KNOWN_HOSTS = "~/.ssh/known_hosts ~/.ssh/known_hosts2"
+
+
+def _config_file_list(value, default):
+    if value is None:
+        value = default
+    if isinstance(value, (list, tuple)):
+        value = " ".join(value)
+    return value.split()
+
+
+def known_hosts_files(ssh_config=None):
+    """The known_hosts files that apply, as OpenSSH chooses them.
+
+    ``GlobalKnownHostsFile`` / ``UserKnownHostsFile`` from the host's ssh config
+    replace the defaults (``/dev/null`` or ``none`` means "no file").
+    Only existing files are returned.
+    """
+    ssh_config = ssh_config or {}
+    files = _config_file_list(
+        ssh_config.get("globalknownhostsfile"), _DFLT_GLOBAL_KNOWN_HOSTS
+    ) + _config_file_list(ssh_config.get("userknownhostsfile"), _DFLT_USER_KNOWN_HOSTS)
     files = [
-        "/etc/ssh/ssh_known_hosts",
-        os.path.expanduser("~/.ssh/known_hosts"),
+        os.path.expanduser(f) for f in files if f.lower() not in _NULL_KNOWN_HOSTS_FILES
     ]
-    for f in (ssh_config or {}).get("userknownhostsfile", "").split():
-        if f.lower() not in _NULL_KNOWN_HOSTS_FILES:
-            files.append(os.path.expanduser(f))
     return [f for f in files if os.path.isfile(f)]
 
 
+def read_known_hosts(files) -> paramiko.HostKeys:
+    """Parse known_hosts ``files`` leniently into a ``paramiko.HostKeys``.
+
+    Lines paramiko can't use (``@cert-authority`` / ``@revoked`` markers, key types
+    it doesn't support, malformed or undecodable lines) are skipped instead of
+    aborting the whole file. Note that ``@revoked`` entries are therefore not
+    honoured.
+    """
+    host_keys = paramiko.HostKeys()
+    for path in files:
+        try:
+            with open(path, "rb") as f:
+                raw_lines = f.read().splitlines()
+        except OSError:
+            continue
+        for lineno, raw in enumerate(raw_lines, 1):
+            try:
+                line = raw.decode("utf-8").strip()
+                if not line or line.startswith(("#", "@")):
+                    continue
+                entry = paramiko.hostkeys.HostKeyEntry.from_line(line, lineno)
+            except Exception:
+                continue
+            if entry is None:
+                continue
+            for hostname in entry.hostnames:
+                host_keys.add(hostname, entry.key.get_name(), entry.key)
+    return host_keys
+
+
+def _host_key_lookup_name(hostname, port=22):
+    return hostname if int(port) == 22 else f"[{hostname}]:{port}"
+
+
+def _seed_host_keys(client, known, *, hostname, port=22, alias=None):
+    """Add the known keys for this server to ``client``, under the name paramiko checks.
+
+    Besides every known_hosts entry, the entries recorded under ``HostKeyAlias`` (if
+    configured) or under the lower-cased host name are copied to the exact name
+    paramiko will look up, as OpenSSH would match them.
+    """
+    client_keys = client.get_host_keys()
+    for name in known.keys():
+        for key_type, key in known[name].items():
+            client_keys.add(name, key_type, key)
+    if not hostname:
+        return
+    lookup_name = _host_key_lookup_name(hostname, port)
+    if alias:
+        candidates = [alias, _host_key_lookup_name(alias, port)]
+    else:
+        candidates = [lookup_name.lower()]
+    for candidate in candidates:
+        for key_type, key in (known.lookup(candidate) or {}).items():
+            client_keys.add(lookup_name, key_type, key)
+
+
 def split_path(path: str) -> tuple[str, str]:
     """
     Split a path into directory and file parts.
@@ -217,6 +292,7 @@ def __init__(
         strict_contains=False,  # Whether to raise KeyError or return False for deep paths in __contains__
         allow_escape=False,  # Whether keys may point outside rootdir (absolute or ``..``)
         missing_host_key_policy=None,  # paramiko policy for unknown host keys (see below)
+        _pinned_host_key=None,  # internal: (name, key) a parent instance connected with
     ):
         """
         Initialize an SSH connection with read-only file access.
@@ -288,20 +364,20 @@ def __init__(
         )
 
         # Initialize the SSH connection
-        ssh_config = get_ssh_config_for_host(host) if host else {}
+        if host and not all([user, url]):
+            ssh_config = get_ssh_config_for_host(host)
+        else:
+            try:  # only needed for host-key options here; don't fail on a bad config
+                ssh_config = get_ssh_config_for_host(host) if host else {}
+            except Exception:
+                ssh_config = {}
         self._ssh = paramiko.SSHClient()
-        for known_hosts_file in _known_hosts_files(ssh_config):
-            try:
-                self._ssh.load_system_host_keys(known_hosts_file)
-            except (OSError, paramiko.SSHException):
-                pass  # an unreadable known_hosts file just contributes no keys
         self._ssh.set_missing_host_key_policy(
             resolve_host_key_policy(missing_host_key_policy, ssh_config)
         )
 
         # If a host alias is provided, try to get config from SSH config file
         if host and not all([user, url]):
-
             # Use values from config if not explicitly provided
             user = user or ssh_config.get("user")
             url = url or ssh_config.get("hostname")
@@ -323,12 +399,33 @@ def __init__(
                     key_filename = expanded_path
                     break
 
+        # Known host keys (and, for a subdirectory instance, the key its parent saw)
+        _seed_host_keys(
+            self._ssh,
+            read_known_hosts(known_hosts_files(ssh_config)),
+            hostname=url,
+            port=port,
+            alias=ssh_config.get("hostkeyalias"),
+        )
+        if _pinned_host_key is not None:
+            pinned_name, pinned_key = _pinned_host_key
+            self._ssh.get_host_keys().add(
+                pinned_name, pinned_key.get_name(), pinned_key
+            )
+
         # Connect using appropriate authentication method
         if key_filename:
             self._ssh.connect(url, port=port, username=user, key_filename=key_filename)
         else:
             self._ssh.connect(url, port=port, username=user, password=password)
 
+        # Remember the server key, so subdirectory instances (new connections) must
+        # see the same one rather than trusting a new key on first use again.
+        self._pinned_host_key = (
+            _host_key_lookup_name(url, port),
+            self._ssh.get_transport().get_remote_server_key(),
+        )
+
         self._sftp = self._ssh.open_sftp()
         self.rootdir = rootdir
         # Save connection details for auxiliary operations like rsync
@@ -348,6 +445,8 @@ def __init__(
     def _key_to_path(self, k):
         """Normalize key ``k`` to a path, refusing (``KeyError``) one that leaves rootdir."""
         path = normalize_path(k)
+        if "\x00" in path:
+            raise KeyError(f"Key contains a NUL character: {k!r}")
         if not getattr(self, "_allow_escape", False) and escapes_root(path):
             raise KeyError(
                 f"Key points outside rootdir (pass allow_escape=True to allow): {k}"
@@ -520,7 +619,11 @@ def __getitem__(self, k):
                     )
 
                 # Create a completely new connection for the subdirectory
-                new_instance = type(self)(**params, rootdir=new_rootdir)
+                new_instance = type(self)(
+                    **params,
+                    rootdir=new_rootdir,
+                    _pinned_host_key=getattr(self, "_pinned_host_key", None),
+                )
                 return new_instance
 
             # Try to open as a file
@@ -550,7 +653,11 @@ def __getitem__(self, k):
                 )
 
             # Create a completely new connection for the subdirectory
-            new_instance = type(self)(**params, rootdir=new_rootdir)
+            new_instance = type(self)(
+                **params,
+                rootdir=new_rootdir,
+                _pinned_host_key=getattr(self, "_pinned_host_key", None),
+            )
             return new_instance
 
         # If it's a file, return its contents
diff --git a/sshdol/tests/test_base.py b/sshdol/tests/test_base.py
index f91cd9d..1017bff 100644
--- a/sshdol/tests/test_base.py
+++ b/sshdol/tests/test_base.py
@@ -1,5 +1,6 @@
 """Test base module for sshdol."""
 
+import os
 import socket
 import pytest
 from sshdol.base import SshFiles, get_ssh_config_for_host
@@ -37,6 +38,10 @@ def _ssh_available(*, connect_timeout=3):
         list(store)
         return True
     except Exception:
+        # When a test server was configured explicitly (as in CI), a failure to
+        # connect is a test failure, not a reason to skip the whole module.
+        if "SSH_TEST_HOST" in os.environ:
+            raise
         return False
 
 
diff --git a/sshdol/tests/test_confinement.py b/sshdol/tests/test_confinement.py
index bad1f91..6b3a58e 100644
--- a/sshdol/tests/test_confinement.py
+++ b/sshdol/tests/test_confinement.py
@@ -6,7 +6,10 @@
 from sshdol.base import (
     RejectUnknownHostKey,
     SshFiles,
+    _seed_host_keys,
     escapes_root,
+    known_hosts_files,
+    read_known_hosts,
     resolve_host_key_policy,
 )
 
@@ -35,9 +38,21 @@ def test_escaping_keys_are_refused(key):
     assert key not in s
 
 
-@pytest.mark.parametrize("key", ["a", "a/b.txt", "a/../b", "./a", "a/"])
-def test_keys_inside_root_pass(key):
-    assert not escapes_root(_store()._key_to_path(key))
+@pytest.mark.parametrize(
+    "key, path", [("a", "a"), ("a/b.txt", "a/b.txt"), ("a/../b", "a/../b"), ("a/", "a")]
+)
+def test_keys_inside_root_pass(key, path):
+    assert _store()._key_to_path(key) == path
+
+
+def test_nul_in_key_refused():
+    with pytest.raises(KeyError, match="NUL"):
+        _store()._key_to_path("..\x00x")
+
+
+def test_escaping_key_not_contained_even_when_strict():
+    s = _store(_max_levels=None, _strict_contains=True)
+    assert "../x" not in s
 
 
 def test_allow_escape_opt_in():
@@ -69,3 +84,60 @@ def test_explicit_host_key_policy_wins():
     )
     policy = paramiko.WarningPolicy()
     assert resolve_host_key_policy(policy) is policy
+
+
+@pytest.fixture
+def ecdsa_key():
+    return paramiko.ECDSAKey.generate()
+
+
+def _line(host, key):
+    return f"{host} {key.get_name()} {key.get_base64()}"
+
+
+def test_read_known_hosts_skips_unusable_lines(tmp_path, ecdsa_key):
+    f = tmp_path / "known_hosts"
+    f.write_bytes(
+        "\n".join(
+            [
+                "# comment",
+                "@cert-authority *.example.org " + ecdsa_key.get_name() + " AAAA",
+                "@revoked other.example " + ecdsa_key.get_name() + " AAAA",
+                "garbage line",
+                "bad.example ssh-ed25519 !!!notbase64!!!",
+            ]
+        ).encode()
+        + b"\n\xff\xfe not utf8\n"
+        + _line("good.example", ecdsa_key).encode()
+        + b"\n"
+    )
+    known = read_known_hosts([str(f)])
+    assert known.lookup("good.example")[ecdsa_key.get_name()] == ecdsa_key
+
+
+def test_known_hosts_files_follow_ssh_config(tmp_path):
+    custom = tmp_path / "kh"
+    custom.write_text("")
+    assert known_hosts_files(
+        {"userknownhostsfile": str(custom), "globalknownhostsfile": "/dev/null"}
+    ) == [str(custom)]
+    assert (
+        known_hosts_files(
+            {"userknownhostsfile": "/dev/null", "globalknownhostsfile": "none"}
+        )
+        == []
+    )
+
+
+def test_seed_host_keys_matches_case_and_alias(ecdsa_key):
+    known = paramiko.HostKeys()
+    known.add("myhost.example", ecdsa_key.get_name(), ecdsa_key)
+    known.add("[aliased]:2222", ecdsa_key.get_name(), ecdsa_key)
+
+    client = paramiko.SSHClient()
+    _seed_host_keys(client, known, hostname="MyHost.Example")
+    assert client.get_host_keys().lookup("MyHost.Example")
+
+    client = paramiko.SSHClient()
+    _seed_host_keys(client, known, hostname="10.0.0.1", port=2222, alias="aliased")
+    assert client.get_host_keys().lookup("[10.0.0.1]:2222")

From 7a8cf3db4363fc174e9f85dba15f32b589b442d6 Mon Sep 17 00:00:00 2001
From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com>
Date: Tue, 22 Sep 2026 16:32:19 +0000
Subject: [PATCH 3/3] review: copy only matching host keys (linear loading);
 fail on unreachable configured test server

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 sshdol/base.py            | 21 ++++++++-------------
 sshdol/tests/test_base.py |  2 ++
 2 files changed, 10 insertions(+), 13 deletions(-)

diff --git a/sshdol/base.py b/sshdol/base.py
index 4aa7b82..5030e8d 100644
--- a/sshdol/base.py
+++ b/sshdol/base.py
@@ -197,10 +197,10 @@ def read_known_hosts(files) -> paramiko.HostKeys:
                 entry = paramiko.hostkeys.HostKeyEntry.from_line(line, lineno)
             except Exception:
                 continue
-            if entry is None:
-                continue
-            for hostname in entry.hostnames:
-                host_keys.add(hostname, entry.key.get_name(), entry.key)
+            if entry is not None:
+                # Append directly (as HostKeys.load does): HostKeys.add rescans all
+                # entries on every call, which is quadratic for large files.
+                host_keys._entries.append(entry)
     return host_keys
 
 
@@ -211,21 +211,16 @@ def _host_key_lookup_name(hostname, port=22):
 def _seed_host_keys(client, known, *, hostname, port=22, alias=None):
     """Add the known keys for this server to ``client``, under the name paramiko checks.
 
-    Besides every known_hosts entry, the entries recorded under ``HostKeyAlias`` (if
-    configured) or under the lower-cased host name are copied to the exact name
-    paramiko will look up, as OpenSSH would match them.
+    Only the entries that apply to this server are copied: those recorded under the
+    name paramiko looks up, its lower-cased form, or ``HostKeyAlias`` (if set).
     """
-    client_keys = client.get_host_keys()
-    for name in known.keys():
-        for key_type, key in known[name].items():
-            client_keys.add(name, key_type, key)
     if not hostname:
         return
+    client_keys = client.get_host_keys()
     lookup_name = _host_key_lookup_name(hostname, port)
+    candidates = [lookup_name, lookup_name.lower()]
     if alias:
         candidates = [alias, _host_key_lookup_name(alias, port)]
-    else:
-        candidates = [lookup_name.lower()]
     for candidate in candidates:
         for key_type, key in (known.lookup(candidate) or {}).items():
             client_keys.add(lookup_name, key_type, key)
diff --git a/sshdol/tests/test_base.py b/sshdol/tests/test_base.py
index 1017bff..16cf151 100644
--- a/sshdol/tests/test_base.py
+++ b/sshdol/tests/test_base.py
@@ -31,6 +31,8 @@ def _ssh_available(*, connect_timeout=3):
         with socket.create_connection((hostname, port), timeout=connect_timeout):
             pass
     except OSError:
+        if "SSH_TEST_HOST" in os.environ:  # explicitly configured (as in CI): fail
+            raise
         return False
     # Reachable — confirm we can actually open the store and do a trivial op.
     try: