security(personas): vet the persona repo URL before cloning it (#521)

coworker/personas/loading.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import re
 import subprocess
 from pathlib import Path
 from typing import Callable, Optional
@@ -69,16 +70,59 @@ def capability_set(m: PersonaManifest) -> set[str]:
     return caps
 
 
+# A persona repo is named by whoever asks for the install, and `git clone` treats parts of
+# that string as instructions rather than as an address: `ext::sh -c ...` runs a command
+# through git's remote-helper mechanism, and a URL starting with `-` is read as an option.
+# Both execute code BEFORE any manifest is parsed, so before the consent screen the install
+# flow relies on. The address must therefore be vetted here, not at the consent step.
+_ALLOWED_GIT_SCHEMES = ("https://", "ssh://", "git://")
+# `user@host:path` — git's scp-like form, the usual way a private persona repo is named.
+_SCP_LIKE = re.compile(r"\A[A-Za-z0-9_.+-]+@[A-Za-z0-9_.-]+:[^\s]*\Z")
+
+
+def validate_git_url(url: str) -> str:
+    """The persona repo URL, or ValueError naming why it was refused.
+
+    Allows the transports that only ever fetch a repository (https, ssh, git, and the
+    scp-like `user@host:path`). Everything else is refused, including any `<helper>::`
+    remote-helper form and local `file://` paths.
+    """
+    if not isinstance(url, str):
+        raise ValueError("persona repo URL must be a string")
+    url = url.strip()
+    if not url:
+        raise ValueError("persona repo URL is empty")
+    if any(c.isspace() or ord(c) < 0x20 for c in url):
+        raise ValueError(
+            f"refusing persona repo URL with whitespace or control characters: {url!r}"
+        )
+    if url.startswith("-"):
+        raise ValueError(
+            f"refusing persona repo URL that git would read as an option: {url!r}"
+        )
+    if "::" in url:
+        # `ext::`, `fd::`, or any other git-remote-<helper> transport: the part after the
+        # marker is handed to a helper program, which is arbitrary code execution.
+        raise ValueError(f"refusing persona repo URL using a git remote helper: {url!r}")
+    lowered = url.lower()
+    if lowered.startswith(_ALLOWED_GIT_SCHEMES) or _SCP_LIKE.match(url):
+        return url
+    raise ValueError(
+        f"refusing persona repo URL: {url!r} — use https://, ssh://, git://, "
+        "or user@host:path"
+    )
+
+
 def git_clone(
     url: str, dest: Path
 ) -> None:  # pragma: no cover - exercised via injection
     """Shallow-clone a persona repo. Injectable so tests don't touch the network."""
+    url = validate_git_url(url)
     dest.parent.mkdir(parents=True, exist_ok=True)
-    subprocess.run(
-        ["git", "clone", "--depth", "1", url, str(dest)],
-        check=True,
-        capture_output=True,
-    )
+    # `-c protocol.ext.allow=never` is the second lock on the remote-helper path that
+    # `validate_git_url` already refuses; `--` keeps the URL out of git's option parser.
+    git = ["git", "-c", "protocol.ext.allow=never", "clone", "--depth", "1", "--"]
+    subprocess.run(git + [url, str(dest)], check=True, capture_output=True)
 
 
 def cache_dir_for(url: str, base: Path) -> Path:
@@ -95,6 +139,10 @@ def clone_persona_repo(
     url: str, base: Path, *, clone: Callable[[str, Path], None] = git_clone
 ) -> Path:
     """Clone (or reuse) a persona repo under ``base`` and return its directory."""
+    # Vetted here as well as in `git_clone`: an injected clone (tests, future callers) must
+    # not become a way around the address check, and a refused URL should never reach the
+    # cache-directory naming either.
+    url = validate_git_url(url)
     dest = cache_dir_for(url, base)
     if not dest.is_dir():
         clone(url, dest)

tests/test_persona_loading.py

@@ -145,3 +145,76 @@ def test_adding_a_connector_grows_capabilities_and_forces_reconsent(tmp_path):
     summaries = reg.install_from_dir(_persona_dir(tmp_path, text=widened))
     assert summaries[0]["replaces"]["capabilities_grew"] is True
     assert reg.is_enabled("acme-ops") is False  # re-consent required
+
+
+# -- persona repo URL vetting (#521) -------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "ext::sh -c 'touch /tmp/pwned'",  # remote helper: runs a command
+        "fd::/dev/null",
+        "--upload-pack=touch /tmp/pwned",  # git reads a leading dash as an option
+        "-u./payload",
+        "file:///etc",  # local path, not a repo the user meant to name
+        "https://example.com/a repo.git",  # whitespace splits into extra argv
+    ],
+)
+def test_refused_persona_repo_urls(url):
+    from coworker.personas.loading import validate_git_url
+
+    with pytest.raises(ValueError):
+        validate_git_url(url)
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://example.com/acme/persona.git",
+        "ssh://git@example.com/acme/persona.git",
+        "git://example.com/acme/persona.git",
+        "git@example.com:acme/persona.git",
+    ],
+)
+def test_accepted_persona_repo_urls(url):
+    from coworker.personas.loading import validate_git_url
+
+    assert validate_git_url(url) == url
+
+
+def test_install_from_git_refuses_helper_url_before_cloning(tmp_path):
+    """The address is vetted ahead of the clone, so an injected clone can't be reached
+    either — `ext::` executes before any manifest or consent screen exists."""
+    reg = PersonaRegistry(state_path=tmp_path / "personas.json")
+    called: list[str] = []
+
+    def fake_clone(url, dest):  # pragma: no cover - must never run
+        called.append(url)
+
+    with pytest.raises(ValueError):
+        reg.install_from_git(
+            "ext::sh -c 'touch /tmp/pwned'",
+            cache_base=tmp_path / "cache",
+            clone=fake_clone,
+        )
+    assert called == []
+    assert "acme-ops" not in reg.ids()
+
+
+def test_git_clone_argv_ends_option_parsing_and_disables_ext(tmp_path, monkeypatch):
+    from coworker.personas import loading
+
+    seen: dict = {}
+
+    def fake_run(argv, **kwargs):
+        seen["argv"] = argv
+        return None
+
+    monkeypatch.setattr(loading.subprocess, "run", fake_run)
+    loading.git_clone("https://example.com/acme/persona.git", tmp_path / "dest")
+    argv = seen["argv"]
+    assert "--" in argv and argv.index("--") < argv.index(
+        "https://example.com/acme/persona.git"
+    )
+    assert "protocol.ext.allow=never" in argv