From 6779d903f2b7f25b1da04eb134caae292fb6dd27 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:44:58 -0400 Subject: [PATCH] feat: enforce branches-only author mutation worktree guard (Closes #274) Add author_mutation_worktree assessment and integrate it into MCP preflight so author mutations fail closed outside branches/ session worktrees, with tests for control-checkout and branches-path cases. Co-Authored-By: Claude Opus 4.8 (1M context) --- author_mutation_worktree.py | 103 +++++++++++++++++++++++++ gitea_mcp_server.py | 72 +++++++++++------ tests/test_author_mutation_worktree.py | 96 +++++++++++++++++++++++ 3 files changed, 247 insertions(+), 24 deletions(-) create mode 100644 author_mutation_worktree.py create mode 100644 tests/test_author_mutation_worktree.py diff --git a/author_mutation_worktree.py b/author_mutation_worktree.py new file mode 100644 index 0000000..590cea7 --- /dev/null +++ b/author_mutation_worktree.py @@ -0,0 +1,103 @@ +"""Branches-only author mutation worktree guard (#274). + +Author/coder mutations must run from a session-owned worktree under the +project's ``branches/`` directory, never from the stable control checkout. +""" + +from __future__ import annotations + +import os + +BASE_BRANCHES = frozenset({"master", "main", "dev"}) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").rstrip("/") + + +def is_path_under_branches(path: str, project_root: str | None = None) -> bool: + """True when *path* resolves inside ``/branches/``.""" + normalized = _normalize_path(path) + if not normalized: + return False + if "/branches/" in f"{normalized}/": + return True + if normalized.endswith("/branches"): + return True + if project_root: + root = _normalize_path(os.path.realpath(project_root)) + real = _normalize_path(os.path.realpath(path)) + if real.startswith(f"{root}/"): + rel = real[len(root) + 1 :] + return rel == "branches" or rel.startswith("branches/") + return False + + +def resolve_mutation_workspace( + worktree_path: str | None, + project_root: str, + *, + active_worktree_env: str | None = None, + author_worktree_env: str | None = None, +) -> str: + """Resolve the workspace path inspected before author mutations.""" + for candidate in (worktree_path, active_worktree_env, author_worktree_env): + text = (candidate or "").strip() + if text: + return os.path.realpath(os.path.abspath(text)) + return os.path.realpath(project_root) + + +def assess_author_mutation_worktree( + *, + workspace_path: str, + project_root: str, + current_branch: str | None = None, + base_branches: frozenset[str] | None = None, +) -> dict: + """Fail closed when author mutations are not rooted under ``branches/``.""" + bases = base_branches or BASE_BRANCHES + reasons: list[str] = [] + root = os.path.realpath(project_root) + workspace = os.path.realpath(workspace_path) + branch = (current_branch or "").strip() + + if workspace == root: + reasons.append( + "author mutation blocked: workspace is the stable control checkout; " + "create or switch to a session-owned worktree under branches/" + ) + elif not is_path_under_branches(workspace, root): + reasons.append( + f"author mutation blocked: workspace '{workspace}' is not under " + f"'{root}/branches/'; create a branches/ worktree first" + ) + + if workspace == root and branch and branch not in bases: + reasons.append( + f"control checkout drift: branch '{branch}' is not a stable base " + f"branch ({'/'.join(sorted(bases))})" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "project_root": root, + "workspace_path": workspace, + "under_branches": is_path_under_branches(workspace, root), + "current_branch": branch or None, + } + + +def format_author_mutation_worktree_error(assessment: dict) -> str: + """Single RuntimeError message for MCP preflight gates.""" + workspace = assessment.get("workspace_path") or "(unknown)" + root = assessment.get("project_root") or "(unknown)" + reasons = "; ".join(assessment.get("reasons") or ["unknown branches-only violation"]) + return ( + f"Branches-only mutation guard (#274): {reasons}. " + f"project root: {root}; workspace: {workspace}. " + "Create a session-owned worktree under branches/ before mutating." + ) \ No newline at end of file diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index b5415eb..5209f27 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -397,6 +397,28 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None): _preflight_resolved_role = resolved_role +def _enforce_branches_only_author_mutation(worktree_path: str | None = 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), + ) + 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, + current_branch=git_state.get("current_branch"), + ) + if assessment["block"]: + raise RuntimeError( + author_mutation_worktree.format_author_mutation_worktree_error(assessment) + ) + + def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None): """Verify that identity and capability were verified prior to session edits.""" global _preflight_reviewer_violation_files @@ -426,32 +448,33 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None "file edits before mutation (fail closed). " f"{_format_preflight_workspace_details(details)}" ) - return - - if _preflight_whoami_violation: - raise RuntimeError( - "Pre-flight order violation: Workspace file edits occurred before " - f"gitea_whoami verification (fail closed). Offending files: " - f"{_format_preflight_files(_preflight_whoami_violation_files)}" - ) - if _preflight_capability_violation: - raise RuntimeError( - "Pre-flight order violation: Workspace file edits occurred before " - f"gitea_resolve_task_capability verification (fail closed). Offending files: " - f"{_format_preflight_files(_preflight_capability_violation_files)}" - ) - - if _preflight_resolved_role == "reviewer": - current = _get_workspace_porcelain() - baseline = _preflight_capability_baseline_porcelain or "" - reviewer_delta = _new_tracked_changes_since(baseline, current) - _preflight_reviewer_violation_files = reviewer_delta - if reviewer_delta: + else: + if _preflight_whoami_violation: raise RuntimeError( - "Reviewer role violation: Reviewer profile is forbidden from modifying " - "tracked workspace files (fail closed). Offending files: " - f"{_format_preflight_files(reviewer_delta)}" + "Pre-flight order violation: Workspace file edits occurred before " + f"gitea_whoami verification (fail closed). Offending files: " + f"{_format_preflight_files(_preflight_whoami_violation_files)}" ) + if _preflight_capability_violation: + raise RuntimeError( + "Pre-flight order violation: Workspace file edits occurred before " + f"gitea_resolve_task_capability verification (fail closed). Offending files: " + f"{_format_preflight_files(_preflight_capability_violation_files)}" + ) + + if _preflight_resolved_role == "reviewer": + current = _get_workspace_porcelain() + baseline = _preflight_capability_baseline_porcelain or "" + reviewer_delta = _new_tracked_changes_since(baseline, current) + _preflight_reviewer_violation_files = reviewer_delta + if reviewer_delta: + raise RuntimeError( + "Reviewer role violation: Reviewer profile is forbidden from modifying " + "tracked workspace files (fail closed). Offending files: " + f"{_format_preflight_files(reviewer_delta)}" + ) + + _enforce_branches_only_author_mutation(worktree_path) from mcp.server.fastmcp import FastMCP # noqa: E402 @@ -476,6 +499,7 @@ import task_capability_map # noqa: E402 import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 +import author_mutation_worktree # noqa: E402 import merged_cleanup_reconcile # noqa: E402 diff --git a/tests/test_author_mutation_worktree.py b/tests/test_author_mutation_worktree.py new file mode 100644 index 0000000..ce04612 --- /dev/null +++ b/tests/test_author_mutation_worktree.py @@ -0,0 +1,96 @@ +"""Tests for branches-only author mutation worktree guard (#274).""" + +from __future__ import annotations + +import sys +import unittest +from pathlib import Path +from unittest import mock + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import author_mutation_worktree as amw # noqa: E402 + + +class TestPathUnderBranches(unittest.TestCase): + ROOT = "/repo/Gitea-Tools" + + def test_branches_subdirectory_passes(self): + self.assertTrue( + amw.is_path_under_branches(f"{self.ROOT}/branches/issue-274", self.ROOT) + ) + + def test_control_checkout_fails(self): + self.assertFalse(amw.is_path_under_branches(self.ROOT, self.ROOT)) + + def test_sibling_directory_fails(self): + self.assertFalse( + amw.is_path_under_branches("/repo/other-checkout", self.ROOT) + ) + + +class TestAssessAuthorMutationWorktree(unittest.TestCase): + ROOT = "/repo/Gitea-Tools" + + def test_branches_worktree_passes(self): + result = amw.assess_author_mutation_worktree( + workspace_path=f"{self.ROOT}/branches/issue-274", + project_root=self.ROOT, + current_branch="feat/issue-274-branches-only-worktrees", + ) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_control_checkout_blocks(self): + result = amw.assess_author_mutation_worktree( + workspace_path=self.ROOT, + project_root=self.ROOT, + current_branch="master", + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertIn("control checkout", result["reasons"][0]) + + def test_drifted_control_branch_blocks(self): + result = amw.assess_author_mutation_worktree( + workspace_path=self.ROOT, + project_root=self.ROOT, + current_branch="feat/some-task", + ) + self.assertTrue(result["block"]) + self.assertTrue(any("drift" in r for r in result["reasons"])) + + +class TestPreflightIntegration(unittest.TestCase): + def test_verify_preflight_blocks_control_checkout_with_test_porcelain(self): + import mcp_server + + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + mcp_server._preflight_resolved_role = "author" + with mock.patch.dict( + "os.environ", + {"GITEA_TEST_PORCELAIN": ""}, + clear=False, + ): + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity() + self.assertIn("Branches-only mutation guard", str(ctx.exception)) + + def test_verify_preflight_allows_branches_worktree(self): + import mcp_server + + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + mcp_server._preflight_resolved_role = "author" + worktree = "/repo/Gitea-Tools/branches/issue-274" + with mock.patch.dict( + "os.environ", + {"GITEA_TEST_PORCELAIN": ""}, + clear=False, + ): + mcp_server.verify_preflight_purity(worktree_path=worktree) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file