diff --git a/coworker/personas/loading.py b/coworker/personas/loading.py index f7478353a..5b997863f 100644 --- a/coworker/personas/loading.py +++ b/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 `::` + 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- 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) diff --git a/tests/test_persona_loading.py b/tests/test_persona_loading.py index a78ac2a5b..869b5ed31 100644 --- a/tests/test_persona_loading.py +++ b/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