feat(mcp): immutable canonical_repository_root for cross-repo namespaces (Closes #706)

The MCP server derived the canonical repository root from the install checkout the server script lives in (PROJECT_ROOT), so any namespace that ran the server against an external repository (e.g. eagenda-author, or the mcp-control-plane reviewer/merger namespaces) failed every mutation: the branches-only / worktree-membership guards (#274) compared the task workspace against the Gitea-Tools .git directory it can never belong to. Separate the two concepts: - PROJECT_ROOT stays the immutable code/install root. - A new, optional, namespace-scoped canonical_repository_root binds the session to the working root of the target repository. Implementation: - canonical_repository_root.py: resolve the binding from profile/env (GITEA_CANONICAL_REPOSITORY_ROOT overrides the profile field), validate existence + git identity, fail closed on missing/conflicting/forged bindings. Preserves the single-repo default when unconfigured. - session_context_binding.py: pin canonical_repository_root into the immutable #714 session context; drift fails closed. - namespace_workspace_binding.py: route the configured target root through resolve_namespace_mutation_context so #274 / membership guards evaluate against the target repository. - gitea_mcp_server.py: seed + enforce the binding in the mutation preflight (both purity-order and FORCE_PRODUCTION_GUARDS paths); validate repository identity and git common-directory membership. - gitea_config.py: validate the optional absolute-path profile field. - gitea_auth.py: surface the config-sourced field through get_profile. No weakening of root-checkout, remote/repository, or anti-stomp guards; the single-repo Gitea-Tools path is unchanged. Tests: two distinct real git repositories/worktrees prove cross-repository bindings resolve, enforce membership, and fail closed on forged/conflicting identity and simultaneous prgs/mdcps isolation. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-17 23:31:39 -04:00
parent 11d1d2e99f
commit 0d8a2c2b1d
7 changed files with 839 additions and 3 deletions
@@ -0,0 +1,423 @@
+"""Issue #706: immutable startup/profile canonical_repository_root capability.
+
+Cross-repository MCP namespaces (e.g. ``eagenda-author``) must bind their
+canonical repository root to the *configured target repository*, not to the
+Gitea-Tools install checkout the server script lives in. These tests prove:
+
+* the configured binding is resolved from profile/env with env precedence,
+* the target repository identity and git common-directory membership are
+  validated (AC3),
+* the branches-only guard (#274) is enforced *inside the target repo* (AC4),
+* missing / conflicting / forged bindings fail closed (AC5),
+* prgs and mdcps namespaces stay simultaneously isolated (AC6),
+* the session binding pins the canonical root immutably,
+* no weakening of the install-root single-repo default (AC8).
+
+Uses two distinct real git repositories/worktrees (AC7).
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import canonical_repository_root as crr  # noqa: E402
+import gitea_config  # noqa: E402
+import namespace_workspace_binding as nwb  # noqa: E402
+import session_context_binding as session_ctx  # noqa: E402
+
+
+def _git(cwd: str, *args: str) -> str:
+    res = subprocess.run(
+        ["git", "-C", cwd, *args],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return res.stdout.strip()
+
+
+def _init_repo(path: Path, remote_url: str, *, remote_name: str = "origin") -> str:
+    """Create a real git repo with one commit and a remote; return realpath root."""
+    path.mkdir(parents=True, exist_ok=True)
+    _git(str(path), "init", "-q")
+    _git(str(path), "config", "user.email", "[email protected]")
+    _git(str(path), "config", "user.name", "Test")
+    _git(str(path), "remote", "add", remote_name, remote_url)
+    (path / "README.md").write_text("seed\n")
+    _git(str(path), "add", "README.md")
+    _git(str(path), "commit", "-q", "-m", "seed")
+    return os.path.realpath(str(path))
+
+
+def _add_worktree(repo_root: str, worktree_path: Path, branch: str) -> str:
+    _git(repo_root, "worktree", "add", "-q", "-b", branch, str(worktree_path))
+    return os.path.realpath(str(worktree_path))
+
+
+# ---------------------------------------------------------------------------
+# configured_canonical_root: resolution + precedence
+# ---------------------------------------------------------------------------
+class TestConfiguredCanonicalRoot(unittest.TestCase):
+    def test_unconfigured_returns_none(self):
+        value, source = crr.configured_canonical_root({}, {})
+        self.assertIsNone(value)
+        self.assertIsNone(source)
+
+    def test_profile_field_used(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "/repo/eAgenda"}, {}
+        )
+        self.assertEqual(value, "/repo/eAgenda")
+        self.assertIn("profile", source)
+
+    def test_env_overrides_profile(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "/repo/eAgenda"},
+            {crr.CANONICAL_ROOT_ENV: "/repo/other"},
+        )
+        self.assertEqual(value, "/repo/other")
+        self.assertIn(crr.CANONICAL_ROOT_ENV, source)
+
+    def test_blank_values_ignored(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "   "},
+            {crr.CANONICAL_ROOT_ENV: ""},
+        )
+        self.assertIsNone(value)
+        self.assertIsNone(source)
+
+
+# ---------------------------------------------------------------------------
+# assess_canonical_repository_root: default (single-repo) path
+# ---------------------------------------------------------------------------
+class TestUnconfiguredFallback(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = self._tmp.name
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_unconfigured_falls_back_to_process_root(self):
+        root = _init_repo(
+            Path(self.tmp) / "install",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        got = crr.assess_canonical_repository_root(
+            configured_value=None,
+            source=None,
+            expected_slug=None,
+            process_project_root=root,
+        )
+        self.assertTrue(got["proven"])
+        self.assertFalse(got["block"])
+        self.assertFalse(got["configured"])
+        self.assertEqual(got["canonical_repo_root"], root)
+
+
+# ---------------------------------------------------------------------------
+# assess_canonical_repository_root: configured cross-repo path
+# ---------------------------------------------------------------------------
+class TestConfiguredCrossRepo(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.install = _init_repo(
+            self.tmp / "Gitea-Tools",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        self.target = _init_repo(
+            self.tmp / "mcp-control-plane",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_valid_configured_root_binds_to_target(self):
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.target,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["proven"], got.get("reasons"))
+        self.assertFalse(got["block"])
+        self.assertTrue(got["configured"])
+        self.assertEqual(got["canonical_repo_root"], self.target)
+        self.assertNotEqual(got["canonical_repo_root"], self.install)
+
+    def test_missing_path_fails_closed(self):
+        got = crr.assess_canonical_repository_root(
+            configured_value=str(self.tmp / "does-not-exist"),
+            source="profile canonical_repository_root",
+            expected_slug=None,
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertFalse(got["proven"])
+        self.assertTrue(any("exist" in r for r in got["reasons"]))
+
+    def test_non_git_path_fails_closed(self):
+        plain = self.tmp / "plain"
+        plain.mkdir()
+        got = crr.assess_canonical_repository_root(
+            configured_value=str(plain),
+            source="profile canonical_repository_root",
+            expected_slug=None,
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("git" in r for r in got["reasons"]))
+
+    def test_identity_mismatch_fails_closed(self):
+        # Configured root is the install repo, but session expects the target
+        # repo identity: a forged/conflicting binding.
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.install,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("identity" in r for r in got["reasons"]))
+
+    def test_unprovable_identity_blocks_when_required(self):
+        noremote = _init_repo(self.tmp / "noremote-src", "x", remote_name="origin")
+        _git(noremote, "remote", "remove", "origin")
+        got = crr.assess_canonical_repository_root(
+            configured_value=noremote,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+            require_binding=True,
+        )
+        self.assertTrue(got["block"])
+
+    def test_repository_identity_slug_reads_remote(self):
+        self.assertEqual(
+            crr.repository_identity_slug(self.target),
+            "Scaled-Tech-Consulting/mcp-control-plane",
+        )
+
+
+# ---------------------------------------------------------------------------
+# Workspace membership + branches-only enforced inside the TARGET repo (AC4)
+# ---------------------------------------------------------------------------
+class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.install = _init_repo(
+            self.tmp / "Gitea-Tools",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        self.target = _init_repo(
+            self.tmp / "mcp-control-plane",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+        (Path(self.target) / "branches").mkdir()
+        self.target_wt = _add_worktree(
+            self.target,
+            Path(self.target) / "branches" / "author-issue-1",
+            "feat/issue-1",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_context_canonical_root_is_configured_target(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="author",
+            worktree_path=self.target_wt,
+            process_project_root=self.install,
+            env={},
+            configured_canonical_root=self.target,
+        )
+        self.assertEqual(ctx["canonical_repo_root"], self.target)
+        self.assertFalse(ctx["roots_aligned"])
+
+    def test_target_worktree_is_member_of_target_root(self):
+        got = nwb.amw.assess_workspace_repo_membership(
+            workspace_path=self.target_wt,
+            canonical_repo_root=self.target,
+        )
+        self.assertTrue(got["proven"], got.get("reasons"))
+
+    def test_target_worktree_not_member_of_install_root(self):
+        got = nwb.amw.assess_workspace_repo_membership(
+            workspace_path=self.target_wt,
+            canonical_repo_root=self.install,
+        )
+        self.assertTrue(got["block"])
+
+    def test_branches_guard_passes_inside_target(self):
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="author",
+            worktree_path=self.target_wt,
+            worktree=None,
+            process_project_root=self.install,
+            env={},
+            current_branch="feat/issue-1",
+            configured_canonical_root=self.target,
+        )
+        self.assertFalse(assessment["block"], assessment.get("reasons"))
+        self.assertEqual(assessment["canonical_repo_root"], self.target)
+
+    def test_target_control_checkout_blocks_branches_guard(self):
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="author",
+            worktree_path=self.target,  # stable target checkout, not branches/
+            worktree=None,
+            process_project_root=self.install,
+            env={},
+            current_branch="master",
+            configured_canonical_root=self.target,
+        )
+        self.assertTrue(assessment["block"])
+
+
+# ---------------------------------------------------------------------------
+# Immutable session pin (AC1/AC5)
+# ---------------------------------------------------------------------------
+class TestSessionCanonicalRootPin(unittest.TestCase):
+    def setUp(self):
+        os.environ["PYTEST_CURRENT_TEST"] = "t"
+        session_ctx._reset_session_context_for_testing()
+
+    def tearDown(self):
+        session_ctx._reset_session_context_for_testing()
+
+    def test_seed_stores_canonical_root(self):
+        ctx = session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        self.assertEqual(ctx["canonical_repository_root"], "/repo/mcp-control-plane")
+
+    def test_canonical_root_drift_fails_closed(self):
+        session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        got = session_ctx.assess_session_context(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            canonical_repository_root="/repo/forged-elsewhere",
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("canonical" in r for r in got["reasons"]))
+
+    def test_matching_canonical_root_passes(self):
+        session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        got = session_ctx.assess_session_context(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        self.assertFalse(got["block"], got["reasons"])
+
+
+# ---------------------------------------------------------------------------
+# Simultaneous prgs / mdcps isolation (AC6)
+# ---------------------------------------------------------------------------
+class TestSimultaneousIsolation(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.prgs = _init_repo(
+            self.tmp / "prgs-repo",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+        self.mdcps = _init_repo(
+            self.tmp / "mdcps-repo",
+            "https://gitea.dadeschools.net/dadeschools/eAgenda.git",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_each_namespace_binds_its_own_target(self):
+        a = crr.assess_canonical_repository_root(
+            configured_value=self.prgs,
+            source="env",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.tmp.as_posix(),
+        )
+        b = crr.assess_canonical_repository_root(
+            configured_value=self.mdcps,
+            source="env",
+            expected_slug="dadeschools/eAgenda",
+            process_project_root=self.tmp.as_posix(),
+        )
+        self.assertEqual(a["canonical_repo_root"], self.prgs)
+        self.assertEqual(b["canonical_repo_root"], self.mdcps)
+        self.assertNotEqual(a["canonical_repo_root"], b["canonical_repo_root"])
+
+    def test_cross_wired_identity_blocks(self):
+        # prgs path claimed under the mdcps identity → forged binding.
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.prgs,
+            source="env",
+            expected_slug="dadeschools/eAgenda",
+            process_project_root=self.tmp.as_posix(),
+        )
+        self.assertTrue(got["block"])
+
+
+# ---------------------------------------------------------------------------
+# Config validation of the profile field (AC5: malformed bindings fail closed)
+# ---------------------------------------------------------------------------
+class TestConfigValidation(unittest.TestCase):
+    def test_absent_is_allowed(self):
+        # single-repo default: no field configured
+        gitea_config._validate_canonical_repository_root("p", None)
+
+    def test_valid_absolute_path_ok(self):
+        gitea_config._validate_canonical_repository_root(
+            "p", "/Users/x/Development/mcp-control-plane"
+        )
+
+    def test_relative_path_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root(
+                "p", "relative/path"
+            )
+
+    def test_empty_string_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root("p", "   ")
+
+    def test_non_string_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root("p", ["/x"])
+
+
+if __name__ == "__main__":
+    unittest.main()