From 6a53308473ee20c9dbc3c47f6fdf515e916c66dc Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 29 Jul 2026 13:19:29 -0400 Subject: [PATCH] fix(runtime): support validated cross-repository canonical roots (#973) --- gitea_config.py | 4 + namespace_workspace_binding.py | 111 +++++++++++- ...est_issue_706_canonical_repository_root.py | 2 +- ...st_issue_973_cross_repo_canonical_roots.py | 160 ++++++++++++++++++ 4 files changed, 271 insertions(+), 6 deletions(-) create mode 100644 tests/test_issue_973_cross_repo_canonical_roots.py diff --git a/gitea_config.py b/gitea_config.py index c69553c..edea975 100644 --- a/gitea_config.py +++ b/gitea_config.py @@ -1180,6 +1180,10 @@ RECOGNIZED_GITEA_ENV_KEYS = frozenset({ "GITEA_SERVER_PROVENANCE", "GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE", + "GITEA_REVIEWER_WORKTREE", + "GITEA_MERGER_WORKTREE", + "GITEA_CANONICAL_REPOSITORY_ROOT", + "GITEA_MCP_SESSION_STATE_TTL_HOURS", "GITEA_DISABLE_KEYCHAIN", "GITEA_CONTROL_PLANE_DB", "GITEA_DB_PATH", diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py index be2ca82..911bf78 100644 --- a/namespace_workspace_binding.py +++ b/namespace_workspace_binding.py @@ -8,8 +8,10 @@ poison workspace purity checks in another namespace. from __future__ import annotations import os +import subprocess import author_mutation_worktree as amw +import canonical_repository_root as crr ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV @@ -152,6 +154,60 @@ def resolve_namespace_workspace( return os.path.realpath(process_project_root), "MCP server process root (default)" +def verify_git_common_directory_membership( + workspace_path: str, + canonical_repo_root: str, +) -> tuple[bool, str | None]: + """Verify that workspace_path belongs to canonical_repo_root via git common-dir or branches containment.""" + ws = (workspace_path or "").strip() + root = (canonical_repo_root or "").strip() + if not ws or not root: + return False, "empty workspace or canonical root path" + + try: + real_ws = os.path.realpath(os.path.abspath(ws)) + real_root = os.path.realpath(os.path.abspath(root)) + except Exception as exc: + return False, f"invalid workspace or root path: {exc}" + + if real_ws == real_root: + return True, None + + if not os.path.isdir(real_ws): + return False, f"workspace directory '{real_ws}' does not exist" + + try: + res = subprocess.run( + ["git", "-C", real_ws, "rev-parse", "--git-common-dir"], + capture_output=True, + text=True, + check=False, + ) + if res.returncode == 0: + common_raw = (res.stdout or "").strip() + common_dir = amw._realpath_git_common_dir(real_ws, common_raw) + real_common = os.path.realpath(common_dir) + canonical_git = os.path.realpath(os.path.join(real_root, ".git")) + + if real_common in (canonical_git, real_root) or os.path.dirname(real_common) == real_root: + return True, None + return ( + False, + f"workspace '{real_ws}' git common directory '{real_common}' does not match " + f"canonical repository root '{real_root}' (.git at '{canonical_git}')" + ) + except Exception: + pass + + if amw.is_path_under_branches(real_ws, real_root): + return True, None + + return ( + False, + f"workspace '{real_ws}' does not belong to canonical repository root '{real_root}'" + ) + + def resolve_namespace_mutation_context( *, role_kind: str, @@ -163,6 +219,8 @@ def resolve_namespace_mutation_context( worktree: str | None = None, profile_name: str | None = None, configured_canonical_root: str | None = None, + expected_slug: str | None = None, + remote: str | None = None, ) -> dict: """Shared workspace resolution for runtime_context and mutation guards. @@ -180,11 +238,30 @@ def resolve_namespace_mutation_context( env_map = env if env is not None else os.environ process_root = os.path.realpath(process_project_root) role = normalize_role_kind(role_kind, profile_name=profile_name) - configured = (configured_canonical_root or "").strip() - if configured: - canonical_root = os.path.realpath(configured) + + configured_val = (configured_canonical_root or "").strip() + if configured_val: + crr_assessment = crr.assess_canonical_repository_root( + configured_value=configured_val, + source="configured_canonical_root", + expected_slug=expected_slug, + process_project_root=process_root, + remote=remote, + ) + canonical_root = crr_assessment["canonical_repo_root"] + roots_aligned = crr_assessment["proven"] else: - canonical_root = amw.resolve_canonical_repo_root(process_root, process_root) + crr_assessment = { + "proven": True, + "block": False, + "reasons": [], + "configured": False, + "canonical_repo_root": amw.resolve_canonical_repo_root(process_root, process_root), + "resolved_slug": None, + "source": None, + } + canonical_root = crr_assessment["canonical_repo_root"] + roots_aligned = (canonical_root == process_root) durable: dict | None = None if role == "author": @@ -234,7 +311,8 @@ def resolve_namespace_mutation_context( "ignored_bindings": demotions + (pollution.get("ignored_bindings") or []), "process_project_root": process_root, "canonical_repo_root": canonical_root, - "roots_aligned": canonical_root == process_root, + "roots_aligned": roots_aligned, + "canonical_root_assessment": crr_assessment, } if durable is not None: result["author_worktree_resolution"] = durable @@ -387,6 +465,8 @@ def assess_namespace_mutation_workspace( profile_name: str | None = None, current_branch: str | None = None, configured_canonical_root: str | None = None, + expected_slug: str | None = None, + remote: str | None = None, ) -> dict: """Evaluate namespace workspace binding before preflight/mutation.""" ctx = resolve_namespace_mutation_context( @@ -399,6 +479,8 @@ def assess_namespace_mutation_workspace( session_lock_worktree=session_lock_worktree, profile_name=profile_name, configured_canonical_root=configured_canonical_root, + expected_slug=expected_slug, + remote=remote, ) mutation_workspace = ctx["workspace_path"] binding_source = ctx["workspace_binding_source"] @@ -422,6 +504,25 @@ def assess_namespace_mutation_workspace( reasons = list(metadata.get("reasons") or []) operator_recovery = ctx.get("operator_recovery") + + crr_reasons = list(ctx.get("canonical_root_assessment", {}).get("reasons") or []) + if crr_reasons: + reasons.extend(crr_reasons) + + if not ctx.get("roots_aligned"): + if not crr_reasons: + reasons.append( + f"unsafe_process_root_workspace_alignment: process root '{process_root}' " + f"and canonical root '{ctx['canonical_repo_root']}' disagree" + ) + + if os.path.exists(mutation_workspace): + valid_common, common_err = verify_git_common_directory_membership( + mutation_workspace, ctx["canonical_repo_root"] + ) + if not valid_common and common_err: + reasons.append(common_err) + if role == "author": # #618 durable resolution already validated existence, membership, # branches/, lock ownership, and traversal safety when present. diff --git a/tests/test_issue_706_canonical_repository_root.py b/tests/test_issue_706_canonical_repository_root.py index a51004f..9120f54 100644 --- a/tests/test_issue_706_canonical_repository_root.py +++ b/tests/test_issue_706_canonical_repository_root.py @@ -243,7 +243,7 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase): configured_canonical_root=self.target, ) self.assertEqual(ctx["canonical_repo_root"], self.target) - self.assertFalse(ctx["roots_aligned"]) + self.assertTrue(ctx["roots_aligned"]) def test_target_worktree_is_member_of_target_root(self): got = nwb.amw.assess_workspace_repo_membership( diff --git a/tests/test_issue_973_cross_repo_canonical_roots.py b/tests/test_issue_973_cross_repo_canonical_roots.py new file mode 100644 index 0000000..15ca7bf --- /dev/null +++ b/tests/test_issue_973_cross_repo_canonical_roots.py @@ -0,0 +1,160 @@ +"""Regression tests for Issue #973: validated cross-repository canonical roots.""" + +from __future__ import annotations + +import os +import shutil +import tempfile +import unittest +from pathlib import Path +import subprocess + +import gitea_config +import namespace_workspace_binding as nwb +import canonical_repository_root as crr + + +class TestIssue973RecognizedEnvKeys(unittest.TestCase): + """Test recognized environment variable keys under #973.""" + + def test_recognized_gitea_env_keys(self): + for key in ( + "GITEA_CANONICAL_REPOSITORY_ROOT", + "GITEA_REVIEWER_WORKTREE", + "GITEA_MERGER_WORKTREE", + "GITEA_MCP_SESSION_STATE_TTL_HOURS", + ): + self.assertIn(key, gitea_config.RECOGNIZED_GITEA_ENV_KEYS) + + def test_get_unconsumed_gitea_env_overrides_ignores_recognized(self): + env = { + "GITEA_CANONICAL_REPOSITORY_ROOT": "/some/path", + "GITEA_REVIEWER_WORKTREE": "/some/reviewer/path", + "GITEA_MERGER_WORKTREE": "/some/merger/path", + "GITEA_MCP_SESSION_STATE_TTL_HOURS": "24", + "GITEA_UNRECOGNIZED_FOO_VAR": "bar", + } + unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env) + self.assertNotIn("GITEA_CANONICAL_REPOSITORY_ROOT", unconsumed) + self.assertNotIn("GITEA_REVIEWER_WORKTREE", unconsumed) + self.assertNotIn("GITEA_MERGER_WORKTREE", unconsumed) + self.assertNotIn("GITEA_MCP_SESSION_STATE_TTL_HOURS", unconsumed) + self.assertIn("GITEA_UNRECOGNIZED_FOO_VAR", unconsumed) + + +class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase): + """Test workspace binding and canonical root validation for cross-repo namespaces.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.tmp_dir = os.path.realpath(self._tmp.name) + + # Create simulated installation root + self.install_root = os.path.join(self.tmp_dir, "Gitea-Tools") + os.makedirs(self.install_root) + subprocess.run(["git", "init", "-b", "master"], cwd=self.install_root, check=True) + + # Create simulated target repository root + self.target_root = os.path.join(self.tmp_dir, "mcp-control-plane") + os.makedirs(self.target_root) + subprocess.run(["git", "init", "-b", "master"], cwd=self.target_root, check=True) + + # Create branches/ directory and a valid worktree in target repository + self.target_branches = os.path.join(self.target_root, "branches") + os.makedirs(self.target_branches) + self.target_worktree = os.path.join(self.target_branches, "rev-pr-99") + os.makedirs(self.target_worktree) + # Create git common dir linking target_worktree to target_root + dot_git_file = os.path.join(self.target_worktree, ".git") + git_dir_target = os.path.join(self.target_root, ".git", "worktrees", "rev-pr-99") + os.makedirs(git_dir_target, exist_ok=True) + with open(dot_git_file, "w") as f: + f.write(f"gitdir: {git_dir_target}\n") + with open(os.path.join(git_dir_target, "commondir"), "w") as f: + f.write("../../..\n") + + def tearDown(self): + self._tmp.cleanup() + + def test_valid_cross_repo_canonical_root(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.target_root, + ) + self.assertEqual(ctx["canonical_repo_root"], self.target_root) + self.assertTrue(ctx["roots_aligned"]) + self.assertTrue(ctx["canonical_root_assessment"]["proven"]) + + def test_nonexistent_configured_canonical_root(self): + nonexistent = os.path.join(self.tmp_dir, "nonexistent-repo") + ctx = nwb.resolve_namespace_mutation_context( + role_kind="reviewer", + worktree_path=self.target_worktree, + process_project_root=self.install_root, + env={}, + configured_canonical_root=nonexistent, + ) + self.assertFalse(ctx["roots_aligned"]) + self.assertFalse(ctx["canonical_root_assessment"]["proven"]) + self.assertTrue(any("does not exist" in r for r in ctx["canonical_root_assessment"]["reasons"])) + + def test_non_git_configured_canonical_root(self): + non_git = os.path.join(self.tmp_dir, "non-git-dir") + os.makedirs(non_git) + ctx = nwb.resolve_namespace_mutation_context( + role_kind="reviewer", + worktree_path=self.target_worktree, + process_project_root=self.install_root, + env={}, + configured_canonical_root=non_git, + ) + self.assertFalse(ctx["roots_aligned"]) + self.assertFalse(ctx["canonical_root_assessment"]["proven"]) + self.assertTrue(any("not a git repository" in r for r in ctx["canonical_root_assessment"]["reasons"])) + + def test_git_common_directory_membership_matching(self): + valid, err = nwb.verify_git_common_directory_membership( + self.target_worktree, self.target_root + ) + self.assertTrue(valid, err) + self.assertIsNone(err) + + def test_git_common_directory_membership_foreign_repo_blocks(self): + # Foreign worktree created under install_root + foreign_wt = os.path.join(self.install_root, "branches", "foreign-wt") + os.makedirs(foreign_wt) + dot_git_file = os.path.join(foreign_wt, ".git") + git_dir_install = os.path.join(self.install_root, ".git", "worktrees", "foreign-wt") + os.makedirs(git_dir_install, exist_ok=True) + with open(dot_git_file, "w") as f: + f.write(f"gitdir: {git_dir_install}\n") + with open(os.path.join(git_dir_install, "commondir"), "w") as f: + f.write("../../..\n") + + valid, err = nwb.verify_git_common_directory_membership( + foreign_wt, self.target_root + ) + self.assertFalse(valid) + self.assertIn("does not belong", err) + + def test_reviewer_worktree_outside_branches_blocks(self): + outside_wt = os.path.join(self.target_root, "outside_branches_wt") + os.makedirs(outside_wt) + + assessment = nwb.assess_namespace_mutation_workspace( + role_kind="reviewer", + worktree_path=outside_wt, + worktree=None, + process_project_root=self.install_root, + env={}, + configured_canonical_root=self.target_root, + ) + self.assertTrue(assessment["block"]) + self.assertTrue(any("is not under" in r for r in assessment["reasons"])) + + +if __name__ == "__main__": + unittest.main()