From a6c7d1491e8f099700f838d21ec497c9a7be13a8 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 29 Jul 2026 16:02:32 -0400 Subject: [PATCH] fix(runtime): fix identity derivation disarming in canonical root guard (Closes #973) --- canonical_repository_root.py | 78 ++++++++++++------- gitea_mcp_server.py | 10 ++- .../test_cross_repo_canonical_consumption.py | 1 + ...st_issue_973_cross_repo_canonical_roots.py | 62 +++++++++++++++ 4 files changed, 121 insertions(+), 30 deletions(-) diff --git a/canonical_repository_root.py b/canonical_repository_root.py index cea8a11..20d7824 100644 --- a/canonical_repository_root.py +++ b/canonical_repository_root.py @@ -132,6 +132,7 @@ def assess_canonical_repository_root( process_project_root: str, remote: str | None = None, require_binding: bool = False, + mode: str = "validation", ) -> dict: """Validate the canonical repository root binding, failing closed on forgery. @@ -140,14 +141,13 @@ def assess_canonical_repository_root( ``configured`` (whether a cross-repo binding was declared), ``resolved_slug`` and ``source``. - Without a configured binding the single-repo default is preserved: the - canonical root is derived from *process_project_root* and never blocks - (unless *require_binding* explicitly demands one). - - With a configured binding the path must exist, be a git repository, and — - when *expected_slug* is known — carry a matching repository identity. A - mismatched or (when *require_binding*) unprovable identity is a forged or - conflicting binding and fails closed. + *mode*: + - ``"validation"`` (default): an independently trusted expected repository + slug is known (or required). Candidate root's observed identity must match. + Unprovable or missing expected identity fails closed when *require_binding* is True. + - ``"derivation"``: caller is deriving the canonical repository identity. + No expected slug exists yet by design. Derivation succeeds if the configured + root exists, is a git repository, and carries a resolvable git remote identity. """ process_root = os.path.realpath(process_project_root) declared = (configured_value or "").strip() @@ -168,12 +168,28 @@ def assess_canonical_repository_root( ) # Single-repo default: canonical root follows the install checkout. derived = resolve_repo_toplevel(process_root) or process_root + resolved_slug = repository_identity_slug(derived, remote=remote) + reasons: list[str] = [] + if mode == "validation" and expected_slug: + expected = expected_slug.strip() + if resolved_slug and resolved_slug.lower() != expected.lower(): + reasons.append( + f"canonical repository root identity mismatch: '{derived}' resolves " + f"to repository '{resolved_slug}' but expected repository identity " + f"is '{expected}' (forged or conflicting binding, fail closed)" + ) + elif not resolved_slug and require_binding: + reasons.append( + f"canonical repository root '{derived}' has no resolvable git " + f"remote identity to confirm authorization for '{expected}' " + "(fail closed)" + ) return _assessment( - proven=True, - reasons=[], + proven=not reasons, + reasons=reasons, configured=False, canonical_repo_root=derived, - resolved_slug=None, + resolved_slug=resolved_slug, source=None, ) @@ -207,26 +223,34 @@ def assess_canonical_repository_root( resolved_slug = repository_identity_slug(toplevel, remote=remote) reasons: list[str] = [] - expected = (expected_slug or "").strip() or None - if expected: - if resolved_slug and resolved_slug.lower() != expected.lower(): + + if mode == "derivation": + if not resolved_slug: reasons.append( - f"canonical repository root identity mismatch: '{toplevel}' resolves " - f"to repository '{resolved_slug}' but the session is authorized for " - f"'{expected}' (forged or conflicting binding, fail closed)" + f"configured canonical repository root '{toplevel}' has no resolvable " + "git remote identity (fail closed)" ) - elif not resolved_slug and require_binding: + else: + expected = (expected_slug or "").strip() or None + if expected: + if resolved_slug and resolved_slug.lower() != expected.lower(): + reasons.append( + f"canonical repository root identity mismatch: '{toplevel}' resolves " + f"to repository '{resolved_slug}' but the session is authorized for " + f"'{expected}' (forged or conflicting binding, fail closed)" + ) + elif not resolved_slug and require_binding: + reasons.append( + f"canonical repository root '{toplevel}' has no resolvable git " + f"remote identity to confirm authorization for '{expected}' " + "(fail closed)" + ) + elif require_binding: reasons.append( - f"canonical repository root '{toplevel}' has no resolvable git " - f"remote identity to confirm authorization for '{expected}' " - "(fail closed)" + f"canonical repository root '{toplevel}' has configured value " + f"'{configured_value}' but authoritative expected repository identity " + "is unprovable or missing (fail closed)" ) - elif require_binding: - reasons.append( - f"canonical repository root '{toplevel}' has configured value " - f"'{configured_value}' but authoritative expected repository identity " - "is unprovable or missing (fail closed)" - ) return _assessment( proven=not reasons, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index f1b6448..602ba2b 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -526,18 +526,21 @@ def _resolve_expected_repository_slug( org: str | None = None, repo: str | None = None, ) -> str | None: - """Resolve expected repository slug from parameters, session context, or process root. + """Resolve expected repository slug from session context, parameters, or process root. Must derive expected repository identity ONLY from trusted sources independent of the candidate configured canonical root (#973 B8). + + Bound session context takes precedence over request parameters so request-supplied + coordinates cannot replace a bound canonical root (#741 / #973 B9). """ - if org and repo: - return session_ctx.format_repository_slug(org, repo) bound = session_ctx.get_session_context() or {} b_org = bound.get("org") b_repo = bound.get("repository") if b_org and b_repo: return session_ctx.format_repository_slug(b_org, b_repo) + if org and repo: + return session_ctx.format_repository_slug(org, repo) eff_remote = remote or bound.get("remote") or _effective_remote() parsed = remote_repo_guard.parse_org_repo_from_remote_url( _process_root_git_remote_url(eff_remote) @@ -2250,6 +2253,7 @@ def _canonical_repository_slug( process_project_root=PROJECT_ROOT, remote=remote, require_binding=True, + mode="derivation", ) slug = assessment.get("resolved_slug") if assessment.get("block") or not slug: diff --git a/tests/test_cross_repo_canonical_consumption.py b/tests/test_cross_repo_canonical_consumption.py index 2eaee02..c397da0 100644 --- a/tests/test_cross_repo_canonical_consumption.py +++ b/tests/test_cross_repo_canonical_consumption.py @@ -313,6 +313,7 @@ class TestCanonicalRootGuardBinding(_ServerHarness): process_project_root=self.install_root, remote="prgs", require_binding=True, + mode="derivation", ) self.assertFalse(got.get("block"), got.get("reasons")) self.assertEqual(got["resolved_slug"], TARGET_SLUG) diff --git a/tests/test_issue_973_cross_repo_canonical_roots.py b/tests/test_issue_973_cross_repo_canonical_roots.py index 91b5232..a854619 100644 --- a/tests/test_issue_973_cross_repo_canonical_roots.py +++ b/tests/test_issue_973_cross_repo_canonical_roots.py @@ -479,6 +479,68 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): self.assertTrue(rel_res.get("success"), rel_res) mock_clear.assert_called_once() + def test_unbound_session_derives_and_retains_configured_target_repository(self): + """B9: Unbound session with a legitimate configured target root derives and retains target repository.""" + prof = { + "profile_name": "prgs-author", + "allowed_repositories": ["Scaled-Tech-Consulting/mcp-control-plane"], + "canonical_repository_root": self.target_root, + } + with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \ + patch.object(mcp_server.session_ctx, "get_session_context", return_value=None): + slug, reasons = mcp_server._canonical_repository_slug(prof, "prgs") + self.assertEqual(slug, "Scaled-Tech-Consulting/mcp-control-plane", f"reasons: {reasons}") + self.assertEqual(reasons, []) + res = mcp_server._trusted_session_repository(prof, "prgs") + self.assertEqual(res["org"], "Scaled-Tech-Consulting") + self.assertEqual(res["repository"], "mcp-control-plane") + + def test_process_root_a_configured_target_b_retains_b(self): + """B9: Process root A (Gitea-Tools) and configured target B (mcp-control-plane) keep B as expected identity when bound.""" + bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"} + with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \ + patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx): + expected = mcp_server._resolve_expected_repository_slug("prgs") + self.assertEqual(expected, "Scaled-Tech-Consulting/mcp-control-plane") + + def test_request_supplied_repository_c_cannot_replace_derived_or_bound_repository_b(self): + """B9: Request-supplied repository C (Timesheet) cannot replace bound repository B (mcp-control-plane).""" + bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"} + with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \ + patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx): + expected = mcp_server._resolve_expected_repository_slug("prgs", org="Scaled-Tech-Consulting", repo="Timesheet") + self.assertEqual(expected, "Scaled-Tech-Consulting/mcp-control-plane") + + def test_known_expected_repo_a_plus_candidate_root_b_fails_closed(self): + """B9: Known expected repository A plus candidate root B fails closed in validation mode.""" + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="env", + expected_slug="Scaled-Tech-Consulting/Gitea-Tools", + process_project_root=self.install_root, + remote="prgs", + require_binding=True, + mode="validation", + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"])) + + def test_validation_mode_with_unprovable_expected_identity_fails_closed(self): + """B9: Validation mode with expected_slug=None and require_binding=True fails closed.""" + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="env", + expected_slug=None, + process_project_root=self.install_root, + remote="prgs", + require_binding=True, + mode="validation", + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertTrue(any("unprovable or missing" in r for r in assessment["reasons"])) + if __name__ == "__main__": unittest.main()