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,

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,

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"]

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)

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", "[email protected]"], 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()