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 (#0)
 
 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,43 @@
 
 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 _ssh_target_from_url(url: str) -> str:
+    """Return the ``[user@]host`` portion of a svn+ssh URL, or the URL itself."""
+    parsed = urlparse(url)
+    host = parsed.hostname or url
+    return f"{parsed.username}@{host}" if parsed.username else host
+
+
+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):
+        target = _ssh_target_from_url(url)
+        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 {target} >> ~/.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 +88,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 +111,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 +170,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 +182,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 +235,7 @@ def externals_from_url(url: str, revision: str = "") -> list[External]:
         if revision:
             cmd += ["--revision", revision]
         cmd += [url]
-        result = run_on_cmdline(logger, cmd)
+        result = run_on_cmdline(logger, cmd, env=_extend_env_for_non_interactive_mode())
         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 +351,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"
@@ -335,6 +397,7 @@ def get_last_changed_revision(target: str | Path) -> str:
                     "last-changed-revision",
                     target_str,
                 ],
+                env=_extend_env_for_non_interactive_mode(),
             )
             .stdout.decode()
             .strip()
@@ -382,6 +445,7 @@ def export(url: str, rev: str = "", dst: str = ".") -> None:
             ["svn", "export", "--non-interactive", "--force"]
             + (["--revision", rev] if rev else [])
             + [url, dst],
+            env=_extend_env_for_non_interactive_mode(),
         )
 
     @staticmethod
@@ -390,7 +454,9 @@ def files_in_path(url_path: str) -> list[str]:
         return [
             str(line)
             for line in run_on_cmdline(
-                logger, ["svn", "list", "--non-interactive", url_path]
+                logger,
+                ["svn", "list", "--non-interactive", url_path],
+                env=_extend_env_for_non_interactive_mode(),
             )
             .stdout.decode()
             .splitlines()

example/dfetch.yaml

@@ -7,7 +7,7 @@ manifest:
     default: true                                                 # Set it as default
 
   - name: sourceforge
-    url-base: svn://svn.code.sf.net/p/
+    url-base: svn+ssh://svn.code.sf.net/p/
 
   projects:
 

tests/test_svn.py

@@ -520,3 +520,64 @@ 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",
+    [
+        ("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"),
+    ],
+)
+def test_svn_remote_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(SvnRemote(url), method)()
+
+
+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")
+
+
+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