From 9d2c652ae897afee2f6de139f9b85ab2569b5a85 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 19 Jul 2026 12:59:08 -0400 Subject: [PATCH] fix(mcp): require proven base equivalence for the create-issue bootstrap Remediates review #473 (REQUEST_CHANGES) on PR #759 / issue #757 AC3/AC4. The bootstrap compared SHAs only inside: if remote_tip and local_tip and remote_tip != local_tip: so agreement was assumed whenever either tip was unknown. A missing local HEAD, an unresolvable live master, or a resolver exception silently granted the control-checkout exemption instead of blocking it. An unknown tip is missing evidence, not proof of equivalence. Changes: * assess_create_issue_bootstrap now blocks unless BOTH tips are known and equal, with distinct reasons for missing local HEAD, unknown live master, and resolver failure. Adds a remote_master_sha_error parameter so a failed resolution is reported as missing evidence rather than absence of a constraint. * Both normalized SHAs and a derived base_tips_verified flag are recorded on the assessment. normalize_sha() treats only case and surrounding whitespace as equivalent spellings of a commit. * bootstrap_permits_control_checkout re-derives the comparison from the recorded tips instead of trusting base_tips_verified, so a hand-built or truncated assessment cannot assert agreement it never proved. * _create_issue_bootstrap_assessment captures the resolver exception and forwards it, replacing the silent except -> None. One shared assessment still serves both the #274 and #604 guards, and _BOOTSTRAP_UNSET is preserved. No MCP tool signature gains a bootstrap argument (AC6). No issue or PR number is special-cased (AC10). Tests: 16 new assertions across two classes covering missing local, missing remote, both missing, empty/whitespace tips, mismatch, resolver exception at the assessment site, and resolver exception through the real verify_preflight_purity path; plus predicate rejection of stripped tips, a forged base_tips_verified flag, and a missing flag. All 16 fail against the sources at adc61255 and pass after this change. The valid non-create_issue lock_issue test is retained unchanged. Validation: focused #757 suite 51 passed (31 subtests); affected guard and preflight suites 217 passed (67 subtests); full suite 3637 passed, 2 failed, 6 skipped, 431 subtests. Both failures (test_issue_702 F1 worktree recovery; test_reconciler_supersession_close org/repo forwarding) are the documented pre-existing baseline failures on bde5c5fb and are not caused by this change. Refs #757 Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_015KRvvrFtaM5FEvQ6LvDJMH --- create_issue_bootstrap.py | 66 +++++- gitea_mcp_server.py | 8 +- ...est_issue_757_bootstrap_guard_agreement.py | 196 ++++++++++++++++++ 3 files changed, 267 insertions(+), 3 deletions(-) diff --git a/create_issue_bootstrap.py b/create_issue_bootstrap.py index 0c188f1..80595ba 100644 --- a/create_issue_bootstrap.py +++ b/create_issue_bootstrap.py @@ -52,6 +52,18 @@ def is_create_issue_task(task: str | None) -> bool: return (task or "").strip() in CREATE_ISSUE_TASKS +def normalize_sha(value: str | None) -> str | None: + """Normalize a Git object id for comparison, or ``None`` when unknown. + + Whitespace and case are the only permitted variation between two spellings + of the same commit; anything else is a different commit. Empty and + whitespace-only values normalize to ``None`` so an unknown tip can never + compare equal to another unknown tip. + """ + normalized = (value or "").strip().lower() + return normalized or None + + def assess_create_issue_bootstrap( *, workspace_path: str, @@ -60,6 +72,7 @@ def assess_create_issue_bootstrap( head_sha: str | None = None, porcelain_status: str = "", remote_master_sha: str | None = None, + remote_master_sha_error: str | None = None, task: str | None = None, ) -> dict[str, Any]: """Assess whether create_issue may proceed from the control checkout. @@ -142,8 +155,32 @@ def assess_create_issue_bootstrap( f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})" ) - remote_tip = (remote_master_sha or "").strip() or None - local_tip = (head_sha or "").strip() or None + # #757 AC3/AC4: base-equivalence must be *proven*, never assumed. An + # unknown tip on either side is not evidence of agreement, so a missing + # local HEAD, an unresolvable live master, or a resolver failure all block. + remote_tip = normalize_sha(remote_master_sha) + local_tip = normalize_sha(head_sha) + resolver_error = (remote_master_sha_error or "").strip() or None + + if not local_tip: + reasons.append( + "create_issue bootstrap blocked: control checkout HEAD SHA is " + "unknown; base equivalence to live master cannot be proven " + "(fail closed)" + ) + + if resolver_error: + reasons.append( + "create_issue bootstrap blocked: live master tip could not be " + f"resolved ({resolver_error}); base equivalence cannot be proven " + "(fail closed)" + ) + elif not remote_tip: + reasons.append( + "create_issue bootstrap blocked: live master tip is unknown; base " + "equivalence to live master cannot be proven (fail closed)" + ) + if remote_tip and local_tip and remote_tip != local_tip: reasons.append( "create_issue bootstrap blocked: control checkout HEAD does not match " @@ -162,6 +199,8 @@ def assess_create_issue_bootstrap( dirty=dirty, under_branches=False, exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP, + local_head_sha=local_tip, + remote_master_sha=remote_tip, ) return _result( @@ -176,6 +215,8 @@ def assess_create_issue_bootstrap( under_branches=False, exact_next_action=EXACT_NEXT_ACTION_POST_CREATE, bootstrap_path="clean_canonical_control_checkout", + local_head_sha=local_tip, + remote_master_sha=remote_tip, ) @@ -235,6 +276,17 @@ def bootstrap_permits_control_checkout( if assessment.get("under_branches") is not False: return False + # Base-equivalence proof (#757 AC3/AC4): both tips must be recorded, + # nonempty, and equal. Re-derived here rather than trusted from the + # assessment's own flag, so a hand-built or truncated assessment cannot + # assert agreement it never proved. + if assessment.get("base_tips_verified") is not True: + return False + local_tip = normalize_sha(assessment.get("local_head_sha")) + remote_tip = normalize_sha(assessment.get("remote_master_sha")) + if not local_tip or not remote_tip or local_tip != remote_tip: + return False + # Binding proof: the assessment must describe *this* workspace and root, # and that workspace must be exactly the canonical control checkout. root = os.path.realpath(canonical_repo_root or "") @@ -279,7 +331,14 @@ def _result( under_branches: bool, exact_next_action: str | None = None, bootstrap_path: str | None = None, + local_head_sha: str | None = None, + remote_master_sha: str | None = None, ) -> dict[str, Any]: + # #757 AC3/AC4: equality is recorded only when BOTH tips are known, so a + # consumer can never read agreement out of two missing values. + base_tips_verified = bool( + local_head_sha and remote_master_sha and local_head_sha == remote_master_sha + ) return { "not_applicable": not_applicable, "allowed": allowed, @@ -294,4 +353,7 @@ def _result( "exact_next_action": exact_next_action, "bootstrap_path": bootstrap_path, "task_scope": "create_issue_only", + "local_head_sha": local_head_sha, + "remote_master_sha": remote_master_sha, + "base_tips_verified": base_tips_verified, } diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index d495ab6..e5d2a6d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -845,12 +845,17 @@ def _create_issue_bootstrap_assessment( ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] git_state = issue_lock_worktree.read_worktree_git_state(workspace) + # #757 AC3/AC4: a resolver failure is not "no constraint" — it is missing + # evidence, and must reach the assessor as such so the bootstrap fails + # closed instead of proceeding without base-equivalence proof. + remote_master_sha_error: str | None = None try: remote_master_sha = root_checkout_guard.resolve_remote_master_sha( ctx["canonical_repo_root"] ) - except Exception: + except Exception as exc: remote_master_sha = None + remote_master_sha_error = f"{type(exc).__name__}: {exc}".strip() or "resolver failed" return _cib.assess_create_issue_bootstrap( workspace_path=workspace, canonical_repo_root=ctx["canonical_repo_root"], @@ -858,6 +863,7 @@ def _create_issue_bootstrap_assessment( head_sha=git_state.get("head_sha"), porcelain_status=git_state.get("porcelain_status") or "", remote_master_sha=remote_master_sha, + remote_master_sha_error=remote_master_sha_error, task=task, ) diff --git a/tests/test_issue_757_bootstrap_guard_agreement.py b/tests/test_issue_757_bootstrap_guard_agreement.py index 9b8b05e..f41af74 100644 --- a/tests/test_issue_757_bootstrap_guard_agreement.py +++ b/tests/test_issue_757_bootstrap_guard_agreement.py @@ -609,5 +609,201 @@ class TestNativeCreateIssueEndToEnd(unittest.TestCase): self.assertIn("control checkout", str(ctx.exception)) +class TestBaseEquivalenceProofRequired(unittest.TestCase): + """#757 AC3/AC4: an unknown tip is not evidence of agreement. + + The original fix compared SHAs only inside + ``if remote_tip and local_tip and remote_tip != local_tip``, so a missing + local HEAD, an unresolvable live master, or a resolver failure all fell + through and *granted* the bootstrap exemption. Base equivalence must be + proven, and anything less must fail closed. + """ + + def _permits(self, assessment): + return cib.bootstrap_permits_control_checkout( + assessment, + task="create_issue", + workspace_path=CONTROL_CHECKOUT_ROOT, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + ) + + def _assert_fails_closed(self, assessment, *, expect_reason): + self.assertTrue(assessment["block"], "assessor must block") + self.assertFalse(assessment["allowed"]) + self.assertFalse(assessment["proven"]) + self.assertFalse(assessment["base_tips_verified"]) + self.assertFalse(self._permits(assessment), "predicate must refuse") + joined = " ".join(assessment["reasons"]).lower() + self.assertIn(expect_reason, joined) + + def test_missing_local_head_fails_closed(self): + self._assert_fails_closed( + proven_bootstrap(head=None), + expect_reason="control checkout head sha is unknown", + ) + + def test_missing_remote_master_fails_closed(self): + self._assert_fails_closed( + proven_bootstrap(remote_sha=None), + expect_reason="live master tip is unknown", + ) + + def test_both_tips_missing_fails_closed(self): + assessment = proven_bootstrap(head=None, remote_sha=None) + self._assert_fails_closed( + assessment, expect_reason="control checkout head sha is unknown" + ) + self.assertIn("live master tip is unknown", " ".join(assessment["reasons"])) + + def test_empty_and_whitespace_tips_fail_closed(self): + for head, remote in (("", MASTER_SHA), (MASTER_SHA, ""), (" ", " ")): + with self.subTest(head=repr(head), remote=repr(remote)): + assessment = proven_bootstrap(head=head, remote_sha=remote) + self.assertTrue(assessment["block"]) + self.assertFalse(self._permits(assessment)) + + def test_mismatched_tips_fail_closed(self): + self._assert_fails_closed( + proven_bootstrap(head=STALE_SHA), + expect_reason="does not match", + ) + + def test_resolver_failure_fails_closed(self): + assessment = cib.assess_create_issue_bootstrap( + workspace_path=CONTROL_CHECKOUT_ROOT, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + current_branch="master", + head_sha=MASTER_SHA, + porcelain_status="", + remote_master_sha=None, + remote_master_sha_error="TimeoutError: remote unreachable", + task="create_issue", + ) + self._assert_fails_closed( + assessment, expect_reason="could not be resolved" + ) + # The operator must be able to see *why*, not just that it blocked. + self.assertIn("remote unreachable", " ".join(assessment["reasons"])) + + def test_proven_bootstrap_records_both_normalized_shas(self): + assessment = proven_bootstrap() + self.assertEqual(assessment["local_head_sha"], MASTER_SHA) + self.assertEqual(assessment["remote_master_sha"], MASTER_SHA) + self.assertTrue(assessment["base_tips_verified"]) + self.assertTrue(self._permits(assessment)) + + def test_tips_are_normalized_before_comparison(self): + """Case and surrounding whitespace are not a different commit.""" + assessment = proven_bootstrap( + head=f" {MASTER_SHA.upper()} ", remote_sha=MASTER_SHA + ) + self.assertTrue(assessment["allowed"]) + self.assertEqual(assessment["local_head_sha"], MASTER_SHA) + self.assertTrue(self._permits(assessment)) + + def test_predicate_rejects_assessment_with_tips_stripped(self): + """A recorded proof that is later removed cannot still permit.""" + for field in ("local_head_sha", "remote_master_sha"): + with self.subTest(field=field): + assessment = dict(proven_bootstrap()) + assessment[field] = None + self.assertFalse(self._permits(assessment)) + + def test_predicate_rejects_forged_verified_flag(self): + """base_tips_verified is re-derived, never trusted on its own.""" + assessment = dict(proven_bootstrap()) + assessment["local_head_sha"] = MASTER_SHA + assessment["remote_master_sha"] = STALE_SHA + assessment["base_tips_verified"] = True + self.assertFalse(self._permits(assessment)) + + def test_predicate_rejects_missing_verified_flag(self): + assessment = dict(proven_bootstrap()) + assessment.pop("base_tips_verified") + self.assertFalse(self._permits(assessment)) + + +class TestServerAssessmentFailsClosedOnResolverError(unittest.TestCase): + """AC3/AC4 at the single server-derived computation site.""" + + def setUp(self): + # verify_preflight_purity short-circuits under pytest; the production + # path only runs with test mode disabled (same setup the #757 e2e uses). + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_resolved_role = "author" + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + self._orig_in_test = srv._preflight_in_test_mode + srv._preflight_in_test_mode = lambda: False + + def tearDown(self): + srv._preflight_in_test_mode = self._orig_in_test + srv._preflight_resolved_task = None + + def _git_state(self): + return { + "current_branch": "master", + "head_sha": MASTER_SHA, + "porcelain_status": "", + } + + def test_resolver_exception_produces_blocking_assessment(self): + """A raising resolver must not become a silent, permissive None.""" + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \ + patch.object( + srv.issue_lock_worktree, + "read_worktree_git_state", + return_value=self._git_state(), + ), \ + patch.object( + srv.root_checkout_guard, + "resolve_remote_master_sha", + side_effect=TimeoutError("remote unreachable"), + ): + assessment = srv._create_issue_bootstrap_assessment("create_issue") + + self.assertIsNotNone(assessment) + self.assertTrue(assessment["block"]) + self.assertFalse(assessment["allowed"]) + self.assertFalse(assessment["base_tips_verified"]) + self.assertIsNone(assessment["remote_master_sha"]) + self.assertIn("could not be resolved", " ".join(assessment["reasons"])) + self.assertFalse( + cib.bootstrap_permits_control_checkout( + assessment, + task="create_issue", + workspace_path=CONTROL_CHECKOUT_ROOT, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + ) + ) + + def test_resolver_exception_blocks_real_preflight(self): + """Production path: verify_preflight_purity must fail closed.""" + srv._preflight_resolved_task = "create_issue" + try: + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \ + patch.object(srv, "_actual_profile_role", return_value="author"), \ + patch.object( + srv, "_effective_workspace_role", return_value="author" + ), \ + patch.object( + srv.issue_lock_worktree, + "read_worktree_git_state", + return_value=self._git_state(), + ), \ + patch.object(srv, "_get_workspace_porcelain", return_value=""), \ + patch.object( + srv.root_checkout_guard, + "resolve_remote_master_sha", + side_effect=TimeoutError("remote unreachable"), + ), \ + patch.object(srv, "_enforce_root_checkout_guard"): + with self.assertRaises(RuntimeError): + srv.verify_preflight_purity(remote="prgs", task="create_issue") + finally: + srv._preflight_resolved_task = None + + if __name__ == "__main__": unittest.main()