diff --git a/roboco/services/workspace.py b/roboco/services/workspace.py index f0bb5d19..ea9b37bd 100644 --- a/roboco/services/workspace.py +++ b/roboco/services/workspace.py @@ -1049,10 +1049,21 @@ class WorkspaceService: workspace=str(workspace), ) except subprocess.CalledProcessError as e: + # F063: a failure anywhere in clone/configure/leakcheck/own leaves + # a half-configured workspace on disk. If _configure_git raised + # before its `remote set-url` scrub, .git/config still carries the + # tokenized auth URL (the project PAT); _assert_no_pat_leak never + # ran, and the next ensure_workspace's health short-circuit would + # skip straight past the leak — mounting the agent on a workspace + # whose .git/config lets it read+exfiltrate the PAT. Destroy the + # workspace so the next ensure_workspace re-clones from scratch. + shutil.rmtree(workspace, ignore_errors=True) raise WorkspaceError( f"Failed to clone repository: {e.stderr or e.stdout}" ) from e except subprocess.TimeoutExpired as e: + # Same PAT-leak hygiene as the CalledProcessError branch. + shutil.rmtree(workspace, ignore_errors=True) raise WorkspaceError( f"Clone timed out after {settings.workspace_clone_timeout}s" ) from e diff --git a/tests/unit/services/test_workspace_clone_pat_leak_cleanup.py b/tests/unit/services/test_workspace_clone_pat_leak_cleanup.py new file mode 100644 index 00000000..4f5501ac --- /dev/null +++ b/tests/unit/services/test_workspace_clone_pat_leak_cleanup.py @@ -0,0 +1,123 @@ +"""F063 — a failed ``_configure_git`` must not leave the project PAT on disk. + +``_clone_repo`` runs ``_do_clone`` (which writes the tokenized auth URL into +``.git/config``), then ``_configure_git`` (which scrubs it via +``git remote set-url origin ``), then ``_assert_no_pat_leak``. If +``_configure_git`` raises ``CalledProcessError`` BEFORE the scrub completes +(disk error, permission issue, broken git), the PAT stays in ``.git/config`` +and ``_assert_no_pat_leak`` never runs. The except clauses raised +``WorkspaceError`` without removing the workspace, so on the next +``ensure_workspace`` the health short-circuit (a valid ``.git`` with HEAD + +objects) skipped straight past the leak — the agent was then mounted on a +workspace whose ``.git/config`` still carried ``https://TOKEN@github.com/...``, +letting it read and exfiltrate the project PAT. + +The fix: the clone-failure except clauses ``rmtree`` the workspace before +raising, so a half-configured clone is destroyed and the next +``ensure_workspace`` re-clones from scratch instead of short-circuiting past +the leak. +""" + +from __future__ import annotations + +import subprocess +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest +from roboco.services.workspace import WorkspaceError, WorkspaceService + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + +def _service() -> WorkspaceService: + session = MagicMock() + return WorkspaceService(session) + + +def _completed(argv: list[str]) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(argv, returncode=0, stdout="", stderr="") + + +def _leak_side_effect( + workspace: Path, token: str +) -> Callable[..., subprocess.CompletedProcess[str]]: + """``subprocess.run`` side effect: a real ``git clone`` writes the + tokenized auth URL into ``.git/config``; a failing ``_configure_git`` + raises before the scrub. Simulate both so the leak is on disk when the + except clause fires.""" + + def _impl(argv: list[str], **_kw: object) -> subprocess.CompletedProcess[str]: + if argv and argv[0] == "git" and len(argv) > 1 and argv[1] == "clone": + # git clone writes the auth URL into .git/config. + cfg = workspace / ".git" / "config" + cfg.parent.mkdir(parents=True, exist_ok=True) + cfg.write_text( + f'[remote "origin"]\n\turl = https://{token}@github.com/o/r\n' + ) + return _completed(argv) + if argv and argv[0] == "git" and len(argv) > 1 and argv[1] == "config": + # _configure_git's first `git config` fails before the scrub. + raise subprocess.CalledProcessError( + returncode=128, cmd=argv, stderr="disk error mid-configure" + ) + return _completed(argv) + + return _impl + + +@pytest.mark.asyncio +async def test_configure_git_failure_rmtrees_leaked_workspace(tmp_path: Path) -> None: + """A ``_configure_git`` failure after the clone wrote the PAT into + ``.git/config`` must destroy the workspace — no PAT left on disk.""" + token = "ghp_LEAKEDTOKEN" + workspace = tmp_path / "ws" + svc = _service() + + with ( + patch( + "roboco.services.workspace.subprocess.run", + side_effect=_leak_side_effect(workspace, token), + ), + pytest.raises(WorkspaceError), + ): + await svc._clone_repo( + workspace, + git_url="https://github.com/o/r", + default_branch="master", + git_token=token, + ) + + # The half-configured clone (with the PAT in .git/config) must be gone, + # so the next ensure_workspace re-clones instead of short-circuiting past + # the leak. + assert not workspace.exists() + + +@pytest.mark.asyncio +async def test_clone_timeout_rmtrees_partial_workspace(tmp_path: Path) -> None: + """A clone timeout must likewise not leave a partial workspace behind — + the same rmtree-on-failure covers the TimeoutExpired branch.""" + workspace = tmp_path / "ws-timeout" + workspace.mkdir(parents=True) + (workspace / ".git").mkdir() + (workspace / ".git" / "config").write_text("partial") + svc = _service() + + def _impl(argv: list[str], **_kw: object) -> subprocess.CompletedProcess[str]: + raise subprocess.TimeoutExpired(cmd=argv, timeout=1) + + with ( + patch("roboco.services.workspace.subprocess.run", side_effect=_impl), + pytest.raises(WorkspaceError), + ): + await svc._clone_repo( + workspace, + git_url="https://github.com/o/r", + default_branch="master", + git_token="ghp_X", + ) + + assert not workspace.exists()