From 9d96cf4cfa3058980bd0544ae1039724851f2677 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 29 Jul 2026 14:44:50 -0400 Subject: [PATCH] fix(runtime): derive expected repository identity independently of configured root (Closes #973) --- canonical_repository_root.py | 6 + gitea_mcp_server.py | 34 ++++- namespace_workspace_binding.py | 1 + ...est_issue_706_canonical_repository_root.py | 2 + ...st_issue_973_cross_repo_canonical_roots.py | 135 ++++++++++++++---- 5 files changed, 146 insertions(+), 32 deletions(-) diff --git a/canonical_repository_root.py b/canonical_repository_root.py index 6fde057..cea8a11 100644 --- a/canonical_repository_root.py +++ b/canonical_repository_root.py @@ -221,6 +221,12 @@ def assess_canonical_repository_root( f"remote identity to confirm authorization for '{expected}' " "(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 7d3fa38..f1b6448 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -500,12 +500,37 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: return workspace +def _process_root_git_remote_url(remote_name: str) -> str | None: + """Best-effort local ``git remote get-url`` strictly inside ``PROJECT_ROOT``. + + #973 (B8): Must derive expected repository identity from an authority + independent of the candidate configured canonical root. + """ + try: + proc = subprocess.run( + ["git", "remote", "get-url", remote_name], + capture_output=True, + text=True, + cwd=PROJECT_ROOT, + ) + if proc.returncode != 0: + return None + url = (proc.stdout or "").strip() + return url or None + except Exception: + return None + + def _resolve_expected_repository_slug( remote: str | None = None, org: str | None = None, repo: str | None = None, ) -> str | None: - """Resolve expected repository slug from parameters, session context, or remote URL.""" + """Resolve expected repository slug from parameters, session context, or process root. + + Must derive expected repository identity ONLY from trusted sources independent + of the candidate configured canonical root (#973 B8). + """ if org and repo: return session_ctx.format_repository_slug(org, repo) bound = session_ctx.get_session_context() or {} @@ -515,7 +540,7 @@ def _resolve_expected_repository_slug( return session_ctx.format_repository_slug(b_org, b_repo) eff_remote = remote or bound.get("remote") or _effective_remote() parsed = remote_repo_guard.parse_org_repo_from_remote_url( - _local_git_remote_url(eff_remote) + _process_root_git_remote_url(eff_remote) ) if parsed: return session_ctx.format_repository_slug(parsed[0], parsed[1]) @@ -969,10 +994,7 @@ def _enforce_canonical_repository_root( if not configured_value: return - bound = session_ctx.get_session_context() or {} - expected_slug = session_ctx.format_repository_slug( - bound.get("org"), bound.get("repository") - ) + expected_slug = _resolve_expected_repository_slug(remote) assessment = crr.assess_canonical_repository_root( configured_value=configured_value, source=source, diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py index 9025be9..a51d4c7 100644 --- a/namespace_workspace_binding.py +++ b/namespace_workspace_binding.py @@ -247,6 +247,7 @@ def resolve_namespace_mutation_context( expected_slug=expected_slug, process_project_root=process_root, remote=remote, + require_binding=True, ) canonical_root = crr_assessment["canonical_repo_root"] roots_aligned = crr_assessment["proven"] diff --git a/tests/test_issue_706_canonical_repository_root.py b/tests/test_issue_706_canonical_repository_root.py index 9120f54..2a5c4cb 100644 --- a/tests/test_issue_706_canonical_repository_root.py +++ b/tests/test_issue_706_canonical_repository_root.py @@ -241,6 +241,7 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase): process_project_root=self.install, env={}, configured_canonical_root=self.target, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertEqual(ctx["canonical_repo_root"], self.target) self.assertTrue(ctx["roots_aligned"]) @@ -268,6 +269,7 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase): env={}, current_branch="feat/issue-1", configured_canonical_root=self.target, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertFalse(assessment["block"], assessment.get("reasons")) self.assertEqual(assessment["canonical_repo_root"], self.target) diff --git a/tests/test_issue_973_cross_repo_canonical_roots.py b/tests/test_issue_973_cross_repo_canonical_roots.py index 47e8ca6..91b5232 100644 --- a/tests/test_issue_973_cross_repo_canonical_roots.py +++ b/tests/test_issue_973_cross_repo_canonical_roots.py @@ -74,6 +74,22 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): subprocess.run(["git", "add", "README.md"], cwd=self.target_root, check=True) subprocess.run(["git", "commit", "-m", "initial"], cwd=self.target_root, check=True) + # Add remotes to simulate real git repositories with identities + subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"], cwd=self.install_root, check=True) + subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git"], cwd=self.target_root, check=True) + + # Create simulated foreign repository root + self.evil_root = os.path.join(self.tmp_dir, "Evil-Repo") + os.makedirs(self.evil_root) + subprocess.run(["git", "init", "-b", "master"], cwd=self.evil_root, check=True) + subprocess.run(["git", "config", "user.email", "evil@example.com"], cwd=self.evil_root, check=True) + subprocess.run(["git", "config", "user.name", "Evil User"], cwd=self.evil_root, check=True) + with open(os.path.join(self.evil_root, "README.md"), "w") as f: + f.write("evil\n") + subprocess.run(["git", "add", "README.md"], cwd=self.evil_root, check=True) + subprocess.run(["git", "commit", "-m", "initial"], cwd=self.evil_root, check=True) + subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Someone-Else/Evil-Repo.git"], cwd=self.evil_root, check=True) + # Create branches/ directory and a valid registered worktree in target repository self.target_branches = os.path.join(self.target_root, "branches") self.target_worktree = os.path.join(self.target_branches, "rev-pr-99") @@ -101,47 +117,107 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): process_project_root=self.install_root, env={}, configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertEqual(ctx["canonical_repo_root"], self.target_root) self.assertTrue(ctx["roots_aligned"]) self.assertTrue(ctx["canonical_root_assessment"]["proven"]) def test_expected_repository_identity_match(self): - with patch("canonical_repository_root.repository_identity_slug", return_value="Scaled-Tech-Consulting/Gitea-Tools"): - assessment = crr.assess_canonical_repository_root( - configured_value=self.target_root, - source="test", - expected_slug="Scaled-Tech-Consulting/Gitea-Tools", - process_project_root=self.install_root, - ) - self.assertTrue(assessment["proven"]) - self.assertFalse(assessment["block"]) + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="test", + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", + process_project_root=self.install_root, + remote="prgs", + ) + self.assertTrue(assessment["proven"]) + self.assertFalse(assessment["block"]) def test_foreign_repository_identity_mismatch(self): - with patch("canonical_repository_root.repository_identity_slug", return_value="Someone-Else/Evil-Repo"): + assessment = crr.assess_canonical_repository_root( + configured_value=self.evil_root, + source="test", + expected_slug="Scaled-Tech-Consulting/Gitea-Tools", + process_project_root=self.install_root, + remote="prgs", + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"])) + + def test_native_repository_binding_mismatch(self): + ctx = nwb.resolve_namespace_mutation_context( + role_kind="reviewer", + worktree_path=self.target_worktree, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.evil_root, + expected_slug="Scaled-Tech-Consulting/Gitea-Tools", + remote="prgs", + ) + self.assertFalse(ctx["roots_aligned"]) + self.assertFalse(ctx["canonical_root_assessment"]["proven"]) + self.assertTrue(any("identity mismatch" in r for r in ctx["canonical_root_assessment"]["reasons"])) + + def test_unpatched_foreign_configured_root_derives_expected_from_process_root_and_blocks(self): + """B8: Production path test where foreign configured root cannot self-authorize.""" + with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \ + patch.object(mcp_server, "_configured_canonical_root", return_value=(self.evil_root, "env")): + expected_slug = mcp_server._resolve_expected_repository_slug("prgs") + self.assertEqual(expected_slug, "Scaled-Tech-Consulting/Gitea-Tools") assessment = crr.assess_canonical_repository_root( - configured_value=self.target_root, - source="test", - expected_slug="Scaled-Tech-Consulting/Gitea-Tools", + configured_value=self.evil_root, + source="env", + expected_slug=expected_slug, process_project_root=self.install_root, + remote="prgs", + require_binding=True, ) self.assertFalse(assessment["proven"]) self.assertTrue(assessment["block"]) + self.assertEqual(assessment["resolved_slug"], "Someone-Else/Evil-Repo") self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"])) - def test_native_repository_binding_mismatch(self): - with patch("canonical_repository_root.repository_identity_slug", return_value="Foreign/Repo"): - ctx = nwb.resolve_namespace_mutation_context( - role_kind="reviewer", - worktree_path=self.target_worktree, + def test_unpatched_valid_cross_repo_matching_session_context(self): + """B8: Valid cross-repo namespace matches when session context is bound to target repo.""" + 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_slug = mcp_server._resolve_expected_repository_slug("prgs") + self.assertEqual(expected_slug, "Scaled-Tech-Consulting/mcp-control-plane") + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="env", + expected_slug=expected_slug, process_project_root=self.install_root, - env={}, - configured_canonical_root=self.target_root, - expected_slug="Scaled-Tech-Consulting/Gitea-Tools", + remote="prgs", + require_binding=True, ) - self.assertFalse(ctx["roots_aligned"]) - self.assertFalse(ctx["canonical_root_assessment"]["proven"]) - self.assertTrue(any("identity mismatch" in r for r in ctx["canonical_root_assessment"]["reasons"])) + self.assertTrue(assessment["proven"]) + self.assertFalse(assessment["block"]) + self.assertEqual(assessment["resolved_slug"], "Scaled-Tech-Consulting/mcp-control-plane") + + def test_unprovable_expected_identity_fails_closed(self): + """B8: If expected repository identity is unprovable for a configured root, fail closed.""" + no_remote_root = os.path.join(self.tmp_dir, "no-remote-process-root") + os.makedirs(no_remote_root) + subprocess.run(["git", "init", "-b", "master"], cwd=no_remote_root, check=True) + with patch.object(mcp_server, "PROJECT_ROOT", no_remote_root), \ + patch.object(mcp_server.session_ctx, "get_session_context", return_value=None): + expected_slug = mcp_server._resolve_expected_repository_slug("prgs") + self.assertIsNone(expected_slug) + assessment = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="env", + expected_slug=expected_slug, + process_project_root=no_remote_root, + remote="prgs", + require_binding=True, + ) + self.assertFalse(assessment["proven"]) + self.assertTrue(assessment["block"]) + self.assertTrue(any("unprovable or missing" in r for r in assessment["reasons"])) def test_missing_canonical_root(self): ctx = nwb.resolve_namespace_mutation_context( @@ -232,6 +308,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): process_project_root=self.install_root, env={}, configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertTrue(assessment["block"]) finally: @@ -246,6 +323,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): process_project_root=self.install_root, env={}, configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertFalse(assessment["block"]) @@ -259,6 +337,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): process_project_root=self.install_root, env={}, configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertTrue(assessment["block"]) self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"])) @@ -271,6 +350,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): process_project_root=self.install_root, env={}, configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertFalse(assessment["block"]) @@ -284,6 +364,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): process_project_root=self.install_root, env={}, configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertTrue(assessment["block"]) self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"])) @@ -299,6 +380,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): process_project_root=self.install_root, env={}, configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertTrue(assessment["block"]) self.assertTrue(any("is not under" in r for r in assessment["reasons"])) @@ -313,6 +395,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): process_project_root=self.install_root, env={}, configured_canonical_root=self.target_root, + expected_slug="Scaled-Tech-Consulting/mcp-control-plane", ) self.assertTrue(assessment["block"]) self.assertTrue(any("does not exist" in r for r in assessment["reasons"])) @@ -382,7 +465,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): remote="prgs", worktree=self.target_worktree, org="Scaled-Tech-Consulting", - repo="Gitea-Tools", + repo="mcp-control-plane", ) self.assertTrue(acq_res.get("success"), acq_res) @@ -391,7 +474,7 @@ class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): worktree=self.target_worktree, remote="prgs", org="Scaled-Tech-Consulting", - repo="Gitea-Tools", + repo="mcp-control-plane", ) self.assertTrue(rel_res.get("success"), rel_res) mock_clear.assert_called_once()