diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 99f355c..1ee4c8a 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1648,6 +1648,45 @@ def _workspace_repository_slug(remote: str | None) -> str | None: return session_ctx.format_repository_slug(parsed[0], parsed[1]) +def _canonical_repository_slug( + profile: dict, + remote: str | None, +) -> tuple[str | None, list[str]]: + """Repository identity of the *configured* canonical root, if any (#739 F2). + + Returns ``(slug, reasons)``. A namespace with no configured + ``canonical_repository_root`` yields ``(None, [])`` so callers keep the + install-derived single-repository default. A configured root that cannot be + resolved to a repository identity yields ``(None, reasons)`` — a fail-closed + signal, never a silent fallback to the installation checkout, because that + fallback is exactly what lets a cross-repository namespace validate against + the wrong repository. + + Env-over-profile precedence and the existence / git-toplevel / resolvable + remote-identity validation all live in ``crr``. + """ + configured_value, configured_source = crr.configured_canonical_root( + profile, os.environ + ) + if not configured_value: + return None, [] + 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 None, list(assessment.get("reasons") or [ + "configured canonical repository root has no resolvable " + "repository identity (fail closed)" + ]) + return slug, [] + + def _trusted_session_repository( profile: dict, remote: str | None, @@ -1686,32 +1725,10 @@ def _trusted_session_repository( # 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) + canonical_slug, canonical_reasons = _canonical_repository_slug(profile, remote) + if canonical_reasons: + return {"org": None, "repository": None, "reasons": canonical_reasons} + slug = canonical_slug or _workspace_repository_slug(remote) scope = session_ctx.assess_repository_scope( workspace_slug=slug, allowed=allowed, @@ -8768,8 +8785,28 @@ def _delete_branch_repository_binding_block( (unproven target). Returns a structured block dict, or ``None`` when the target is authorized. + + #739 F2: the trusted identity is the session's configured + ``canonical_repository_root`` when the namespace declares one. Deriving it + from ``_workspace_repository_slug`` alone reads the *installation* checkout + (``_local_git_remote_url`` runs in ``PROJECT_ROOT``, always Gitea-Tools), + which inverts the guard for a cross-repository namespace: the genuinely + bound target is rejected and Gitea-Tools is accepted. A configured root that + cannot be resolved fails closed rather than falling back to the install + identity; unconfigured namespaces keep the install-derived default. """ - workspace_slug = _workspace_repository_slug(remote) + canonical_slug, canonical_reasons = _canonical_repository_slug( + get_profile(), remote + ) + if canonical_reasons: + return { + "success": False, + "performed": False, + "required_permission": "gitea.branch.delete", + "reasons": canonical_reasons, + "blocker_kind": "repository_binding", + } + workspace_slug = canonical_slug or _workspace_repository_slug(remote) workspace_parts = ( session_ctx.parse_repository_slug(workspace_slug) if workspace_slug else None ) @@ -14036,6 +14073,16 @@ def gitea_get_runtime_context( remote: Known instance — 'dadeschools' or 'prgs'. host: Override the Gitea host. """ + # #739 F1: normalize the remote *before* any host lookup, identity + # resolution, or session seeding. ``remote`` defaults to 'dadeschools' + # across the whole tool surface, so a caller that omits it on a prgs-hosted + # namespace would otherwise pin the argument default: the identity lookup + # below would authenticate against the wrong host, and because first-bind is + # first-write-wins a later, correct ``gitea_whoami(remote="prgs")`` could not + # repair the binding. ``gitea_whoami`` has always normalized here; this is + # the same call, not a new policy. Explicit non-default remotes pass through + # untouched, and a dadeschools-hosted profile still resolves to dadeschools. + remote = _effective_remote(remote) profile = get_profile() config = gitea_config.load_config() reveal = _reveal_endpoints() @@ -14213,13 +14260,23 @@ def gitea_assess_master_parity( Never mutates and makes no network calls. Read-only operations are never blocked by staleness; only mutating operations fail closed while stale. + The top-level ``startup_head``/``current_head``/``in_parity`` fields keep + their original meaning: the *Gitea-Tools server implementation* only. #739 + F3 adds two separately labelled dimensions so a cross-repository namespace + can tell them apart — ``server_implementation`` (this installation, restated + under an explicit name) and ``target_repository`` (the configured canonical + target checkout and its last-known remote master). Only the server dimension + gates mutations. + Returns: dict with 'in_parity', 'stale', 'restart_required', 'startup_head', - 'current_head', 'mutation_gate_enforced', 'summary', 'reasons', and a - 'report' recovery payload when stale. + 'current_head', 'mutation_gate_enforced', 'summary', 'reasons', + 'server_implementation', 'target_repository', and a 'report' recovery + payload when stale. """ parity = _current_master_parity() enforced = not master_parity_gate.gate_disabled() + canonical_root, canonical_source = _configured_canonical_root() out = { "in_parity": parity["in_parity"], "stale": parity["stale"], @@ -14231,6 +14288,17 @@ def gitea_assess_master_parity( "summary": master_parity_gate.format_parity(parity), "reasons": parity["reasons"], "process_root": PROJECT_ROOT, + "server_implementation": { + "installation_root": PROJECT_ROOT, + "startup_head": parity["startup_head"], + "current_head": parity["current_head"], + "stale": parity["stale"], + "determinable": parity["determinable"], + }, + "target_repository": master_parity_gate.assess_target_repository_parity( + canonical_root=canonical_root, + source=canonical_source, + ), } if parity["stale"] and enforced: out["report"] = master_parity_gate.parity_report(parity) diff --git a/master_parity_gate.py b/master_parity_gate.py index efb7314..2656453 100644 --- a/master_parity_gate.py +++ b/master_parity_gate.py @@ -166,3 +166,128 @@ def format_parity(assessment: dict) -> str: if not assessment.get("determinable"): return "parity indeterminate (baseline or current HEAD unknown)" return f"in parity at {_short(assessment.get('current_head'))}" + + +# --------------------------------------------------------------------------- +# Target-repository parity (#739 F3) +# +# Everything above measures ONE dimension: the Gitea-Tools server's own +# implementation commit, comparing the SHA this process was loaded from against +# the SHA now on disk at PROJECT_ROOT. That is deliberate and is left untouched +# — it is what proves the in-memory capability gates are current. +# +# It is not, however, a statement about the repository a cross-repository +# namespace actually mutates. The assessment below is a separate, separately +# labelled dimension for the configured canonical target repository. It never +# feeds the mutation gate and never changes startup_head/current_head. +# --------------------------------------------------------------------------- + +DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master" + + +def _git_capture(root: str, *args: str) -> str | None: + """Run a read-only git command in *root*; ``None`` on any failure. + + Deliberately does not honour ``GITEA_TEST_CURRENT_HEAD``: that override + exists to pin the *server's* HEAD, and applying it here would make a target + repository silently report the server's forced SHA. + """ + if not root: + return None + try: + res = subprocess.run( + ["git", "-C", root, *args], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return None + if res.returncode != 0: + return None + return (res.stdout or "").strip() or None + + +def assess_target_repository_parity( + *, + canonical_root: str | None, + source: str | None, + tracking_ref: str = DEFAULT_TARGET_TRACKING_REF, +) -> dict: + """Assess the configured cross-repository target checkout. + + Reports the target's canonical root, repository identity, checked-out + commit, and last-known remote master commit, plus whether the checkout is + behind that ref. No network call is made: the remote side is read from the + existing remote-tracking ref, so a target that has never been fetched is + reported as indeterminate rather than guessed at. + + An unconfigured namespace is ``configured=False`` and never ``stale`` — the + single-repository default has no second dimension to be stale about. A + configured root that cannot be read is ``determinable=False`` with reasons. + """ + result = { + "configured": bool(canonical_root), + "canonical_repository_root": None, + "source": source, + "repository_slug": None, + "checkout_head": None, + "tracking_ref": tracking_ref, + "remote_tracking_head": None, + "determinable": False, + "stale": False, + "reasons": [], + } + if not canonical_root: + result["determinable"] = True + return result + + result["canonical_repository_root"] = canonical_root + if not os.path.isdir(canonical_root): + result["reasons"].append( + f"configured canonical repository root '{canonical_root}' " + f"does not exist or is not a directory" + ) + return result + + toplevel = _git_capture(canonical_root, "rev-parse", "--show-toplevel") + if not toplevel: + result["reasons"].append( + f"configured canonical repository root '{canonical_root}' " + f"is not a git checkout" + ) + return result + + head = _git_capture(canonical_root, "rev-parse", "HEAD") + if not head: + result["reasons"].append( + f"target repository HEAD could not be read at '{canonical_root}'" + ) + return result + result["checkout_head"] = head + result["determinable"] = True + + remote_url = _git_capture(canonical_root, "remote", "get-url", "origin") + if remote_url: + # Local import keeps this module dependency-light for its startup role. + import remote_repo_guard + + parsed = remote_repo_guard.parse_org_repo_from_remote_url(remote_url) + if parsed: + result["repository_slug"] = f"{parsed[0]}/{parsed[1]}" + if not result["repository_slug"]: + result["reasons"].append( + "target repository identity could not be derived from its git remote" + ) + + tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref) + if not tracking_head: + result["reasons"].append( + f"remote-tracking ref '{tracking_ref}' is unknown in the target " + f"checkout; target staleness is indeterminate (no fetch is " + f"performed by this assessment)" + ) + return result + result["remote_tracking_head"] = tracking_head + result["stale"] = tracking_head != head + return result diff --git a/tests/test_cross_repo_canonical_consumption.py b/tests/test_cross_repo_canonical_consumption.py new file mode 100644 index 0000000..2eaee02 --- /dev/null +++ b/tests/test_cross_repo_canonical_consumption.py @@ -0,0 +1,704 @@ +"""Complete canonical-root consumption for cross-repository MCP namespaces. + +#706 introduced an immutable, configured ``canonical_repository_root`` and +routed the #274 filesystem guards and the session repository *slug* through it. +Three consumption paths were left behind, and this module drives each one +through its production entry point: + +* **A — remote initialization.** ``gitea_get_runtime_context`` never normalized + its ``remote`` argument, while ``gitea_whoami`` did. A fresh process whose + first native call is the runtime-context path therefore pinned the + ``dadeschools`` argument default instead of the profile's configured remote, + and — because first-bind is first-write-wins — no later correct call could + repair it. + +* **C — reconciler branch deletion.** The delete-branch repository-binding + guard derived its expected slug from ``_workspace_repository_slug``, which + reads the git remote of the *installation* checkout (always Gitea-Tools), + rather than from the session's configured canonical root. + +* **D — parity evidence.** ``gitea_assess_master_parity`` proves Gitea-Tools + *server implementation* parity only, which is intentional. It carried no + target-repository dimension at all, so a cross-repository namespace had no + evidence that its target checkout was current. The existing + ``startup_head``/``current_head`` semantics are preserved unchanged and the + target-repository assessment is reported under separately labelled fields. + +Groups B and E assert *intentional* behaviour (the #274 guards already consume +the canonical root; the configuration surface already validates a +repository-specific namespace) so that a regression in either is caught. + +Real git repositories are used throughout; no network calls are made. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import canonical_repository_root as crr # noqa: E402 +import gitea_config # noqa: E402 +import gitea_mcp_server as srv # noqa: E402 +import master_parity_gate # noqa: E402 +import mcp_server # noqa: E402 +import namespace_workspace_binding as nwb # noqa: E402 +import session_context_binding as session_ctx # noqa: E402 + +INSTALL_ORG = "Scaled-Tech-Consulting" +INSTALL_REPO = "Gitea-Tools" +INSTALL_SLUG = f"{INSTALL_ORG}/{INSTALL_REPO}" +INSTALL_URL = f"https://gitea.prgs.cc/{INSTALL_SLUG}.git" + +TARGET_ORG = "Scaled-Tech-Consulting" +TARGET_REPO = "mcp-control-plane" +TARGET_SLUG = f"{TARGET_ORG}/{TARGET_REPO}" +TARGET_URL = f"https://gitea.prgs.cc/{TARGET_SLUG}.git" + + +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: + path.mkdir(parents=True, exist_ok=True) + _git(str(path), "init", "-q") + _git(str(path), "config", "user.email", "t@example.com") + _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)) + + +# --------------------------------------------------------------------------- +# Shared profile/config construction +# --------------------------------------------------------------------------- +_AUTHOR_OPS = [ + "gitea.read", + "gitea.issue.create", + "gitea.issue.comment", + "gitea.pr.create", + "gitea.pr.comment", + "gitea.branch.create", + "gitea.branch.push", + "gitea.repo.commit", +] +_AUTHOR_FORBIDDEN = [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.request_changes", +] +# A profile that may approve or merge must forbid authoring (gitea_config's +# reviewer-identity deadlock rule). +_REVIEWER_OPS = [ + "gitea.read", + "gitea.pr.approve", + "gitea.pr.request_changes", + "gitea.pr.comment", + "gitea.issue.comment", +] +_REVIEWER_FORBIDDEN = ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"] +_MERGER_OPS = ["gitea.read", "gitea.pr.merge", "gitea.pr.comment"] +_MERGER_FORBIDDEN = [ + "gitea.pr.create", + "gitea.branch.push", + "gitea.pr.approve", + "gitea.pr.request_changes", +] +_RECONCILER_OPS = [ + "gitea.read", + "gitea.pr.close", + "gitea.pr.comment", + "gitea.branch.delete", +] +_RECONCILER_FORBIDDEN = [ + "gitea.pr.create", + "gitea.branch.push", + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.request_changes", +] + +_ROLE_MATRIX = { + "author": (_AUTHOR_OPS, _AUTHOR_FORBIDDEN), + "reviewer": (_REVIEWER_OPS, _REVIEWER_FORBIDDEN), + "merger": (_MERGER_OPS, _MERGER_FORBIDDEN), + "reconciler": (_RECONCILER_OPS, _RECONCILER_FORBIDDEN), +} + + +def _profile( + role: str, + *, + canonical_root: str | None = None, + allowed_repositories: list[str] | None = None, + username: str = "jcwalker3", +) -> dict: + allowed, forbidden = _ROLE_MATRIX[role] + profile = { + "enabled": True, + "context": "prgs", + "role": role, + "username": username, + "base_url": "https://gitea.prgs.cc", + "auth": {"type": "env", "name": f"GITEA_TOKEN_PRGS_{role.upper()}"}, + "allowed_operations": list(allowed), + "forbidden_operations": list(forbidden), + "execution_profile": f"prgs-{role}", + } + if canonical_root is not None: + profile["canonical_repository_root"] = canonical_root + if allowed_repositories is not None: + profile["allowed_repositories"] = list(allowed_repositories) + return profile + + +def _config(profiles: dict) -> dict: + return { + "version": 2, + "rules": {"allow_runtime_switching": True}, + "contexts": { + "prgs": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"}, + }, + "mdcps": { + "enabled": True, + "gitea": { + "enabled": True, + "base_url": "https://gitea.dadeschools.net", + }, + }, + }, + "profiles": profiles, + } + + +class _ServerHarness(unittest.TestCase): + """Temp profiles.json + pinned install remote + no network.""" + + def setUp(self): + self._dir = tempfile.TemporaryDirectory() + self.tmp = self._dir.name + self.config_path = os.path.join(self.tmp, "profiles.json") + session_ctx._reset_session_context_for_testing() + gitea_config._active_profile_override = None + mcp_server._IDENTITY_CACHE.clear() + srv._MUTATION_AUTHORITY = None + # The install checkout's remote is Gitea-Tools regardless of the + # developer's layout; pin it so "install-derived" is deterministic. + self._remote_url = patch.object( + srv, "_local_git_remote_url", side_effect=self._install_remote_url + ) + self._remote_url.start() + + def tearDown(self): + self._remote_url.stop() + session_ctx._reset_session_context_for_testing() + gitea_config._active_profile_override = None + mcp_server._IDENTITY_CACHE.clear() + srv._MUTATION_AUTHORITY = None + self._dir.cleanup() + + def _install_remote_url(self, remote_name): + return INSTALL_URL if remote_name in ("prgs", "origin") else None + + def _write_config(self, profiles: dict) -> None: + with open(self.config_path, "w", encoding="utf-8") as fh: + fh.write(json.dumps(_config(profiles))) + + def _env(self, profile_name: str, **extra) -> dict: + env = { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile_name, + "GITEA_TOKEN_PRGS_AUTHOR": "t", + "GITEA_TOKEN_PRGS_REVIEWER": "t", + "GITEA_TOKEN_PRGS_MERGER": "t", + "GITEA_TOKEN_PRGS_RECONCILER": "t", + } + env.update(extra) + return env + + def _api(self, method, url, header): + return { + "login": "jcwalker3", + "full_name": "Test", + "id": 1, + "email": "t@example.com", + } + + def _live(self): + return patch("gitea_mcp_server.api_request", side_effect=self._api) + + +# =========================================================================== +# A. Remote initialization +# =========================================================================== +class TestRemoteInitializationFirstCall(_ServerHarness): + """A prgs namespace must never pin the dadeschools argument default.""" + + def setUp(self): + super().setUp() + self._write_config( + {"prgs-author": _profile("author", allowed_repositories=[INSTALL_SLUG])} + ) + + def test_runtime_context_first_call_does_not_bind_default_remote(self): + """Runtime-context as the FIRST native call, relying on the default.""" + with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live(): + srv.gitea_get_runtime_context() + ctx = session_ctx.get_session_context() + self.assertIsNotNone(ctx, "first call must establish a session binding") + self.assertEqual(ctx["remote"], "prgs") + self.assertEqual(ctx["host"], "gitea.prgs.cc") + self.assertEqual(ctx["org"], INSTALL_ORG) + self.assertEqual(ctx["repository"], INSTALL_REPO) + + def test_runtime_context_first_call_reports_effective_remote(self): + with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live(): + result = srv.gitea_get_runtime_context() + self.assertEqual(result["remote"], "prgs") + + def test_whoami_first_call_remains_correct(self): + """Control: the already-normalizing path is unchanged.""" + with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live(): + srv.gitea_whoami() + ctx = session_ctx.get_session_context() + self.assertEqual(ctx["remote"], "prgs") + self.assertEqual(ctx["host"], "gitea.prgs.cc") + + def test_explicit_remote_argument_still_honoured(self): + """Normalization only fills the default; explicit values are untouched.""" + with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live(): + result = srv.gitea_get_runtime_context(remote="prgs") + self.assertEqual(result["remote"], "prgs") + + def test_mdcps_profile_default_is_not_rewritten_to_prgs(self): + """A dadeschools-hosted profile keeps the dadeschools remote.""" + profile = _profile("author", allowed_repositories=[INSTALL_SLUG]) + profile["context"] = "mdcps" + profile["base_url"] = "https://gitea.dadeschools.net" + self._write_config({"prgs-author": profile}) + with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live(): + result = srv.gitea_get_runtime_context() + self.assertEqual(result["remote"], "dadeschools") + + +# =========================================================================== +# B. Remote/repository guard (intentional behaviour — regression fence) +# =========================================================================== +class TestCanonicalRootGuardBinding(_ServerHarness): + def setUp(self): + super().setUp() + self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL) + self.install_root = _init_repo(Path(self.tmp) / "install", INSTALL_URL) + + def test_canonical_target_root_resolves_to_target_slug(self): + got = crr.assess_canonical_repository_root( + configured_value=self.target_root, + source="profile.canonical_repository_root", + expected_slug=None, + process_project_root=self.install_root, + remote="prgs", + require_binding=True, + ) + self.assertFalse(got.get("block"), got.get("reasons")) + self.assertEqual(got["resolved_slug"], TARGET_SLUG) + + def test_install_root_identity_remains_gitea_tools(self): + got = crr.assess_canonical_repository_root( + configured_value=None, + source=None, + expected_slug=None, + process_project_root=self.install_root, + remote="prgs", + ) + self.assertEqual( + os.path.realpath(got["canonical_repo_root"]), self.install_root + ) + + def test_guards_validate_against_canonical_target_root(self): + ctx = nwb.resolve_namespace_mutation_context( + role_kind="author", + worktree_path=None, + process_project_root=self.install_root, + configured_canonical_root=self.target_root, + ) + self.assertEqual( + os.path.realpath(ctx["canonical_repo_root"]), self.target_root + ) + + def test_explicit_coordinates_cannot_override_canonical_binding(self): + """Explicit org/repo may confirm, never authorize, a binding.""" + confirm = session_ctx.assess_repository_override( + requested_org=TARGET_ORG, + requested_repo=TARGET_REPO, + bound_org=TARGET_ORG, + bound_repo=TARGET_REPO, + ) + self.assertFalse(confirm.get("block")) + override = session_ctx.assess_repository_override( + requested_org=TARGET_ORG, + requested_repo=TARGET_REPO, + bound_org=INSTALL_ORG, + bound_repo=INSTALL_REPO, + ) + self.assertTrue(override.get("block")) + + def test_gitea_tools_rooted_namespace_cannot_reach_target_repository(self): + """No canonical root configured → session stays pinned to Gitea-Tools.""" + profile = _profile("author", allowed_repositories=[INSTALL_SLUG]) + with patch.dict(os.environ, {}, clear=False): + os.environ.pop(crr.CANONICAL_ROOT_ENV, None) + resolved = srv._trusted_session_repository( + profile, "prgs", for_mutation=True + ) + self.assertEqual(resolved["repository"], INSTALL_REPO) + self.assertNotEqual(resolved["repository"], TARGET_REPO) + + def test_target_rooted_namespace_binds_target_repository(self): + profile = _profile( + "author", + canonical_root=self.target_root, + allowed_repositories=[TARGET_SLUG], + ) + resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True) + self.assertEqual(resolved["org"], TARGET_ORG) + self.assertEqual(resolved["repository"], TARGET_REPO) + + +# =========================================================================== +# C. Reconciler branch deletion +# =========================================================================== +class TestDeleteBranchRepositoryBinding(_ServerHarness): + """The binding guard must consult the canonical root, not PROJECT_ROOT. + + No branch is ever deleted here: only the pre-deletion binding guard is + exercised. + """ + + def setUp(self): + super().setUp() + self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL) + self._write_config( + { + "prgs-reconciler": _profile( + "reconciler", + canonical_root=self.target_root, + allowed_repositories=[TARGET_SLUG], + ), + "gt-reconciler": _profile( + "reconciler", allowed_repositories=[INSTALL_SLUG] + ), + } + ) + + def test_target_rooted_reconciler_validates_target_deletion(self): + with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False): + block = srv._delete_branch_repository_binding_block( + "prgs", org=TARGET_ORG, repo=TARGET_REPO + ) + self.assertIsNone(block, block) + + def test_target_rooted_reconciler_fails_closed_for_install_repository(self): + with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False): + block = srv._delete_branch_repository_binding_block( + "prgs", org=INSTALL_ORG, repo=INSTALL_REPO + ) + self.assertIsNotNone(block) + self.assertEqual(block["blocker_kind"], "repository_binding") + self.assertFalse(block["performed"]) + + def test_install_rooted_reconciler_fails_closed_for_target_repository(self): + with patch.dict(os.environ, self._env("gt-reconciler"), clear=False): + os.environ.pop(crr.CANONICAL_ROOT_ENV, None) + block = srv._delete_branch_repository_binding_block( + "prgs", org=TARGET_ORG, repo=TARGET_REPO + ) + self.assertIsNotNone(block) + self.assertEqual(block["blocker_kind"], "repository_binding") + + def test_env_configured_canonical_root_is_honoured(self): + env = self._env("gt-reconciler") + env[crr.CANONICAL_ROOT_ENV] = self.target_root + with patch.dict(os.environ, env, clear=False): + allowed = srv._delete_branch_repository_binding_block( + "prgs", org=TARGET_ORG, repo=TARGET_REPO + ) + blocked = srv._delete_branch_repository_binding_block( + "prgs", org=INSTALL_ORG, repo=INSTALL_REPO + ) + self.assertIsNone(allowed, allowed) + self.assertIsNotNone(blocked) + + def test_unresolvable_canonical_root_fails_closed(self): + env = self._env("gt-reconciler") + env[crr.CANONICAL_ROOT_ENV] = os.path.join(self.tmp, "does-not-exist") + with patch.dict(os.environ, env, clear=False): + block = srv._delete_branch_repository_binding_block( + "prgs", org=TARGET_ORG, repo=TARGET_REPO + ) + self.assertIsNotNone(block) + self.assertEqual(block["blocker_kind"], "repository_binding") + + def test_no_request_parameter_can_override_canonical_identity(self): + """Every explicit coordinate that is not the canonical target is + refused; none of them can *establish* the binding.""" + # Both repositories share an org, so a wrong *org* case must name a + # genuinely different owner to be meaningful. + cases = [ + (INSTALL_ORG, INSTALL_REPO), + (TARGET_ORG, INSTALL_REPO), + (TARGET_ORG, "some-other-repo"), + ("someone-else", TARGET_REPO), + ] + with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False): + for org, repo in cases: + with self.subTest(org=org, repo=repo): + block = srv._delete_branch_repository_binding_block( + "prgs", org=org, repo=repo + ) + self.assertIsNotNone(block, f"{org}/{repo} must fail closed") + + +# =========================================================================== +# D. Parity semantics +# =========================================================================== +class TestTargetRepositoryParityAssessment(unittest.TestCase): + """Server-implementation parity is preserved; target parity is additive.""" + + def setUp(self): + self._dir = tempfile.TemporaryDirectory() + self.tmp = self._dir.name + + def tearDown(self): + self._dir.cleanup() + + def _target(self) -> str: + return _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL) + + def test_unconfigured_target_is_reported_not_stale(self): + got = master_parity_gate.assess_target_repository_parity( + canonical_root=None, source=None + ) + self.assertFalse(got["configured"]) + self.assertFalse(got["stale"]) + self.assertIsNone(got["canonical_repository_root"]) + + def test_configured_target_reports_checkout_head_and_slug(self): + root = self._target() + head = _git(root, "rev-parse", "HEAD") + got = master_parity_gate.assess_target_repository_parity( + canonical_root=root, source="profile.canonical_repository_root" + ) + self.assertTrue(got["configured"]) + self.assertEqual(got["canonical_repository_root"], root) + self.assertEqual(got["checkout_head"], head) + self.assertEqual(got["repository_slug"], TARGET_SLUG) + + def test_target_stale_when_remote_tracking_ref_is_ahead(self): + root = self._target() + head = _git(root, "rev-parse", "HEAD") + # Simulate a fetched remote-tracking ref that has advanced. + _git(root, "checkout", "-q", "-b", "advanced") + (Path(root) / "next.md").write_text("next\n") + _git(root, "add", "next.md") + _git(root, "commit", "-q", "-m", "advance") + advanced = _git(root, "rev-parse", "HEAD") + _git(root, "update-ref", "refs/remotes/origin/master", advanced) + _git(root, "checkout", "-q", "--detach", head) + got = master_parity_gate.assess_target_repository_parity( + canonical_root=root, source="profile.canonical_repository_root" + ) + self.assertEqual(got["checkout_head"], head) + self.assertEqual(got["remote_tracking_head"], advanced) + self.assertTrue(got["stale"]) + + def test_missing_root_is_not_determinable_and_fails_closed(self): + got = master_parity_gate.assess_target_repository_parity( + canonical_root=os.path.join(self.tmp, "absent"), + source="profile.canonical_repository_root", + ) + self.assertTrue(got["configured"]) + self.assertFalse(got["determinable"]) + self.assertTrue(got["reasons"]) + + +class TestParityToolEvidenceDimensions(_ServerHarness): + def setUp(self): + super().setUp() + self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL) + self._write_config( + { + "prgs-author": _profile( + "author", + canonical_root=self.target_root, + allowed_repositories=[TARGET_SLUG], + ) + } + ) + + def test_existing_server_parity_semantics_are_unchanged(self): + with patch.dict(os.environ, self._env("prgs-author"), clear=False): + result = srv.gitea_assess_master_parity(remote="prgs") + # startup_head/current_head keep their Gitea-Tools implementation + # meaning and must not be redefined to the target repository. + self.assertEqual(result["process_root"], srv.PROJECT_ROOT) + self.assertEqual( + result["startup_head"], srv._STARTUP_PARITY.get("startup_head") + ) + self.assertIn("in_parity", result) + + def test_evidence_distinguishes_server_and_target_dimensions(self): + with patch.dict(os.environ, self._env("prgs-author"), clear=False): + result = srv.gitea_assess_master_parity(remote="prgs") + server = result["server_implementation"] + self.assertEqual(server["installation_root"], srv.PROJECT_ROOT) + self.assertEqual(server["current_head"], result["current_head"]) + self.assertIn("stale", server) + + target = result["target_repository"] + self.assertTrue(target["configured"]) + self.assertEqual(target["canonical_repository_root"], self.target_root) + self.assertEqual(target["repository_slug"], TARGET_SLUG) + self.assertEqual( + target["checkout_head"], _git(self.target_root, "rev-parse", "HEAD") + ) + self.assertIn("stale", target) + + def test_unconfigured_namespace_reports_target_as_unconfigured(self): + self._write_config( + {"prgs-author": _profile("author", allowed_repositories=[INSTALL_SLUG])} + ) + env = self._env("prgs-author") + with patch.dict(os.environ, env, clear=False): + os.environ.pop(crr.CANONICAL_ROOT_ENV, None) + result = srv.gitea_assess_master_parity(remote="prgs") + self.assertFalse(result["target_repository"]["configured"]) + + +# =========================================================================== +# E. Startup / configuration validation for a candidate namespace set +# =========================================================================== +class TestCandidateNamespaceConfiguration(_ServerHarness): + """Four repository-specific profiles for the target repository.""" + + def setUp(self): + super().setUp() + self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL) + self.profiles = { + f"mcpcp-{role}": _profile( + role, + canonical_root=self.target_root, + allowed_repositories=[TARGET_SLUG], + ) + for role in ("author", "reviewer", "merger", "reconciler") + } + self._write_config(self.profiles) + + def test_configuration_audit_succeeds(self): + with patch.dict(os.environ, self._env("mcpcp-author"), clear=False): + audit = srv.gitea_audit_config() + self.assertTrue(audit.get("configured")) + names = {row["name"] for row in audit["profiles"]} + self.assertEqual(names, set(self.profiles)) + + def test_no_inline_credentials_are_present(self): + raw = json.loads(Path(self.config_path).read_text()) + for name, profile in raw["profiles"].items(): + with self.subTest(profile=name): + self.assertNotIn("token", profile) + self.assertNotIn("password", profile) + self.assertNotIn("secret", profile) + self.assertIn(profile["auth"]["type"], ("env", "keychain")) + + def test_bind_time_validation_succeeds_for_target_repository(self): + for name, profile in self.profiles.items(): + with self.subTest(profile=name): + resolved = srv._trusted_session_repository( + profile, "prgs", for_mutation=True + ) + self.assertEqual(resolved["reasons"], []) + self.assertEqual(resolved["org"], TARGET_ORG) + self.assertEqual(resolved["repository"], TARGET_REPO) + + def test_wrong_repository_root_fails_closed(self): + wrong_root = _init_repo(Path(self.tmp) / "wrong", INSTALL_URL) + profile = _profile( + "author", canonical_root=wrong_root, allowed_repositories=[TARGET_SLUG] + ) + resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True) + self.assertIsNone(resolved["repository"]) + self.assertTrue(resolved["reasons"]) + + def test_existing_install_profiles_are_unchanged_in_behaviour(self): + profile = _profile("author", allowed_repositories=[INSTALL_SLUG]) + resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True) + self.assertEqual(resolved["org"], INSTALL_ORG) + self.assertEqual(resolved["repository"], INSTALL_REPO) + + def _denied(self, profile: dict, operation: str) -> bool: + ok, _ = gitea_config.check_operation( + operation, + profile["allowed_operations"], + profile["forbidden_operations"], + ) + return not ok + + def test_role_separation_is_enforced(self): + expectations = { + "author": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.request_changes", + ], + "reviewer": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"], + "merger": [ + "gitea.pr.create", + "gitea.branch.push", + "gitea.pr.approve", + "gitea.pr.request_changes", + ], + "reconciler": [ + "gitea.pr.create", + "gitea.branch.push", + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.request_changes", + ], + } + for role, denied_ops in expectations.items(): + profile = self.profiles[f"mcpcp-{role}"] + for op in denied_ops: + with self.subTest(role=role, operation=op): + self.assertTrue( + self._denied(profile, op), + f"{role} must not be permitted {op}", + ) + + def test_each_role_retains_its_own_capability(self): + permitted = { + "author": "gitea.pr.create", + "reviewer": "gitea.pr.approve", + "merger": "gitea.pr.merge", + "reconciler": "gitea.branch.delete", + } + for role, op in permitted.items(): + with self.subTest(role=role, operation=op): + self.assertFalse(self._denied(self.profiles[f"mcpcp-{role}"], op)) + + +if __name__ == "__main__": + unittest.main()