Prevent hanging in svn+ssh

Author: spoorcc

CHANGELOG.rst

@@ -21,6 +21,7 @@ Release 0.14.0 (unreleased)
 * Prevent SSH command injection (#1152)
 * Allow manifests with no ``projects`` key so ``dfetch add`` can bootstrap empty manifest (#1197)
 * Fix ``ValueError`` when generating a PackageURL (e.g. for an SBOM) from an empty or path-only remote URL
+* Run ``svn+ssh://`` connections in non-interactive mode to prevent hanging (#1230)
 
 Release 0.13.0 (released 2026-03-30)
 ====================================

dfetch/vcs/svn.py

@@ -1,12 +1,15 @@
 """Svn repository."""
 
 import contextlib
+import functools
 import os
 import pathlib
 import re
-from collections.abc import Callable, Generator, Sequence
+from collections.abc import Callable, Generator, Mapping, Sequence
 from pathlib import Path
+from types import MappingProxyType
 from typing import NamedTuple
+from urllib.parse import urlparse
 
 from dfetch.log import get_logger
 from dfetch.util.cmdline import SubprocessCommandError, run_on_cmdline
@@ -15,11 +18,65 @@
 
 logger = get_logger(__name__)
 
+_SSH_HOST_KEY_MSGS = ("host key verification failed", "authenticity of host")
+
+
+class SshHostKeyError(RuntimeError):
+    """Raised when SVN cannot connect due to an untrusted SSH host key."""
+
+
+# As a cli tool, we can safely assume this remains stable during the runtime, caching for speed is better
+@functools.lru_cache
+def _extend_env_for_non_interactive_mode() -> Mapping[str, str]:
+    """Extend the environment vars for svn running in non-interactive mode."""
+    env = os.environ.copy()
+    ssh_cmd = env.get("SVN_SSH", "ssh")
+    if "BatchMode=" not in ssh_cmd:
+        ssh_cmd += " -o BatchMode=yes"
+    else:
+        logger.debug('BatchMode already configured in SVN_SSH: "%s"', ssh_cmd)
+    env["SVN_SSH"] = ssh_cmd
+    return MappingProxyType(env)
+
+
+def _raise_if_ssh_host_key_error(url: str, exc: SubprocessCommandError) -> None:
+    """Raise a helpful SshHostKeyError if *exc* looks like an SSH host-key failure."""
+    stderr_lower = exc.stderr.lower()
+    if any(msg in stderr_lower for msg in _SSH_HOST_KEY_MSGS):
+        parsed = urlparse(url)
+        host = parsed.hostname or url
+        target = f"{parsed.username}@{host}" if parsed.username else host
+        raise SshHostKeyError(
+            f"SSH host key verification failed while connecting to '{url}'.\n"
+            "Add the host to your known hosts file, for example by running:\n"
+            f"  ssh-keyscan {host} >> ~/.ssh/known_hosts\n"
+            "Or test the SSH connection manually:\n"
+            f"  ssh -T {target}"
+        ) from exc
+
+
+def _run_svn(args: list[str], *, url: str = "") -> str:
+    """Run an svn subcommand and return decoded stdout.
+
+    Uses --non-interactive and the non-interactive SSH env on every call.
+    SSH host-key failures are converted to SshHostKeyError so callers don't
+    need to handle that case individually.
+    """
+    try:
+        result = run_on_cmdline(
+            logger,
+            ["svn", "--non-interactive"] + args,
+            env=_extend_env_for_non_interactive_mode(),
+        )
+        return str(result.stdout.decode())
+    except SubprocessCommandError as exc:
+        _raise_if_ssh_host_key_error(url, exc)
+        raise
+
 
 def get_svn_version() -> tuple[str, str]:
     """Get the name and version of svn."""
-    result = run_on_cmdline(logger, ["svn", "--version", "--non-interactive"])
-    first_line = result.stdout.decode().split("\n")[0]
+    first_line = _run_svn(["--version"]).split("\n", maxsplit=1)[0]
     if "version" not in first_line.lower():
         raise RuntimeError(f"Unexpected svn --version output format: {first_line}")
     tool, version = first_line.replace(",", "").split("version", maxsplit=1)
@@ -49,8 +106,10 @@ def __init__(self, remote: str) -> None:
     def is_svn(self) -> bool:
         """Check if is SVN."""
         try:
-            run_on_cmdline(logger, ["svn", "info", self._remote, "--non-interactive"])
+            _run_svn(["info", self._remote], url=self._remote)
             return True
+        except SshHostKeyError:
+            raise
         except SubprocessCommandError as exc:
             if exc.stderr.startswith("svn: E170013"):
                 raise RuntimeError(
@@ -64,26 +123,19 @@ def is_svn(self) -> bool:
     def list_of_branches(self) -> list[str]:
         """List branch names from the ``branches/`` directory."""
         try:
-            result = run_on_cmdline(
-                logger,
-                ["svn", "ls", "--non-interactive", f"{self._remote}/branches"],
-            )
+            output = _run_svn(["ls", f"{self._remote}/branches"], url=self._remote)
             return [
-                line.strip("/\r")
-                for line in result.stdout.decode().splitlines()
-                if line.strip("/\r")
+                line.strip("/\r") for line in output.splitlines() if line.strip("/\r")
             ]
+        except SshHostKeyError:
+            raise
         except (SubprocessCommandError, RuntimeError):
             return []
 
     def list_of_tags(self) -> list[str]:
         """Get list of all available tags."""
-        result = run_on_cmdline(
-            logger, ["svn", "ls", "--non-interactive", f"{self._remote}/tags"]
-        )
-        return [
-            str(tag).strip("/\r") for tag in result.stdout.decode().split("\n") if tag
-        ]
+        output = _run_svn(["ls", f"{self._remote}/tags"], url=self._remote)
+        return [str(tag).strip("/\r") for tag in output.split("\n") if tag]
 
     @contextlib.contextmanager
     def browse_tree(
@@ -115,17 +167,17 @@ def ls(path: str = "") -> list[tuple[str, bool]]:
     def ls_tree(self, url_path: str) -> list[tuple[str, bool]]:
         """List immediate children of *url_path* as ``(name, is_dir)`` pairs."""
         try:
-            result = run_on_cmdline(
-                logger, ["svn", "ls", "--non-interactive", url_path]
-            )
+            output = _run_svn(["ls", url_path], url=url_path)
             entries: list[tuple[str, bool]] = []
-            for line in result.stdout.decode().splitlines():
+            for line in output.splitlines():
                 line = line.strip("\r")
                 if not line:
                     continue
                 is_dir = line.endswith("/")
                 entries.append((line.rstrip("/"), is_dir))
             return entries
+        except SshHostKeyError:
+            raise
         except (SubprocessCommandError, RuntimeError):
             return []
 
@@ -146,39 +198,25 @@ def is_svn(self) -> bool:
         """Check if is SVN."""
         try:
             with in_directory(self._path):
-                run_on_cmdline(logger, ["svn", "info", "--non-interactive"])
+                _run_svn(["info"])
             return True
         except (SubprocessCommandError, RuntimeError):
             return False
 
     def externals(self) -> list[External]:
         """Get list of externals."""
         with in_directory(self._path):
-            result = run_on_cmdline(
-                logger,
-                [
-                    "svn",
-                    "--non-interactive",
-                    "propget",
-                    "svn:externals",
-                    "-R",
-                ],
-            )
+            output = _run_svn(["propget", "svn:externals", "-R"])
             repo_root = SvnRepo.get_info_from_target()["Repository Root"]
-            return SvnRepo._parse_externals(
-                result.stdout.decode(), repo_root, toplevel=self._path
-            )
+            return SvnRepo._parse_externals(output, repo_root, toplevel=self._path)
 
     @staticmethod
     def externals_from_url(url: str, revision: str = "") -> list[External]:
         """Get list of externals from a remote SVN URL."""
-        cmd = ["svn", "--non-interactive", "propget", "svn:externals", "-R"]
-        if revision:
-            cmd += ["--revision", revision]
-        cmd += [url]
-        result = run_on_cmdline(logger, cmd)
+        extra = ["--revision", revision] if revision else []
+        output = _run_svn(["propget", "svn:externals", "-R"] + extra + [url], url=url)
         repo_root = SvnRepo.get_info_from_target(url)["Repository Root"]
-        normalized = SvnRepo._normalize_url_prefix(result.stdout.decode(), url)
+        normalized = SvnRepo._normalize_url_prefix(output, url)
         return SvnRepo._parse_externals(normalized, repo_root)
 
     @staticmethod
@@ -291,9 +329,7 @@ def _split_url(url: str, repo_root: str) -> tuple[str, str, str, str]:
     def get_info_from_target(target: str = "") -> dict[str, str]:
         """Get the info of the given target."""
         try:
-            result = run_on_cmdline(
-                logger, ["svn", "info", "--non-interactive", target.strip()]
-            ).stdout.decode()
+            output = _run_svn(["info", target.strip()], url=target)
         except SubprocessCommandError as exc:
             if exc.stderr.startswith("svn: E170013"):
                 raise RuntimeError(
@@ -306,7 +342,7 @@ def get_info_from_target(target: str = "") -> dict[str, str]:
             key.strip(): value.strip()
             for key, value in (
                 line.split(":", maxsplit=1)
-                for line in result.split(os.linesep)
+                for line in output.split(os.linesep)
                 if line and ":" in line
             )
         }
@@ -324,36 +360,16 @@ def get_last_changed_revision(target: str | Path) -> str:
                 return parsed_version.group("digits")
             raise RuntimeError(f"svnversion output was unexpected: {version}")
 
-        return str(
-            run_on_cmdline(
-                logger,
-                [
-                    "svn",
-                    "info",
-                    "--non-interactive",
-                    "--show-item",
-                    "last-changed-revision",
-                    target_str,
-                ],
-            )
-            .stdout.decode()
-            .strip()
-        )
+        return _run_svn(
+            ["info", "--show-item", "last-changed-revision", target_str],
+            url=target_str,
+        ).strip()
 
     @staticmethod
     def untracked_files(path: str, ignore: Sequence[str]) -> list[str]:
         """Get list of untracked files in the working copy."""
-        result = (
-            run_on_cmdline(
-                logger,
-                ["svn", "status", "--non-interactive", path],
-            )
-            .stdout.decode()
-            .splitlines()
-        )
-
         files = []
-        for line in result:
+        for line in _run_svn(["status", path]).splitlines():
             if line.startswith("?"):
                 file_path = line[1:].strip()
                 if not any(
@@ -377,24 +393,15 @@ def export(url: str, rev: str = "", dst: str = ".") -> None:
         """
         if rev and not rev.isdigit():
             raise ValueError(f"SVN revision must be digits only, got: {rev!r}")
-        run_on_cmdline(
-            logger,
-            ["svn", "export", "--non-interactive", "--force"]
-            + (["--revision", rev] if rev else [])
-            + [url, dst],
+        _run_svn(
+            ["export", "--force"] + (["--revision", rev] if rev else []) + [url, dst],
+            url=url,
         )
 
     @staticmethod
     def files_in_path(url_path: str) -> list[str]:
         """List all files in path at the given url."""
-        return [
-            str(line)
-            for line in run_on_cmdline(
-                logger, ["svn", "list", "--non-interactive", url_path]
-            )
-            .stdout.decode()
-            .splitlines()
-        ]
+        return _run_svn(["list", url_path], url=url_path).splitlines()
 
     @staticmethod
     def ignored_files(path: str) -> Sequence[str]:
@@ -403,16 +410,9 @@ def ignored_files(path: str) -> Sequence[str]:
             return []
 
         with in_directory(path):
-            result = (
-                run_on_cmdline(
-                    logger,
-                    ["svn", "status", "--non-interactive", "--no-ignore", "."],
-                )
-                .stdout.decode()
-                .splitlines()
-            )
+            lines = _run_svn(["status", "--no-ignore", "."]).splitlines()
 
-        return [line[1:].strip() for line in result if line.startswith("I")]
+        return [line[1:].strip() for line in lines if line.startswith("I")]
 
     @staticmethod
     def any_changes_or_untracked(path: str) -> bool:
@@ -421,18 +421,7 @@ def any_changes_or_untracked(path: str) -> bool:
             raise RuntimeError("Path does not exist.")
 
         with in_directory(path):
-            return bool(
-                run_on_cmdline(
-                    logger,
-                    [
-                        "svn",
-                        "status",
-                        ".",
-                    ],
-                )
-                .stdout.decode()
-                .splitlines()
-            )
+            return bool(_run_svn(["status", "."]).splitlines())
 
     def create_diff(
         self,
@@ -441,7 +430,7 @@ def create_diff(
         ignore: Sequence[str],
     ) -> Patch:
         """Generate a relative diff patch."""
-        cmd = ["svn", "diff", "--non-interactive", "--ignore-properties", "."]
+        cmd = ["diff", "--ignore-properties", "."]
 
         if old_revision:
             cmd.extend(
@@ -452,26 +441,15 @@ def create_diff(
             )
 
         with in_directory(self._path):
-            patch_text = run_on_cmdline(logger, cmd).stdout
+            patch_text = _run_svn(cmd)
 
         if not patch_text.strip():
             return Patch.empty().convert_type(PatchType.SVN)
-        return Patch.from_bytes(patch_text).filter(ignore)
+        return Patch.from_string(patch_text).filter(ignore)
 
     def get_username(self) -> str:
         """Get the username of the local svn repo."""
         try:
-            result = run_on_cmdline(
-                logger,
-                [
-                    "svn",
-                    "info",
-                    "--non-interactive",
-                    "--show-item",
-                    "author",
-                    self._path,
-                ],
-            )
-            return str(result.stdout.decode().strip())
+            return _run_svn(["info", "--show-item", "author", self._path]).strip()
         except SubprocessCommandError:
             return ""