diff --git a/mcp_server.py b/mcp_server.py index 0218616..6f196ea 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -156,6 +156,89 @@ PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) +import json + +_preflight_whoami_called = False +_preflight_capability_called = False +_preflight_whoami_violation = False +_preflight_capability_violation = False +_preflight_resolved_role = None + +def record_preflight_check(type_name: str, resolved_role: str | None = None): + """Record a pre-flight check (whoami or capability) and check for workspace edits.""" + global _preflight_whoami_called, _preflight_capability_called + global _preflight_whoami_violation, _preflight_capability_violation + global _preflight_resolved_role + + in_test = "pytest" in sys.modules or "unittest" in sys.modules + if os.environ.get("GITEA_TEST_FORCE_DIRTY"): + is_dirty = True + elif in_test: + is_dirty = False + else: + is_dirty = False + try: + res = subprocess.run( + ["git", "status", "--porcelain"], + capture_output=True, text=True, cwd=PROJECT_ROOT + ) + for line in res.stdout.splitlines(): + if line and not line.startswith("??"): + is_dirty = True + break + except Exception: + pass + + if is_dirty: + if type_name == "whoami" and not _preflight_whoami_called: + _preflight_whoami_violation = True + if type_name == "capability" and not _preflight_capability_called: + _preflight_capability_violation = True + + if type_name == "whoami": + _preflight_whoami_called = True + elif type_name == "capability": + _preflight_capability_called = True + if resolved_role: + _preflight_resolved_role = resolved_role + +def verify_preflight_purity(remote: str | None = None): + """Verify that identity and capability were verified prior to edits, and that reviewers made no edits.""" + in_test = "pytest" in sys.modules or "unittest" in sys.modules + if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"): + return + + if not _preflight_whoami_called: + raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)") + if not _preflight_capability_called: + raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)") + + if _preflight_whoami_violation: + raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)") + if _preflight_capability_violation: + raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)") + + if os.environ.get("GITEA_TEST_FORCE_DIRTY"): + is_dirty = True + elif in_test: + is_dirty = False + else: + is_dirty = False + try: + res = subprocess.run( + ["git", "status", "--porcelain"], + capture_output=True, text=True, cwd=PROJECT_ROOT + ) + for line in res.stdout.splitlines(): + if line and not line.startswith("??"): + is_dirty = True + break + except Exception: + pass + + if _preflight_resolved_role == "reviewer" and is_dirty: + raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)") + from mcp.server.fastmcp import FastMCP # noqa: E402 from gitea_auth import ( # noqa: E402 @@ -465,6 +548,7 @@ def gitea_create_issue( "number": None, "reasons": block_reasons, } + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) @@ -531,6 +615,7 @@ def gitea_create_pr( Returns: dict with 'number' of the created PR ('url' only with the reveal opt-in). """ + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/pulls" @@ -1141,6 +1226,7 @@ def gitea_submit_pr_review( authenticated user, profile name, PR author, PR number, head SHA checked, and the reasons/gates passed or blocked. Never secrets. """ + verify_preflight_purity(remote) action = (action or "").strip().lower() result = { "requested_action": action, @@ -1294,6 +1380,8 @@ def gitea_edit_pr( if not payload: raise ValueError("At least one field to edit (title, body, state, base) must be provided.") + verify_preflight_purity(remote) + # PR closure is a first-class capability, distinct from retitling or # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close # never touches the network. @@ -1411,6 +1499,7 @@ def gitea_commit_files( Returns: dict with success status and commit/branch information. """ + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/contents" @@ -1505,6 +1594,7 @@ def gitea_merge_pr( reasons/gates passed or blocked, and merge result / merge commit if available. Never secrets. """ + verify_preflight_purity(remote) do = (do or "").strip().lower() result = { "performed": False, @@ -1884,6 +1974,7 @@ def gitea_delete_branch( Returns: dict with 'success' and 'message'. """ + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) import urllib.parse @@ -1915,6 +2006,7 @@ def gitea_close_issue( Returns: dict with 'success' boolean and 'message'. """ + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/issues/{issue_number}" @@ -2213,6 +2305,7 @@ def gitea_create_issue_comment( (permission blocks also carry a structured 'permission_report', #142). """ + verify_preflight_purity(remote) gate_reasons = _profile_operation_gate("gitea.issue.comment") reasons = list(gate_reasons) if not (body or "").strip(): @@ -2781,6 +2874,7 @@ def gitea_whoami( 'email', 'server', 'remote', and 'profile' (safe runtime profile metadata: profile_name + allowed_operations; never the token). """ + record_preflight_check("whoami") if remote not in REMOTES: raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}") h = host or REMOTES[remote]["host"] @@ -3300,6 +3394,7 @@ def gitea_mark_issue( if action not in ("start", "done"): raise ValueError(f"action must be 'start' or 'done', got '{action}'") + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) @@ -3382,6 +3477,7 @@ def gitea_create_label( Returns: dict containing the created label details. """ + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) @@ -3421,6 +3517,7 @@ def gitea_set_issue_labels( Returns: list of all labels currently applied to the issue. """ + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) @@ -3691,6 +3788,8 @@ def gitea_resolve_task_capability( required_permission = TASK_MAP[task]["permission"] required_role = TASK_MAP[task]["role"] + record_preflight_check("capability", required_role) + profile = get_profile() config = gitea_config.load_config() diff --git a/review_proofs.py b/review_proofs.py index bd4b817..0af3110 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -749,6 +749,7 @@ HANDOFF_BASE_FIELDS = ( ("Files changed", ("files changed", "changed", "files")), ("Validation", ("validation",)), ("Mutations", ("mutations",)), + ("Workspace mutations", ("workspace mutations",)), ("Current status", ("current status", "status")), ("Blockers", ("blockers",)), ("Next", ("next",)), @@ -799,7 +800,7 @@ def _handoff_section_lines(report_text): return lines[start:] -def assess_controller_handoff(report_text, role=None): +def assess_controller_handoff(report_text, role=None, local_edits=False): """Issue #182: final reports without a Controller Handoff downgrade. Verdicts: @@ -824,10 +825,14 @@ def assess_controller_handoff(report_text, role=None): } labels = [] + fields_dict = {} for line in section: stripped = line.strip().lstrip("-*").strip() if ":" in stripped: - labels.append(stripped.split(":", 1)[0].strip().lower()) + k, v = stripped.split(":", 1) + label = k.strip().lower() + labels.append(label) + fields_dict[label] = v.strip() required = list(HANDOFF_BASE_FIELDS) required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ())) @@ -846,6 +851,19 @@ def assess_controller_handoff(report_text, role=None): "reasons": [f"handoff missing required field: {m}" for m in missing], } + + if local_edits: + workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower() + if not workspace_mutations_val or workspace_mutations_val == "none": + return { + "verdict": "incomplete", + "downgraded": True, + "missing_fields": ["Workspace mutations"], + "reasons": [ + "Workspace mutations cannot be 'none' or empty when local edits exist" + ], + } + return { "verdict": "complete", "downgraded": False, diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 5c2806f..14731e4 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2427,3 +2427,87 @@ class TestVerifyMutationAuthority(unittest.TestCase): with patch.dict(os.environ, {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}): mcp_server.verify_mutation_authority("prgs") + + +# --------------------------------------------------------------------------- +# Pre-flight ordering and workspace edit block (#210) +# --------------------------------------------------------------------------- +class TestPreflightVerification(unittest.TestCase): + """Assert that pre-flight verification gates order correctly.""" + + def setUp(self): + # Save real global variables + import mcp_server + self.orig_whoami_called = mcp_server._preflight_whoami_called + self.orig_capability_called = mcp_server._preflight_capability_called + self.orig_whoami_violation = mcp_server._preflight_whoami_violation + self.orig_capability_violation = mcp_server._preflight_capability_violation + self.orig_resolved_role = mcp_server._preflight_resolved_role + + # Reset state for each test + mcp_server._preflight_whoami_called = False + mcp_server._preflight_capability_called = False + mcp_server._preflight_whoami_violation = False + mcp_server._preflight_capability_violation = False + mcp_server._preflight_resolved_role = None + + def tearDown(self): + # Restore real global variables + import mcp_server + mcp_server._preflight_whoami_called = self.orig_whoami_called + mcp_server._preflight_capability_called = self.orig_capability_called + mcp_server._preflight_whoami_violation = self.orig_whoami_violation + mcp_server._preflight_capability_violation = self.orig_capability_violation + mcp_server._preflight_resolved_role = self.orig_resolved_role + if "GITEA_TEST_FORCE_DIRTY" in os.environ: + del os.environ["GITEA_TEST_FORCE_DIRTY"] + + def test_preflight_whoami_violation(self): + import mcp_server + os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" + mcp_server._preflight_capability_called = True + mcp_server.record_preflight_check("whoami") + self.assertTrue(mcp_server._preflight_whoami_violation) + + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity() + self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception)) + + def test_preflight_capability_violation(self): + import mcp_server + os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" + mcp_server._preflight_whoami_called = True + mcp_server.record_preflight_check("capability", resolved_role="author") + self.assertTrue(mcp_server._preflight_capability_violation) + + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity() + self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception)) + + def test_preflight_not_called_fails_closed(self): + import mcp_server + os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity() + self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception)) + + mcp_server._preflight_whoami_called = True + with self.assertRaises(RuntimeError) as ctx2: + mcp_server.verify_preflight_purity() + self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception)) + + def test_preflight_reviewer_mutation_violation(self): + import mcp_server + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + mcp_server._preflight_resolved_role = "reviewer" + + # When dirty, reviewer edits are blocked + os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity() + self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception)) + + # When clean, reviewer is allowed + del os.environ["GITEA_TEST_FORCE_DIRTY"] + mcp_server.verify_preflight_purity() diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 8bd3276..ed01d99 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -773,6 +773,7 @@ class TestControllerHandoff(unittest.TestCase): "- Files changed: review_proofs.py", "- Validation: 700 passed, 6 skipped", "- Mutations: one PR opened", + "- Workspace mutations: none", "- Current status: PR open", "- Blockers: none", "- Next: review PR #999", @@ -859,6 +860,29 @@ class TestControllerHandoff(unittest.TestCase): self.assertIn("assess_controller_handoff", skill) self.assertIn("issue #182", skill) + def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self): + # 1. Workspace mutations: none is rejected when local_edits is True + incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ + "- Selected issue: #196", + "- Claim/comment status: comment-claimed", + "- PR number opened: #203", + "- No review/merge: confirmed", + ]) + res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True) + self.assertEqual(res["verdict"], "incomplete") + self.assertIn("Workspace mutations", res["missing_fields"]) + + # 2. Workspace mutations: edited files is allowed when local_edits is True + complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([ + "- Selected issue: #196", + "- Claim/comment status: comment-claimed", + "- PR number opened: #203", + "- No review/merge: confirmed", + ]) + res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True) + self.assertEqual(res2["verdict"], "complete") + + class TestPRInventoryTrustGate(unittest.TestCase): """Issue #194: unit tests for the PR inventory trust gate."""