fix(mcp): seed cross-repo session identity from configured canonical root (#706 F1)
Addresses REQUEST_CHANGES review 457 on PR #736. Root cause: _trusted_session_repository derived the session org/repository solely from _workspace_repository_slug -> _local_git_remote_url, which runs `git remote get-url` in PROJECT_ROOT (the Gitea-Tools install checkout). A cross-repository namespace with a configured canonical_repository_root then pinned the Gitea-Tools identity while #274 filesystem membership bound the external target, so _enforce_canonical_repository_root failed closed on a self-inflicted identity mismatch and the mcp-control-plane / eagenda namespaces stayed blocked end-to-end. Fix: when a canonical_repository_root is configured (env over profile) and passes existence / git-toplevel / resolvable-remote-identity validation, the session repository slug is derived from repository_identity_slug(configured root) instead of the install remote. The derived identity is still authorized by the profile allowed_repositories allowlist (never self-authorizing); the canonical filesystem root and org/repo identity remain immutable for the session; invalid / non-git / unresolvable / unallowlisted / forged / drift cases fail closed; unconfigured single-repo Gitea-Tools namespaces keep the install-derived slug unchanged. Tests: new tests/test_issue_706_f1_seed_identity_integration.py drives the real _seed_session_context -> _enforce_canonical_repository_root path with two real temporary git repositories (install=Gitea-Tools, configured target= mcp-control-plane): env + profile config, reviewer + merger role kinds, and fail-closed cases (nonexistent / non-git / unallowlisted / forged identity drift). Reproduces the F1 block before the fix; green after. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
+37
-1
@@ -1675,7 +1675,43 @@ def _trusted_session_repository(
|
|||||||
"repository": None,
|
"repository": None,
|
||||||
"reasons": [str(exc)],
|
"reasons": [str(exc)],
|
||||||
}
|
}
|
||||||
slug = _workspace_repository_slug(remote)
|
# #706 F1: when a cross-repository canonical root is configured, the session
|
||||||
|
# repository identity must be derived from that validated *target* repository,
|
||||||
|
# not from the install-checkout git remote. ``_workspace_repository_slug``
|
||||||
|
# reads ``_local_git_remote_url`` in ``PROJECT_ROOT`` (always Gitea-Tools), so
|
||||||
|
# without this a cross-repo namespace pinned Gitea-Tools while #274 filesystem
|
||||||
|
# membership bound the external root, and ``_enforce_canonical_repository_root``
|
||||||
|
# then failed closed on a self-inflicted identity mismatch. The derived
|
||||||
|
# identity is still authorized by the profile allowlist below (never
|
||||||
|
# self-authorizing). Env-over-profile precedence and fail-closed validation
|
||||||
|
# (existence, git toplevel, resolvable remote identity) are handled by
|
||||||
|
# ``crr``; unconfigured single-repo namespaces keep the install-derived slug.
|
||||||
|
configured_value, configured_source = crr.configured_canonical_root(
|
||||||
|
profile, os.environ
|
||||||
|
)
|
||||||
|
if configured_value:
|
||||||
|
assessment = crr.assess_canonical_repository_root(
|
||||||
|
configured_value=configured_value,
|
||||||
|
source=configured_source,
|
||||||
|
expected_slug=None,
|
||||||
|
process_project_root=PROJECT_ROOT,
|
||||||
|
remote=remote,
|
||||||
|
require_binding=True,
|
||||||
|
)
|
||||||
|
slug = assessment.get("resolved_slug")
|
||||||
|
if assessment.get("block") or not slug:
|
||||||
|
return {
|
||||||
|
"org": None,
|
||||||
|
"repository": None,
|
||||||
|
"reasons": assessment.get("reasons")
|
||||||
|
or [
|
||||||
|
"configured canonical repository root has no resolvable "
|
||||||
|
"repository identity; session repository cannot be "
|
||||||
|
"authorized (fail closed)"
|
||||||
|
],
|
||||||
|
}
|
||||||
|
else:
|
||||||
|
slug = _workspace_repository_slug(remote)
|
||||||
scope = session_ctx.assess_repository_scope(
|
scope = session_ctx.assess_repository_scope(
|
||||||
workspace_slug=slug,
|
workspace_slug=slug,
|
||||||
allowed=allowed,
|
allowed=allowed,
|
||||||
|
|||||||
@@ -0,0 +1,221 @@
|
|||||||
|
"""Issue #706 F1 integration regression (review 457).
|
||||||
|
|
||||||
|
The unit tests in ``test_issue_706_canonical_repository_root.py`` exercise
|
||||||
|
``crr.assess_canonical_repository_root`` / ``session_ctx.seed_session_context_if_unbound``
|
||||||
|
with a *preselected* slug, so they never drive the real end-to-end path that
|
||||||
|
review 457 found broken:
|
||||||
|
|
||||||
|
_seed_session_context -> _trusted_session_repository -> _workspace_repository_slug
|
||||||
|
-> _local_git_remote_url (cwd=PROJECT_ROOT, always Gitea-Tools)
|
||||||
|
|
||||||
|
Before the fix, the session repository identity was pinned from the *install*
|
||||||
|
checkout remote even when a cross-repository ``canonical_repository_root`` was
|
||||||
|
configured to an external repository (e.g. mcp-control-plane). The mutation
|
||||||
|
preflight then derived ``expected_slug`` from that install-derived pin and
|
||||||
|
``_enforce_canonical_repository_root`` failed closed on a self-inflicted
|
||||||
|
identity mismatch, so the stated cross-repo namespaces stayed blocked.
|
||||||
|
|
||||||
|
These tests use two *real* temporary git repositories and drive the live
|
||||||
|
``mcp_server._seed_session_context`` and ``mcp_server._enforce_canonical_repository_root``
|
||||||
|
functions, asserting the session pins the *configured target* identity and that
|
||||||
|
enforcement accepts the target worktree — while every fail-closed property is
|
||||||
|
preserved.
|
||||||
|
"""
|
||||||
|
|
||||||
|
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 mcp_server # noqa: E402 # loads gitea_mcp_server.py into this namespace
|
||||||
|
import canonical_repository_root as crr # noqa: E402
|
||||||
|
import session_context_binding as session_ctx # noqa: E402
|
||||||
|
|
||||||
|
INSTALL_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||||
|
TARGET_SLUG = "Scaled-Tech-Consulting/mcp-control-plane"
|
||||||
|
TARGET_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git"
|
||||||
|
OTHER_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/other-repo.git"
|
||||||
|
|
||||||
|
|
||||||
|
def _git(cwd: str, *args: str) -> str:
|
||||||
|
return subprocess.run(
|
||||||
|
["git", "-C", cwd, *args], capture_output=True, text=True, check=True
|
||||||
|
).stdout.strip()
|
||||||
|
|
||||||
|
|
||||||
|
def _init_repo(path: Path, remote_url: str) -> str:
|
||||||
|
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", "prgs", 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, wt: Path, branch: str) -> str:
|
||||||
|
_git(repo_root, "worktree", "add", "-q", "-b", branch, str(wt))
|
||||||
|
return os.path.realpath(str(wt))
|
||||||
|
|
||||||
|
|
||||||
|
def _profile(role: str, *, canonical: str | None = None,
|
||||||
|
allowed=(TARGET_SLUG,)) -> dict:
|
||||||
|
p = {
|
||||||
|
"profile_name": f"mcp-control-plane-{role}",
|
||||||
|
"role": role,
|
||||||
|
"username": "svc",
|
||||||
|
"allowed_operations": ["gitea.read"],
|
||||||
|
"forbidden_operations": [],
|
||||||
|
"allowed_repositories": list(allowed),
|
||||||
|
}
|
||||||
|
if canonical is not None:
|
||||||
|
p["canonical_repository_root"] = canonical
|
||||||
|
return p
|
||||||
|
|
||||||
|
|
||||||
|
class _Base(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self._tmp = tempfile.TemporaryDirectory()
|
||||||
|
tmp = Path(self._tmp.name)
|
||||||
|
self.install = _init_repo(tmp / "Gitea-Tools", INSTALL_URL)
|
||||||
|
self.target = _init_repo(tmp / "mcp-control-plane", TARGET_URL)
|
||||||
|
(Path(self.target) / "branches").mkdir()
|
||||||
|
self.target_wt = _add_worktree(
|
||||||
|
self.target, Path(self.target) / "branches" / "author-issue-1",
|
||||||
|
"feat/issue-1",
|
||||||
|
)
|
||||||
|
# PROJECT_ROOT and the install git remote are the Gitea-Tools install
|
||||||
|
# checkout — exactly the source that (mis)seeded the session before.
|
||||||
|
self._p_root = mock.patch.object(mcp_server, "PROJECT_ROOT", self.install)
|
||||||
|
self._p_url = mock.patch.object(
|
||||||
|
mcp_server, "_local_git_remote_url", return_value=INSTALL_URL
|
||||||
|
)
|
||||||
|
self._p_root.start()
|
||||||
|
self._p_url.start()
|
||||||
|
session_ctx._reset_session_context_for_testing()
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
mock.patch.stopall()
|
||||||
|
session_ctx._reset_session_context_for_testing()
|
||||||
|
|
||||||
|
def _seed(self, profile, env=None):
|
||||||
|
session_ctx._reset_session_context_for_testing()
|
||||||
|
with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
|
||||||
|
mock.patch.dict(os.environ, env or {}, clear=False):
|
||||||
|
return mcp_server._seed_session_context(
|
||||||
|
profile=profile, remote="prgs", host="gitea.prgs.cc",
|
||||||
|
identity="svc",
|
||||||
|
)
|
||||||
|
|
||||||
|
def _enforce(self, profile, env=None):
|
||||||
|
with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
|
||||||
|
mock.patch.dict(os.environ, env or {}, clear=False):
|
||||||
|
mcp_server._enforce_canonical_repository_root(
|
||||||
|
self.target_wt, remote="prgs"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestF1SeedsConfiguredTargetIdentity(_Base):
|
||||||
|
def test_env_config_seeds_target_not_install(self):
|
||||||
|
ctx = self._seed(
|
||||||
|
_profile("reviewer"),
|
||||||
|
env={crr.CANONICAL_ROOT_ENV: self.target},
|
||||||
|
)
|
||||||
|
self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
|
||||||
|
self.assertEqual(ctx["repository"], "mcp-control-plane")
|
||||||
|
# Regression guard: must NOT be the install repo.
|
||||||
|
self.assertNotEqual(ctx["repository"], "Gitea-Tools")
|
||||||
|
|
||||||
|
def test_profile_field_seeds_target_not_install(self):
|
||||||
|
ctx = self._seed(_profile("merger", canonical=self.target))
|
||||||
|
self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
|
||||||
|
self.assertEqual(ctx["repository"], "mcp-control-plane")
|
||||||
|
self.assertNotEqual(ctx["repository"], "Gitea-Tools")
|
||||||
|
|
||||||
|
def test_enforce_accepts_target_after_seed_reviewer(self):
|
||||||
|
prof = _profile("reviewer", canonical=self.target)
|
||||||
|
self._seed(prof)
|
||||||
|
# Would raise RuntimeError on the self-inflicted identity mismatch
|
||||||
|
# before the fix.
|
||||||
|
self._enforce(prof)
|
||||||
|
|
||||||
|
def test_enforce_accepts_target_after_seed_merger(self):
|
||||||
|
prof = _profile("merger", canonical=self.target)
|
||||||
|
self._seed(prof)
|
||||||
|
self._enforce(prof)
|
||||||
|
|
||||||
|
def test_env_overrides_profile_for_seed(self):
|
||||||
|
# profile points at install; env points at the real target → env wins.
|
||||||
|
prof = _profile("reviewer", canonical=self.install,
|
||||||
|
allowed=(TARGET_SLUG,))
|
||||||
|
ctx = self._seed(prof, env={crr.CANONICAL_ROOT_ENV: self.target})
|
||||||
|
self.assertEqual(ctx["repository"], "mcp-control-plane")
|
||||||
|
|
||||||
|
|
||||||
|
class TestUnconfiguredDefaultUnchanged(_Base):
|
||||||
|
def test_unconfigured_keeps_install_identity(self):
|
||||||
|
prof = _profile("author", allowed=("Scaled-Tech-Consulting/Gitea-Tools",))
|
||||||
|
ctx = self._seed(prof) # no env, no profile canonical field
|
||||||
|
self.assertEqual(ctx["repository"], "Gitea-Tools")
|
||||||
|
self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
|
||||||
|
|
||||||
|
|
||||||
|
class TestFailClosed(_Base):
|
||||||
|
def _trusted(self, profile, env=None, *, for_mutation=True):
|
||||||
|
with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
|
||||||
|
mock.patch.dict(os.environ, env or {}, clear=False):
|
||||||
|
return mcp_server._trusted_session_repository(
|
||||||
|
profile, "prgs", for_mutation=for_mutation
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_nonexistent_configured_root_fails_closed(self):
|
||||||
|
res = self._trusted(
|
||||||
|
_profile("reviewer"),
|
||||||
|
env={crr.CANONICAL_ROOT_ENV: self.target + "-missing"},
|
||||||
|
)
|
||||||
|
self.assertIsNone(res["repository"])
|
||||||
|
self.assertTrue(res["reasons"])
|
||||||
|
|
||||||
|
def test_non_git_configured_root_fails_closed(self):
|
||||||
|
plain = Path(self._tmp.name) / "plain"
|
||||||
|
plain.mkdir()
|
||||||
|
res = self._trusted(
|
||||||
|
_profile("reviewer"), env={crr.CANONICAL_ROOT_ENV: str(plain)}
|
||||||
|
)
|
||||||
|
self.assertIsNone(res["repository"])
|
||||||
|
self.assertTrue(any("git" in r for r in res["reasons"]))
|
||||||
|
|
||||||
|
def test_unallowlisted_target_identity_fails_closed(self):
|
||||||
|
# Configured target is valid, but the profile does not authorize it.
|
||||||
|
res = self._trusted(
|
||||||
|
_profile("reviewer", canonical=self.target,
|
||||||
|
allowed=("Scaled-Tech-Consulting/Gitea-Tools",)),
|
||||||
|
)
|
||||||
|
self.assertIsNone(res["repository"])
|
||||||
|
self.assertTrue(any("scope" in r.lower() for r in res["reasons"]))
|
||||||
|
|
||||||
|
def test_forged_identity_conflict_fails_closed_at_enforce(self):
|
||||||
|
# Seed the target, then present a *different* configured root at
|
||||||
|
# enforcement time: the session pin no longer matches → fail closed.
|
||||||
|
other = _init_repo(Path(self._tmp.name) / "other", OTHER_URL)
|
||||||
|
prof_seed = _profile("reviewer", canonical=self.target)
|
||||||
|
self._seed(prof_seed)
|
||||||
|
prof_drift = _profile(
|
||||||
|
"reviewer", canonical=other,
|
||||||
|
allowed=(TARGET_SLUG, "Scaled-Tech-Consulting/other-repo"),
|
||||||
|
)
|
||||||
|
with self.assertRaises(RuntimeError):
|
||||||
|
self._enforce(prof_drift)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user