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,14 @@
 """Svn repository."""
 
 import contextlib
+import functools
 import os
 import pathlib
 import re
 from collections.abc import Callable, Generator, Sequence
 from pathlib import Path
 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,6 +17,38 @@
 
 logger = get_logger(__name__)
 
+_SSH_HOST_KEY_MSGS = ("host key verification failed", "authenticity of host")
+
+
+# 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() -> dict[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 env
+
+
+def _raise_if_ssh_host_key_error(url: str, exc: SubprocessCommandError) -> None:
+    """Raise a helpful RuntimeError 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 RuntimeError(
+            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 get_svn_version() -> tuple[str, str]:
     """Get the name and version of svn."""
@@ -49,9 +83,14 @@ 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_on_cmdline(
+                logger,
+                ["svn", "info", self._remote, "--non-interactive"],
+                env=_extend_env_for_non_interactive_mode(),
+            )
             return True
         except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(self._remote, exc)
             if exc.stderr.startswith("svn: E170013"):
                 raise RuntimeError(
                     f">>>{exc.cmd}<<< failed!\n"
@@ -67,20 +106,30 @@ def list_of_branches(self) -> list[str]:
             result = run_on_cmdline(
                 logger,
                 ["svn", "ls", "--non-interactive", f"{self._remote}/branches"],
+                env=_extend_env_for_non_interactive_mode(),
             )
             return [
                 line.strip("/\r")
                 for line in result.stdout.decode().splitlines()
                 if line.strip("/\r")
             ]
-        except (SubprocessCommandError, RuntimeError):
+        except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(self._remote, exc)
+            return []
+        except 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"]
-        )
+        try:
+            result = run_on_cmdline(
+                logger,
+                ["svn", "ls", "--non-interactive", f"{self._remote}/tags"],
+                env=_extend_env_for_non_interactive_mode(),
+            )
+        except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(self._remote, exc)
+            raise
         return [
             str(tag).strip("/\r") for tag in result.stdout.decode().split("\n") if tag
         ]
@@ -116,7 +165,9 @@ 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]
+                logger,
+                ["svn", "ls", "--non-interactive", url_path],
+                env=_extend_env_for_non_interactive_mode(),
             )
             entries: list[tuple[str, bool]] = []
             for line in result.stdout.decode().splitlines():
@@ -126,7 +177,10 @@ def ls_tree(self, url_path: str) -> list[tuple[str, bool]]:
                 is_dir = line.endswith("/")
                 entries.append((line.rstrip("/"), is_dir))
             return entries
-        except (SubprocessCommandError, RuntimeError):
+        except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(url_path, exc)
+            return []
+        except RuntimeError:
             return []
 
 
@@ -176,7 +230,13 @@ def externals_from_url(url: str, revision: str = "") -> list[External]:
         if revision:
             cmd += ["--revision", revision]
         cmd += [url]
-        result = run_on_cmdline(logger, cmd)
+        try:
+            result = run_on_cmdline(
+                logger, cmd, env=_extend_env_for_non_interactive_mode()
+            )
+        except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(url, exc)
+            raise
         repo_root = SvnRepo.get_info_from_target(url)["Repository Root"]
         normalized = SvnRepo._normalize_url_prefix(result.stdout.decode(), url)
         return SvnRepo._parse_externals(normalized, repo_root)
@@ -292,9 +352,12 @@ 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()]
+                logger,
+                ["svn", "info", "--non-interactive", target.strip()],
+                env=_extend_env_for_non_interactive_mode(),
             ).stdout.decode()
         except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(target, exc)
             if exc.stderr.startswith("svn: E170013"):
                 raise RuntimeError(
                     f">>>{exc.cmd}<<< failed!\n"
@@ -324,8 +387,8 @@ 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(
+        try:
+            result = run_on_cmdline(
                 logger,
                 [
                     "svn",
@@ -335,10 +398,12 @@ def get_last_changed_revision(target: str | Path) -> str:
                     "last-changed-revision",
                     target_str,
                 ],
+                env=_extend_env_for_non_interactive_mode(),
             )
-            .stdout.decode()
-            .strip()
-        )
+        except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(target_str, exc)
+            raise
+        return str(result.stdout.decode().strip())
 
     @staticmethod
     def untracked_files(path: str, ignore: Sequence[str]) -> list[str]:
@@ -377,24 +442,31 @@ 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],
-        )
+        try:
+            run_on_cmdline(
+                logger,
+                ["svn", "export", "--non-interactive", "--force"]
+                + (["--revision", rev] if rev else [])
+                + [url, dst],
+                env=_extend_env_for_non_interactive_mode(),
+            )
+        except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(url, exc)
+            raise
 
     @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]
+        try:
+            result = run_on_cmdline(
+                logger,
+                ["svn", "list", "--non-interactive", url_path],
+                env=_extend_env_for_non_interactive_mode(),
             )
-            .stdout.decode()
-            .splitlines()
-        ]
+        except SubprocessCommandError as exc:
+            _raise_if_ssh_host_key_error(url_path, exc)
+            raise
+        return [str(line) for line in result.stdout.decode().splitlines()]
 
     @staticmethod
     def ignored_files(path: str) -> Sequence[str]:

tests/test_svn.py

@@ -520,3 +520,88 @@ def test_externals_from_url_nonstd_layout_branch_is_space():
         assert result[0].url == nonstd_url
         assert result[0].revision == ""
         assert result[0].path == "Database"
+
+
+@pytest.mark.parametrize(
+    "method,url,call_args",
+    [
+        ("is_svn", "svn+ssh://svn.code.sf.net/project", ()),
+        ("list_of_tags", "svn+ssh://svn.code.sf.net/project", ()),
+        ("list_of_branches", "svn+ssh://svn.code.sf.net/project", ()),
+        (
+            "ls_tree",
+            "svn+ssh://svn.code.sf.net/project",
+            ("svn+ssh://svn.code.sf.net/project",),
+        ),
+    ],
+)
+def test_svn_remote_raises_hint_on_ssh_host_key_failure(method, url, call_args):
+    stderr = "Host key verification failed."
+    with patch("dfetch.vcs.svn.run_on_cmdline") as mock_run:
+        mock_run.side_effect = SubprocessCommandError(["svn"], "", stderr, 1)
+
+        with pytest.raises(RuntimeError, match="known hosts"):
+            getattr(SvnRemote(url), method)(*call_args)
+
+
+def test_get_info_from_target_raises_hint_on_ssh_host_key_failure():
+    stderr = "Host key verification failed."
+    with patch("dfetch.vcs.svn.run_on_cmdline") as mock_run:
+        mock_run.side_effect = SubprocessCommandError(["svn", "info"], "", stderr, 1)
+
+        with pytest.raises(RuntimeError, match="known hosts"):
+            SvnRepo.get_info_from_target("svn+ssh://svn.code.sf.net/project")
+
+
+@pytest.mark.parametrize(
+    "method,url",
+    [
+        ("externals_from_url", "svn+ssh://svn.code.sf.net/project"),
+        ("get_last_changed_revision", "svn+ssh://svn.code.sf.net/project"),
+        ("export", "svn+ssh://svn.code.sf.net/project"),
+        ("files_in_path", "svn+ssh://svn.code.sf.net/project"),
+    ],
+)
+def test_svn_repo_raises_hint_on_ssh_host_key_failure(method, url):
+    stderr = "Host key verification failed."
+    with patch("dfetch.vcs.svn.run_on_cmdline") as mock_run:
+        mock_run.side_effect = SubprocessCommandError(["svn"], "", stderr, 1)
+
+        with pytest.raises(RuntimeError, match="known hosts"):
+            getattr(SvnRepo, method)(url)
+
+
+def test_ssh_hint_includes_hostname():
+    stderr = "Host key verification failed."
+    with patch("dfetch.vcs.svn.run_on_cmdline") as mock_run:
+        mock_run.side_effect = SubprocessCommandError(["svn"], "", stderr, 1)
+
+        with pytest.raises(RuntimeError, match="svn.code.sf.net"):
+            SvnRemote("svn+ssh://svn.code.sf.net/project").is_svn()
+
+
+def test_ssh_hint_includes_user_when_present_in_url():
+    stderr = "Host key verification failed."
+    with patch("dfetch.vcs.svn.run_on_cmdline") as mock_run:
+        mock_run.side_effect = SubprocessCommandError(["svn"], "", stderr, 1)
+
+        with pytest.raises(RuntimeError, match="myuser@svn.code.sf.net"):
+            SvnRemote("svn+ssh://myuser@svn.code.sf.net/project").is_svn()
+
+
+def test_svn_ssh_env_has_batch_mode():
+    from dfetch.vcs.svn import _extend_env_for_non_interactive_mode
+
+    _extend_env_for_non_interactive_mode.cache_clear()
+    env = _extend_env_for_non_interactive_mode()
+    assert "BatchMode=yes" in env["SVN_SSH"]
+
+
+def test_svn_ssh_env_preserves_existing_batch_mode(monkeypatch):
+    from dfetch.vcs.svn import _extend_env_for_non_interactive_mode
+
+    monkeypatch.setenv("SVN_SSH", "ssh -o BatchMode=yes -i /my/key")
+    _extend_env_for_non_interactive_mode.cache_clear()
+    env = _extend_env_for_non_interactive_mode()
+    assert env["SVN_SSH"].count("BatchMode=yes") == 1
+    assert "-i /my/key" in env["SVN_SSH"]