diff --git a/author_mutation_worktree.py b/author_mutation_worktree.py index 04b6933..29dc05d 100644 --- a/author_mutation_worktree.py +++ b/author_mutation_worktree.py @@ -7,8 +7,11 @@ project's ``branches/`` directory, never from the stable control checkout. from __future__ import annotations import os +import subprocess BASE_BRANCHES = frozenset({"master", "main", "dev"}) +ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE" +AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" def _normalize_path(path: str) -> str: @@ -48,6 +51,130 @@ def resolve_mutation_workspace( return os.path.realpath(project_root) +def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str: + """Resolve ``git rev-parse --git-common-dir`` relative to *workspace_path*.""" + raw = (common_dir or "").strip() + if not raw: + return raw + if os.path.isabs(raw): + return os.path.realpath(raw) + return os.path.realpath(os.path.join(workspace_path, raw)) + + +def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str: + """Return the stable repository root for *workspace_path* via git metadata (#460).""" + path = (workspace_path or "").strip() + fallback = os.path.realpath(fallback_project_root) + if not path: + return fallback + try: + res = subprocess.run( + ["git", "-C", path, "rev-parse", "--git-common-dir"], + capture_output=True, + text=True, + check=True, + ) + common = _realpath_git_common_dir(path, res.stdout) + except Exception: + return fallback + if common.endswith(f"{os.sep}.git"): + return os.path.dirname(common) + if os.path.basename(common) == ".git": + return os.path.dirname(common) + return fallback + + +def resolve_author_mutation_context( + worktree_path: str | None, + process_project_root: str, + *, + active_worktree_env: str | None = None, + author_worktree_env: str | None = None, +) -> dict: + """Shared workspace resolution for runtime_context and mutation guards (#460).""" + workspace = resolve_mutation_workspace( + worktree_path, + process_project_root, + active_worktree_env=active_worktree_env, + author_worktree_env=author_worktree_env, + ) + process_root = os.path.realpath(process_project_root) + # Canonical repository identity comes from the MCP process checkout (#460), + # not from the declared task workspace being validated. + canonical_root = resolve_canonical_repo_root(process_root, process_root) + return { + "workspace_path": workspace, + "process_project_root": process_root, + "canonical_repo_root": canonical_root, + "roots_aligned": canonical_root == process_root, + } + + +def assess_workspace_repo_membership( + *, + workspace_path: str, + canonical_repo_root: str, +) -> dict: + """Fail closed when *workspace_path* is not a git worktree of *canonical_repo_root*.""" + workspace = os.path.realpath(workspace_path) + root = os.path.realpath(canonical_repo_root) + reasons: list[str] = [] + + if not os.path.exists(workspace): + reasons.append(f"worktree path '{workspace}' does not exist") + return _membership_assessment(False, reasons, workspace, root, None) + + if not os.path.isdir(workspace): + reasons.append(f"worktree path '{workspace}' is not a directory") + return _membership_assessment(False, reasons, workspace, root, None) + + try: + res = subprocess.run( + ["git", "-C", workspace, "rev-parse", "--git-common-dir"], + capture_output=True, + text=True, + check=True, + ) + common_dir = _realpath_git_common_dir(workspace, res.stdout) + except Exception: + reasons.append(f"worktree '{workspace}' is not a valid git repository") + return _membership_assessment(False, reasons, workspace, root, None) + + expected_dir = os.path.realpath(os.path.join(root, ".git")) + if common_dir != expected_dir: + reasons.append( + f"worktree '{workspace}' does not belong to the target repository '{root}'" + ) + return _membership_assessment(not reasons, reasons, workspace, root, common_dir) + + +def _membership_assessment( + proven: bool, + reasons: list[str], + workspace: str, + root: str, + common_dir: str | None, +) -> dict: + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "workspace_path": workspace, + "canonical_repo_root": root, + "git_common_dir": common_dir, + } + + +def format_workspace_repo_membership_error(assessment: dict) -> str: + workspace = assessment.get("workspace_path") or "(unknown)" + root = assessment.get("canonical_repo_root") or "(unknown)" + reasons = "; ".join(assessment.get("reasons") or ["unknown repository membership violation"]) + return ( + f"Branches-only mutation guard (#274): {reasons} (fail closed). " + f"canonical repository root: {root}; workspace: {workspace}." + ) + + def assess_author_mutation_worktree( *, workspace_path: str, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index beb6084..5fd8c48 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -190,14 +190,22 @@ def _ensure_process_start_porcelain() -> str: def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: """Resolve the workspace root inspected by pre-flight guards.""" - path = (worktree_path or "").strip() - if not path: - path = (os.environ.get(ACTIVE_WORKTREE_ENV) or "").strip() - if not path: - path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip() - if not path: - path = PROJECT_ROOT - return os.path.realpath(os.path.abspath(path)) + return author_mutation_worktree.resolve_mutation_workspace( + worktree_path, + PROJECT_ROOT, + active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), + author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), + ) + + +def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict: + """Canonical workspace + repository root for runtime_context and guards (#460).""" + return author_mutation_worktree.resolve_author_mutation_context( + worktree_path, + PROJECT_ROOT, + active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), + author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), + ) def _get_git_root(path: str) -> str | None: @@ -267,21 +275,31 @@ def _format_preflight_files(files: list[str]) -> str: def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict: - workspace = _resolve_preflight_workspace_path(worktree_path) + ctx = _resolve_author_mutation_context(worktree_path) + workspace = ctx["workspace_path"] inspected_root = _get_git_root(workspace) - control_root = os.path.realpath(PROJECT_ROOT) + process_root = ctx["process_project_root"] + canonical_root = ctx["canonical_repo_root"] active_root = os.path.realpath(inspected_root or workspace) - if active_root == control_root: + if active_root == canonical_root: dirty_scope = "control checkout" else: dirty_scope = "active task workspace" - return { - "mcp_server_process_root": control_root, + details = { + "mcp_server_process_root": process_root, + "canonical_repository_root": canonical_root, "active_task_workspace_root": active_root, "inspected_git_root": inspected_root, "dirty_files": list(dirty_files), "dirty_scope": dirty_scope, + "workspace_roots_aligned": ctx["roots_aligned"], } + if not ctx["roots_aligned"]: + details["workspace_root_mismatch"] = ( + "runtime_context and mutation guard use canonical repository root " + f"'{canonical_root}' instead of MCP process root '{process_root}'" + ) + return details def _format_preflight_workspace_details(details: dict) -> str: @@ -402,16 +420,12 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> """#274: author mutations must run from a branches/ session worktree.""" if _preflight_resolved_role == "reviewer": return - workspace = author_mutation_worktree.resolve_mutation_workspace( - worktree_path, - PROJECT_ROOT, - active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), - author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), - ) + ctx = _resolve_author_mutation_context(worktree_path) + workspace = ctx["workspace_path"] git_state = issue_lock_worktree.read_worktree_git_state(workspace) assessment = author_mutation_worktree.assess_author_mutation_worktree( workspace_path=workspace, - project_root=PROJECT_ROOT, + project_root=ctx["canonical_repo_root"], current_branch=git_state.get("current_branch"), ) if assessment["block"]: @@ -440,43 +454,23 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" ) - workspace = author_mutation_worktree.resolve_mutation_workspace( - worktree_path, - PROJECT_ROOT, - active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), - author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), - ) + ctx = _resolve_author_mutation_context(worktree_path) + workspace = ctx["workspace_path"] + canonical_root = ctx["canonical_repo_root"] + process_root = ctx["process_project_root"] real_workspace = os.path.realpath(workspace) - real_root = os.path.realpath(PROJECT_ROOT) - if real_workspace != real_root: + if real_workspace != process_root: if not _preflight_in_test_mode(): - if not os.path.exists(real_workspace): + membership = author_mutation_worktree.assess_workspace_repo_membership( + workspace_path=workspace, + canonical_repo_root=canonical_root, + ) + if membership["block"]: raise RuntimeError( - f"Branches-only mutation guard (#274): worktree path '{workspace}' does not exist (fail closed)" - ) - if not os.path.isdir(real_workspace): - raise RuntimeError( - f"Branches-only mutation guard (#274): worktree path '{workspace}' is not a directory (fail closed)" - ) - try: - res = subprocess.run( - ["git", "-C", real_workspace, "rev-parse", "--git-common-dir"], - capture_output=True, - text=True, - check=True, - ) - common_dir = os.path.realpath(res.stdout.strip()) - expected_dir = os.path.realpath(os.path.join(real_root, ".git")) - if common_dir != expected_dir: - raise RuntimeError( - f"Branches-only mutation guard (#274): worktree '{workspace}' does not belong to the target repository '{PROJECT_ROOT}' (fail closed)" + author_mutation_worktree.format_workspace_repo_membership_error( + membership ) - except Exception as e: - if isinstance(e, RuntimeError): - raise e - raise RuntimeError( - f"Branches-only mutation guard (#274): worktree '{workspace}' is not a valid git repository (fail closed)" ) dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace))) diff --git a/tests/test_create_issue_workspace_guard.py b/tests/test_create_issue_workspace_guard.py index 2cd9f41..8d1d1f6 100644 --- a/tests/test_create_issue_workspace_guard.py +++ b/tests/test_create_issue_workspace_guard.py @@ -106,20 +106,32 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase): @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"}) @patch("os.path.exists", return_value=True) @patch("os.path.isdir", return_value=True) + @patch("author_mutation_worktree.subprocess.run") @patch("subprocess.run") - def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth): - # Mock subprocess.run for git --git-common-dir to return a different path - mock_res = MagicMock() - mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n" - mock_run.return_value = mock_res - + def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_amw_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth): wrong_repo_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1") + def _subprocess_side_effect(cmd, *args, **kwargs): + mock_res = MagicMock(returncode=0) + if "--git-common-dir" in cmd: + cwd = cmd[cmd.index("-C") + 1] if "-C" in cmd else "" + if cwd == wrong_repo_path: + mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n" + else: + mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n" + else: + mock_res.stdout = "" + return mock_res + + mock_run.side_effect = _subprocess_side_effect + mock_amw_run.side_effect = _subprocess_side_effect + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): - with self.assertRaises(RuntimeError) as ctx: - srv.gitea_create_issue( - title="Test issue", body="body", worktree_path=wrong_repo_path - ) + with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""): + with self.assertRaises(RuntimeError) as ctx: + srv.gitea_create_issue( + title="Test issue", body="body", worktree_path=wrong_repo_path + ) self.assertIn("does not belong to the target repository", str(ctx.exception)) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) diff --git a/tests/test_workspace_guard_alignment.py b/tests/test_workspace_guard_alignment.py new file mode 100644 index 0000000..ddcb67f --- /dev/null +++ b/tests/test_workspace_guard_alignment.py @@ -0,0 +1,153 @@ +"""Tests for runtime_context / mutation-guard workspace alignment (#460).""" + +from __future__ import annotations + +import os +import sys +import unittest +from pathlib import Path +from unittest import mock +from unittest.mock import MagicMock + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import author_mutation_worktree as amw # noqa: E402 +import gitea_mcp_server as srv # noqa: E402 + +CONTROL_ROOT = str(Path(__file__).resolve().parents[3]) +BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1]) +MCP_PROCESS_ROOT = BRANCHES_WORKTREE + + +class TestCanonicalRepoRoot(unittest.TestCase): + @mock.patch("subprocess.run") + def test_resolves_main_repo_from_branches_worktree(self, mock_run): + mock_run.return_value = MagicMock( + returncode=0, + stdout=f"{CONTROL_ROOT}/.git\n", + ) + root = amw.resolve_canonical_repo_root(BRANCHES_WORKTREE, MCP_PROCESS_ROOT) + self.assertEqual(root, CONTROL_ROOT) + + def test_falls_back_when_git_unavailable(self): + root = amw.resolve_canonical_repo_root("/missing/path", MCP_PROCESS_ROOT) + self.assertEqual(root, os.path.realpath(MCP_PROCESS_ROOT)) + + +class TestWorkspaceRepoMembership(unittest.TestCase): + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + @mock.patch("subprocess.run") + def test_valid_branches_worktree_accepted(self, mock_run, *_exists): + mock_run.return_value = MagicMock( + returncode=0, + stdout=f"{CONTROL_ROOT}/.git\n", + ) + result = amw.assess_workspace_repo_membership( + workspace_path=BRANCHES_WORKTREE, + canonical_repo_root=CONTROL_ROOT, + ) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + @mock.patch("subprocess.run") + def test_wrong_repo_rejected(self, mock_run, *_exists): + mock_run.return_value = MagicMock( + returncode=0, + stdout="/other/repo/.git\n", + ) + result = amw.assess_workspace_repo_membership( + workspace_path=BRANCHES_WORKTREE, + canonical_repo_root=CONTROL_ROOT, + ) + self.assertTrue(result["block"]) + self.assertIn("does not belong", result["reasons"][0]) + + @mock.patch("os.path.exists", return_value=False) + def test_missing_worktree_rejected(self, *_exists): + result = amw.assess_workspace_repo_membership( + workspace_path=f"{CONTROL_ROOT}/branches/missing-worktree", + canonical_repo_root=CONTROL_ROOT, + ) + self.assertTrue(result["block"]) + self.assertIn("does not exist", result["reasons"][0]) + + +class TestRuntimeContextGuardAlignment(unittest.TestCase): + def setUp(self): + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_resolved_role = "author" + self._orig_in_test = srv._preflight_in_test_mode + srv._preflight_in_test_mode = lambda: False + self._env_patch = mock.patch.dict( + os.environ, + {}, + clear=False, + ) + self._env_patch.start() + os.environ.pop("GITEA_ACTIVE_WORKTREE", None) + os.environ.pop("GITEA_AUTHOR_WORKTREE", None) + + def tearDown(self): + srv._preflight_in_test_mode = self._orig_in_test + self._env_patch.stop() + + def test_runtime_context_and_guard_share_resolved_workspace(self): + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + ctx = srv._resolve_author_mutation_context(BRANCHES_WORKTREE) + status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE) + self.assertEqual(ctx["workspace_path"], os.path.realpath(BRANCHES_WORKTREE)) + self.assertEqual(ctx["canonical_repo_root"], CONTROL_ROOT) + self.assertFalse(ctx["roots_aligned"]) + self.assertEqual( + status["preflight_workspace"]["active_task_workspace_root"], + os.path.realpath(BRANCHES_WORKTREE), + ) + self.assertEqual( + status["preflight_workspace"]["canonical_repository_root"], + CONTROL_ROOT, + ) + self.assertIn("workspace_root_mismatch", status["preflight_workspace"]) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_declared_branches_worktree_passes_when_mcp_root_differs( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock( + returncode=0, + stdout=f"{CONTROL_ROOT}/.git\n", + ) + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False): + srv.verify_preflight_purity(worktree_path=BRANCHES_WORKTREE) + + def test_stable_checkout_still_rejected(self): + with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT): + with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity() + self.assertIn("stable control checkout", str(ctx.exception)) + + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + @mock.patch("subprocess.run") + def test_non_branches_worktree_rejected(self, mock_run, *_exists): + outside = "/tmp/outside-repo-checkout" + mock_run.return_value = MagicMock( + returncode=0, + stdout=f"{CONTROL_ROOT}/.git\n", + ) + with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT): + with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity(worktree_path=outside) + self.assertIn("not under", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file