From caaec9ac7a94c02239c7cab7504d22ddab286931 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sat, 25 Jul 2026 19:17:42 -0400 Subject: [PATCH] feat(mcp): enforce strict cross-project mutation boundaries (Closes #707) --- gitea_mcp_server.py | 39 +++++--- remote_repo_guard.py | 67 +++++++++++-- tests/test_cross_project_mutation_boundary.py | 94 +++++++++++++++++++ 3 files changed, 182 insertions(+), 18 deletions(-) create mode 100644 tests/test_cross_project_mutation_boundary.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 915eaad..8e43cd6 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -3325,10 +3325,16 @@ def _effective_remote(remote: str) -> str: return remote -def _resolve(remote: str, host: str | None, org: str | None, repo: str | None): +def _resolve( + remote: str, + host: str | None, + org: str | None, + repo: str | None, + for_mutation: bool = False, +): """Resolve remote + overrides to (host, org, repo). - #714 / #530: when the caller omits org and/or repo, prefer the + #714 / #530 / #707: when the caller omits org and/or repo, prefer the workspace-aligned git remote over ``REMOTES`` defaults (e.g. bare ``remote=prgs`` must not force ``Timesheet`` when the checkout is ``Gitea-Tools``). Explicit caller org/repo always win and are validated @@ -3414,6 +3420,7 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None): # Workspace-filled sides are intentional alignment for #530. org_explicit=org_explicit or filled_org, repo_explicit=repo_explicit or filled_repo, + for_mutation=for_mutation, ) return (resolved_host, resolved_org, resolved_repo) @@ -3475,8 +3482,9 @@ def _enforce_remote_repo_guard( *, org_explicit: bool, repo_explicit: bool, + for_mutation: bool = False, ) -> None: - """Fail closed on a remote/repo mismatch vs. the local git remote (#530). + """Fail closed on a remote/repo mismatch vs. the local git remote (#530/#707). Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is set, so the unit suite (which calls tools with bare remotes against mocked APIs) @@ -3488,6 +3496,12 @@ def _enforce_remote_repo_guard( ): return local_remote_url = _local_git_remote_url(remote) + primary_org = None + primary_repo = None + ctx = session_ctx.get_session_context() + if ctx: + primary_org = ctx.get("org") + primary_repo = ctx.get("repository") assessment = remote_repo_guard.assess_remote_repo_match( remote=remote, resolved_org=resolved_org, @@ -3495,6 +3509,9 @@ def _enforce_remote_repo_guard( local_remote_url=local_remote_url, org_explicit=org_explicit, repo_explicit=repo_explicit, + for_mutation=for_mutation, + primary_org=primary_org, + primary_repo=primary_repo, ) if assessment["block"]: raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment)) @@ -4063,7 +4080,7 @@ def gitea_lock_issue( resolved_worktree = issue_lock_worktree.resolve_author_worktree_path( worktree_path, _canonical_local_git_root() ) - h, o, r = _resolve(remote, host, org, repo) + h, o, r = _resolve(remote, host, org, repo, for_mutation=True) existing_issue_lock = _load_existing_issue_lock( remote=remote, org=o, repo=r, issue_number=issue_number ) @@ -5239,7 +5256,7 @@ def gitea_create_pr( org=org, repo=repo, ) - h, o, r = _resolve(remote, host, org, repo) + h, o, r = _resolve(remote, host, org, repo, for_mutation=True) # ── Issue Lock Validation (Issue #194 / #196 / #443) ── lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head) @@ -9560,7 +9577,7 @@ def gitea_edit_pr( required_permission="gitea.pr.close", ) - h, o, r = _resolve(remote, host, org, repo) + h, o, r = _resolve(remote, host, org, repo, for_mutation=True) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}" @@ -9844,7 +9861,7 @@ def gitea_commit_files( verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo) processed_files, source_proofs = _prepare_commit_payload_files(files) - h, o, r = _resolve(remote, host, org, repo) + h, o, r = _resolve(remote, host, org, repo, for_mutation=True) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/contents" @@ -9980,7 +9997,7 @@ def gitea_publish_unpublished_issue_branch( repo=repo, ) - h, o, r = _resolve(remote, host, org, repo) + h, o, r = _resolve(remote, host, org, repo, for_mutation=True) git_remote = (git_remote_name or remote or "").strip() existing_lock = issue_lock_store.load_issue_lock( @@ -10183,7 +10200,7 @@ def gitea_bootstrap_author_issue_worktree( repo=repo, ) - h, o, r = _resolve(remote, host, org, repo) + h, o, r = _resolve(remote, host, org, repo, for_mutation=True) canonical_root = _canonical_local_git_root() import author_issue_bootstrap @@ -12131,7 +12148,7 @@ def gitea_reconcile_merged_cleanups( "blocker_kind": "invalid_pr_number", } - h, o, r = _resolve(remote, host, org, repo) + h, o, r = _resolve(remote, host, org, repo, for_mutation=True) auth = _auth(h) base = repo_api_url(h, o, r) @@ -12521,7 +12538,7 @@ def gitea_assess_already_landed_reconciliation( "permission_report": _permission_block_report("gitea.read"), } - h, o, r = _resolve(remote, host, org, repo) + h, o, r = _resolve(remote, host, org, repo, for_mutation=True) auth = _auth(h) pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) diff --git a/remote_repo_guard.py b/remote_repo_guard.py index 2757eec..7e86afd 100644 --- a/remote_repo_guard.py +++ b/remote_repo_guard.py @@ -54,20 +54,62 @@ def assess_remote_repo_match( local_remote_url: str | None, org_explicit: bool, repo_explicit: bool, + for_mutation: bool = False, + primary_org: str | None = None, + primary_repo: str | None = None, ) -> dict: """Fail closed when the resolved org/repo disagrees with the local git remote. - The guard is intentionally conservative: + The guard enforces two protection levels: - * When the caller passed both ``org`` and ``repo`` explicitly, their intent is - authoritative and the guard never blocks. - * When the local git remote URL is unavailable (``None``/empty), corroboration - is impossible, so the guard does not block (best-effort only). - * Otherwise, the resolved ``org/repo`` slug must appear in the local remote URL - (case-insensitive); if it does not, the guard blocks. + 1. Cross-Project Mutation Boundary (#707): + When ``for_mutation`` is True (codebase mutation operation: branch, commit, PR, + merge, branch deletion), the target repository (``resolved_org/resolved_repo``) + MUST match the primary authorized project context (``primary_org/primary_repo`` + or parsed from ``local_remote_url``). Any attempt to mutate a different project + fails closed, even if explicit org/repo were passed. Metadata operations (such + as creating issues or commenting on issues) across projects remain allowed. + + 2. Workspace Mismatch Guard (#530): + When ``for_mutation`` is False (or targets match), bare remotes must resolve + to an org/repo slug present in the local git remote URL. When explicit org/repo + are supplied for non-mutation operations, the caller's intent is authoritative. """ reasons: list[str] = [] + eff_primary_org = primary_org + eff_primary_repo = primary_repo + if not eff_primary_org or not eff_primary_repo: + parsed_primary = parse_org_repo_from_remote_url(local_remote_url) + if parsed_primary: + eff_primary_org = eff_primary_org or parsed_primary[0] + eff_primary_repo = eff_primary_repo or parsed_primary[1] + + # #707: Enforce cross-project codebase mutation boundary if for_mutation is True + if for_mutation and eff_primary_org and eff_primary_repo: + if ( + resolved_org.lower() != eff_primary_org.lower() + or resolved_repo.lower() != eff_primary_repo.lower() + ): + reasons.append( + f"Cross-project mutation guard (#707): Attempted codebase mutation targeting " + f"'{resolved_org}/{resolved_repo}' outside of primary authorized project " + f"context '{eff_primary_org}/{eff_primary_repo}'" + ) + return { + "proven": False, + "block": True, + "cross_project_mutation_block": True, + "reasons": reasons, + "remote": remote, + "resolved_org": resolved_org, + "resolved_repo": resolved_repo, + "primary_org": eff_primary_org, + "primary_repo": eff_primary_repo, + "local_remote_url": local_remote_url, + "remediation": f"Cross-project codebase work is forbidden. Create an issue in the target repository ('{resolved_org}/{resolved_repo}') instead.", + } + if org_explicit and repo_explicit: return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url) @@ -88,6 +130,16 @@ def assess_remote_repo_match( def format_remote_repo_guard_error(assessment: dict) -> str: """Single RuntimeError message for the MCP resolver gate.""" + if assessment.get("cross_project_mutation_block"): + resolved = f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}" + primary = f"{assessment.get('primary_org')}/{assessment.get('primary_repo')}" + return ( + f"Cross-project mutation guard (#707): Attempted codebase mutation targeting '{resolved}' " + f"outside of primary authorized project context '{primary}'. " + f"Cross-project codebase work (creating branches, committing files, creating PRs) is forbidden; " + f"create an issue in the target repository ('{resolved}') instead." + ) + reasons = "; ".join( assessment.get("reasons") or ["remote/repo resolution mismatch"] ) @@ -120,3 +172,4 @@ def _assessment( "local_remote_url": local_remote_url, "remediation": REMEDIATION, } + diff --git a/tests/test_cross_project_mutation_boundary.py b/tests/test_cross_project_mutation_boundary.py new file mode 100644 index 0000000..8559852 --- /dev/null +++ b/tests/test_cross_project_mutation_boundary.py @@ -0,0 +1,94 @@ +import os +import sys +import unittest +from unittest import mock + +import remote_repo_guard +import gitea_mcp_server as server +from session_context_binding import _reset_session_context_for_testing, bind_session_context + + +LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + + +class TestCrossProjectMutationBoundary(unittest.TestCase): + def setUp(self): + os.environ["PYTEST_CURRENT_TEST"] = "test" + _reset_session_context_for_testing() + + def tearDown(self): + _reset_session_context_for_testing() + os.environ.pop("PYTEST_CURRENT_TEST", None) + + def test_cross_project_codebase_mutation_blocked(self): + """Codebase mutations targeting another project must fail closed (#707).""" + assessment = remote_repo_guard.assess_remote_repo_match( + remote="prgs", + resolved_org="Other-Org", + resolved_repo="Other-Repo", + local_remote_url=LOCAL_GITEA_TOOLS_URL, + org_explicit=True, + repo_explicit=True, + for_mutation=True, + ) + self.assertTrue(assessment["block"]) + self.assertTrue(assessment.get("cross_project_mutation_block")) + self.assertEqual(assessment["primary_org"], "Scaled-Tech-Consulting") + self.assertEqual(assessment["primary_repo"], "Gitea-Tools") + + err_msg = remote_repo_guard.format_remote_repo_guard_error(assessment) + self.assertIn("Cross-project mutation guard (#707)", err_msg) + self.assertIn("Attempted codebase mutation targeting 'Other-Org/Other-Repo'", err_msg) + self.assertIn("outside of primary authorized project context 'Scaled-Tech-Consulting/Gitea-Tools'", err_msg) + self.assertIn("create an issue in the target repository ('Other-Org/Other-Repo') instead", err_msg) + + def test_same_project_codebase_mutation_allowed(self): + """Codebase mutations targeting the primary project context are allowed.""" + assessment = remote_repo_guard.assess_remote_repo_match( + remote="prgs", + resolved_org="Scaled-Tech-Consulting", + resolved_repo="Gitea-Tools", + local_remote_url=LOCAL_GITEA_TOOLS_URL, + org_explicit=True, + repo_explicit=True, + for_mutation=True, + ) + self.assertFalse(assessment["block"]) + self.assertFalse(assessment.get("cross_project_mutation_block", False)) + + def test_cross_project_metadata_operation_allowed(self): + """Metadata operations (issue creation, comments) across project boundaries are allowed.""" + assessment = remote_repo_guard.assess_remote_repo_match( + remote="prgs", + resolved_org="Other-Org", + resolved_repo="Other-Repo", + local_remote_url=LOCAL_GITEA_TOOLS_URL, + org_explicit=True, + repo_explicit=True, + for_mutation=False, + ) + self.assertTrue(assessment["proven"]) + self.assertFalse(assessment["block"]) + + @mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"}) + @mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL) + def test_mcp_server_resolve_cross_project_mutation_fails_closed(self, mock_remote): + """_resolve with for_mutation=True blocks cross-project targets.""" + with self.assertRaises(RuntimeError) as ctx: + server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=True) + + err = str(ctx.exception) + self.assertIn("Cross-project mutation guard (#707)", err) + self.assertIn("create an issue in the target repository", err) + + @mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"}) + @mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL) + def test_mcp_server_resolve_cross_project_metadata_succeeds(self, mock_remote): + """_resolve with for_mutation=False allows explicit cross-project target.""" + host, org, repo = server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=False) + self.assertEqual(org, "Other-Org") + self.assertEqual(repo, "Other-Repo") + + +if __name__ == "__main__": + unittest.main()