From 0d8a2c2b1d2b61321b9f5597c2a108291d4df4fb Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 17 Jul 2026 23:31:39 -0400 Subject: [PATCH] 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) --- canonical_repository_root.py | 267 +++++++++++ gitea_auth.py | 4 + gitea_config.py | 27 ++ gitea_mcp_server.py | 84 +++- namespace_workspace_binding.py | 19 +- session_context_binding.py | 18 + ...est_issue_706_canonical_repository_root.py | 423 ++++++++++++++++++ 7 files changed, 839 insertions(+), 3 deletions(-) create mode 100644 canonical_repository_root.py create mode 100644 tests/test_issue_706_canonical_repository_root.py diff --git a/canonical_repository_root.py b/canonical_repository_root.py new file mode 100644 index 0000000..6fde057 --- /dev/null +++ b/canonical_repository_root.py @@ -0,0 +1,267 @@ +"""Immutable canonical repository root for cross-repository namespaces (#706). + +The Gitea-Tools MCP server historically derived the ``canonical_repo_root`` from +the *install checkout* the server script lives in +(``PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))``). A namespace +that runs the same server script against an *external* repository (e.g. +``eagenda-author`` targeting ``eAgenda``) then failed every mutation: the +branches-only / worktree-membership guards (#274) compared the task workspace +against the Gitea-Tools ``.git`` directory, which it can never belong to. + +This module separates two distinct concepts: + +* the immutable code/install root (``PROJECT_ROOT``) — where the server lives, and +* the namespace-scoped **canonical repository root** — the working root of the + repository whose issues/PRs the namespace mutates. + +The canonical repository root is configured per namespace (profile field or an +environment variable, typically set alongside the namespace ``cwd`` in the MCP +config). It is validated (existence, git identity, git common-directory +membership) and pinned immutably into the session context so a later call cannot +forge or swap it. When *no* binding is configured the single-repo default is +preserved unchanged: the canonical root is derived from the process checkout. +""" + +from __future__ import annotations + +import os +import subprocess +from typing import Mapping + +import remote_repo_guard + +# Namespace-scoped override, typically exported next to the server ``cwd`` in the +# MCP config for a cross-repository namespace. +CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT" + +# Candidate git remote names probed when deriving repository identity. +_IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps") + + +def configured_canonical_root( + profile: Mapping | None, + env: Mapping | None, +) -> tuple[str | None, str | None]: + """Return ``(value, source)`` for the declared canonical repository root. + + Precedence: the ``GITEA_CANONICAL_REPOSITORY_ROOT`` environment variable + (namespace-scoped) overrides the profile ``canonical_repository_root`` + field. Blank values are treated as unset. Returns ``(None, None)`` when no + binding is declared (the single-repo default). + """ + env_map = env if env is not None else os.environ + env_val = (env_map.get(CANONICAL_ROOT_ENV) or "").strip() + if env_val: + return env_val, f"{CANONICAL_ROOT_ENV} environment variable" + if profile: + prof_val = (profile.get("canonical_repository_root") or "").strip() + if prof_val: + return prof_val, "profile canonical_repository_root" + return None, None + + +def resolve_repo_toplevel(path: str) -> str | None: + """Realpath of the git working-tree top level for *path*, or None.""" + text = (path or "").strip() + if not text: + return None + try: + res = subprocess.run( + ["git", "-C", text, "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + check=True, + ) + except Exception: + return None + top = (res.stdout or "").strip() + return os.path.realpath(top) if top else None + + +def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None: + """``owner/repository`` derived from a git remote configured at *path*. + + Tries the caller-named remote first, then a small set of known remote names, + then whatever remote the repository actually has. Returns None when no remote + URL is parseable (identity cannot be proven). + """ + text = (path or "").strip() + if not text: + return None + + ordered: list[str] = [] + for name in (remote, *_IDENTITY_REMOTE_CANDIDATES): + clean = (name or "").strip() + if clean and clean not in ordered: + ordered.append(clean) + + try: + listed = subprocess.run( + ["git", "-C", text, "remote"], + capture_output=True, + text=True, + check=True, + ).stdout.split() + except Exception: + listed = [] + for name in listed: + if name and name not in ordered: + ordered.append(name) + + for name in ordered: + try: + url = subprocess.run( + ["git", "-C", text, "remote", "get-url", name], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + except Exception: + continue + parsed = remote_repo_guard.parse_org_repo_from_remote_url(url) + if parsed: + return f"{parsed[0]}/{parsed[1]}" + return None + + +def assess_canonical_repository_root( + *, + configured_value: str | None, + source: str | None, + expected_slug: str | None, + process_project_root: str, + remote: str | None = None, + require_binding: bool = False, +) -> dict: + """Validate the canonical repository root binding, failing closed on forgery. + + Returns a dict with ``proven`` / ``block`` / ``reasons`` plus the resolved + ``canonical_repo_root`` (the value downstream guards must use), + ``configured`` (whether a cross-repo binding was declared), + ``resolved_slug`` and ``source``. + + Without a configured binding the single-repo default is preserved: the + canonical root is derived from *process_project_root* and never blocks + (unless *require_binding* explicitly demands one). + + With a configured binding the path must exist, be a git repository, and — + when *expected_slug* is known — carry a matching repository identity. A + mismatched or (when *require_binding*) unprovable identity is a forged or + conflicting binding and fails closed. + """ + process_root = os.path.realpath(process_project_root) + declared = (configured_value or "").strip() + + if not declared: + if require_binding: + return _assessment( + proven=False, + reasons=[ + "no canonical_repository_root configured for a cross-repository " + f"namespace; set {CANONICAL_ROOT_ENV} or the profile " + "canonical_repository_root field (fail closed)" + ], + configured=False, + canonical_repo_root=process_root, + resolved_slug=None, + source=None, + ) + # Single-repo default: canonical root follows the install checkout. + derived = resolve_repo_toplevel(process_root) or process_root + return _assessment( + proven=True, + reasons=[], + configured=False, + canonical_repo_root=derived, + resolved_slug=None, + source=None, + ) + + real = os.path.realpath(os.path.abspath(declared)) + if not os.path.isdir(real): + return _assessment( + proven=False, + reasons=[ + f"configured canonical repository root '{real}' does not exist " + "or is not a directory (fail closed)" + ], + configured=True, + canonical_repo_root=real, + resolved_slug=None, + source=source, + ) + + toplevel = resolve_repo_toplevel(real) + if not toplevel: + return _assessment( + proven=False, + reasons=[ + f"configured canonical repository root '{real}' is not a git " + "repository (fail closed)" + ], + configured=True, + canonical_repo_root=real, + resolved_slug=None, + source=source, + ) + + resolved_slug = repository_identity_slug(toplevel, remote=remote) + reasons: list[str] = [] + expected = (expected_slug or "").strip() or None + if expected: + if resolved_slug and resolved_slug.lower() != expected.lower(): + reasons.append( + f"canonical repository root identity mismatch: '{toplevel}' resolves " + f"to repository '{resolved_slug}' but the session is authorized for " + f"'{expected}' (forged or conflicting binding, fail closed)" + ) + elif not resolved_slug and require_binding: + reasons.append( + f"canonical repository root '{toplevel}' has no resolvable git " + f"remote identity to confirm authorization for '{expected}' " + "(fail closed)" + ) + + return _assessment( + proven=not reasons, + reasons=reasons, + configured=True, + canonical_repo_root=toplevel, + resolved_slug=resolved_slug, + source=source, + ) + + +def format_canonical_repository_root_error(assessment: Mapping) -> str: + """Single RuntimeError message for MCP preflight gates.""" + root = assessment.get("canonical_repo_root") or "(unknown)" + source = assessment.get("source") or "(unconfigured)" + reasons = "; ".join( + assessment.get("reasons") or ["unknown canonical repository root violation"] + ) + return ( + f"Canonical repository root guard (#706): {reasons}. " + f"binding source: {source}; canonical repository root: {root}. " + "Configure a valid canonical_repository_root for the target repository " + "and relaunch; do not point it at the Gitea-Tools install checkout." + ) + + +def _assessment( + *, + proven: bool, + reasons: list[str], + configured: bool, + canonical_repo_root: str, + resolved_slug: str | None, + source: str | None, +) -> dict: + return { + "proven": proven, + "block": not proven, + "reasons": list(reasons), + "configured": configured, + "canonical_repo_root": canonical_repo_root, + "resolved_slug": resolved_slug, + "source": source, + } diff --git a/gitea_auth.py b/gitea_auth.py index 88bd878..ef7caac 100644 --- a/gitea_auth.py +++ b/gitea_auth.py @@ -802,6 +802,10 @@ def get_profile(): # environment variable must never widen or forge the set of # repositories a session may bind to. "allowed_repositories": _json_list("allowed_repositories"), + # #706 cross-repository canonical root binding. Config-sourced here (the + # namespace-scoped GITEA_CANONICAL_REPOSITORY_ROOT env override is applied + # by canonical_repository_root.configured_canonical_root, not widened here). + "canonical_repository_root": jp.get("canonical_repository_root") or None, "audit_label": audit_label, "token_source_name": token_source, "auth_source_type": auth_type, diff --git a/gitea_config.py b/gitea_config.py index 3e0adef..b83056c 100644 --- a/gitea_config.py +++ b/gitea_config.py @@ -502,6 +502,30 @@ def _validate_allowed_repositories(name, raw): ) +def _validate_canonical_repository_root(name, raw): + """Validate the optional per-profile canonical repository root (#706). + + ``canonical_repository_root`` binds a cross-repository namespace to the + working root of its target repository (separate from the immutable + Gitea-Tools install checkout). It is an absolute filesystem path; existence + and git identity are validated at bind time by the runtime guard, not here + (config validation stays filesystem-independent). Absent means the + single-repo default and is allowed. + """ + if raw is None: + return + if not isinstance(raw, str) or not raw.strip(): + raise ConfigError( + f"profile '{name}' canonical_repository_root must be a non-empty " + "absolute path string to the target repository working root" + ) + if not os.path.isabs(raw.strip()): + raise ConfigError( + f"profile '{name}' canonical_repository_root {raw!r} must be an " + "absolute path" + ) + + def _reject_inline_secrets(kind, name, obj): for key in _INLINE_SECRET_KEYS: if key in obj: @@ -577,6 +601,9 @@ def _load_v2_contexts(data, path): if not isinstance(allowed, list) or not isinstance(forbidden, list): raise ConfigError(f"profile '{name}' operation fields must be lists") _validate_allowed_repositories(name, raw.get("allowed_repositories")) + _validate_canonical_repository_root( + name, raw.get("canonical_repository_root") + ) allowed_n = {_normalize_op("gitea", op, name) for op in allowed} forbidden_n = {_normalize_op("gitea", op, name) for op in forbidden} # Reviewer-identity deadlock rule (#100/#103) applies here unchanged. diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 01d016d..f9eeb9c 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -194,6 +194,7 @@ MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE" RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE" import namespace_workspace_binding as nwb # noqa: E402 +import canonical_repository_root as crr # noqa: E402 # #706 cross-repo canonical root import mcp_namespace_health # noqa: E402 import stale_binding_recovery # noqa: E402 @@ -376,6 +377,22 @@ def _assess_stale_active_binding(auto_recover: bool = False) -> dict: return report +def _configured_canonical_root() -> tuple[str | None, str | None]: + """Declared cross-repository canonical root for the active namespace (#706). + + Returns ``(value, source)`` from the profile/env binding, or ``(None, None)`` + for the single-repo default. The raw declared value is threaded into + workspace resolution so the branches-only / membership guards evaluate + against the configured target repository; the strict identity/existence + checks run separately in :func:`_enforce_canonical_repository_root`. + """ + try: + profile = get_profile() + except Exception: + profile = {} + return crr.configured_canonical_root(profile, os.environ) + + def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: """Resolve the namespace-scoped workspace root inspected by pre-flight guards. @@ -399,8 +416,9 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict: - """Canonical namespace workspace + repository root for guards (#460/#510).""" + """Canonical namespace workspace + repository root for guards (#460/#510/#706).""" role = _effective_workspace_role() + configured_root, _source = _configured_canonical_root() return nwb.resolve_namespace_mutation_context( role_kind=role, worktree_path=worktree_path, @@ -409,6 +427,7 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic _reviewer_session_worktree() if role in {"reviewer", "merger"} else None ), profile_name=get_profile().get("profile_name"), + configured_canonical_root=configured_root, ) @@ -721,6 +740,54 @@ def record_preflight_check( _preflight_resolved_task = resolved_task +def _enforce_canonical_repository_root( + worktree_path: str | None = None, + *, + remote: str | None = None, +) -> None: + """#706: validate the immutable cross-repository canonical root binding. + + A no-op for the single-repo default (no configured binding). When a + namespace declares a ``canonical_repository_root`` (profile/env), the + configured target repository must exist, be a git repository, and carry a + repository identity matching the session's authorized repository. Missing, + conflicting, or forged bindings fail closed, and the validated root is + checked against the immutable session pin so it cannot be swapped mid + session. + """ + configured_value, source = _configured_canonical_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") + ) + assessment = crr.assess_canonical_repository_root( + configured_value=configured_value, + source=source, + expected_slug=expected_slug, + process_project_root=PROJECT_ROOT, + remote=remote, + require_binding=True, + ) + if assessment["block"]: + raise RuntimeError( + crr.format_canonical_repository_root_error(assessment) + ) + + drift = session_ctx.assess_session_context( + profile_name=get_profile().get("profile_name"), + remote=remote, + canonical_repository_root=assessment["canonical_repo_root"], + ) + if drift["block"]: + raise RuntimeError( + "Canonical repository root guard (#706): " + + "; ".join(drift["reasons"]) + ) + + def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None: """#274: author file/branch mutations must run from a branches/ worktree. @@ -1069,6 +1136,7 @@ def verify_preflight_purity( ) # Historical path: root + branches after purity-order when dirty paths live. + _enforce_canonical_repository_root(worktree_path, remote=remote) _enforce_root_checkout_guard(worktree_path) _enforce_branches_only_author_mutation(worktree_path) _enforce_issue_scope_guard( @@ -1102,6 +1170,7 @@ def verify_preflight_purity( # #683: under pytest unit isolation, FORCE_PRODUCTION_GUARDS still runs # production root + branches + issue scope (no silent no-op of guards). if production_active: + _enforce_canonical_repository_root(worktree_path, remote=remote) _enforce_root_checkout_guard(worktree_path) _enforce_branches_only_author_mutation(worktree_path) _enforce_issue_scope_guard( @@ -1598,6 +1667,18 @@ def _seed_session_context( """ expected = (profile.get("username") or "").strip() or None trusted = _trusted_session_repository(profile, remote) + # #706: pin the configured cross-repository canonical root immutably so a + # later call cannot forge/swap it. Store the resolved toplevel so drift + # checks compare against the same value the mutation gate validates. + configured_value, _crr_source = crr.configured_canonical_root( + profile, os.environ + ) + canonical_root_pin = None + if configured_value: + canonical_root_pin = ( + crr.resolve_repo_toplevel(configured_value) + or os.path.realpath(configured_value) + ) return session_ctx.seed_session_context_if_unbound( profile_name=profile.get("profile_name") or "", remote=remote, @@ -1608,6 +1689,7 @@ def _seed_session_context( role_kind=_profile_role_kind(profile), expected_username=expected, source=source, + canonical_repository_root=canonical_root_pin, ) import issue_work_duplicate_gate # noqa: E402 import issue_workflow_labels # noqa: E402 diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py index 3f44f3d..8ad63a2 100644 --- a/namespace_workspace_binding.py +++ b/namespace_workspace_binding.py @@ -109,8 +109,17 @@ def resolve_namespace_mutation_context( session_lease_worktree: str | None = None, worktree: str | None = None, profile_name: str | None = None, + configured_canonical_root: str | None = None, ) -> dict: - """Shared workspace resolution for runtime_context and mutation guards.""" + """Shared workspace resolution for runtime_context and mutation guards. + + When *configured_canonical_root* is supplied (a cross-repository namespace + bound to an external target repository, #706), the canonical repository root + is that configured target rather than the MCP install checkout. This keeps + the branches-only / worktree-membership guards (#274) evaluating against the + repository the namespace actually mutates. Without it the single-repo + default is preserved: the canonical root follows the process checkout. + """ demotions: list[str] = [] workspace, binding_source = resolve_namespace_workspace( role_kind=role_kind, @@ -132,7 +141,11 @@ def resolve_namespace_mutation_context( env=env, profile_name=profile_name, ) - canonical_root = amw.resolve_canonical_repo_root(process_root, process_root) + configured = (configured_canonical_root or "").strip() + if configured: + canonical_root = os.path.realpath(configured) + else: + canonical_root = amw.resolve_canonical_repo_root(process_root, process_root) return { "workspace_path": workspace, "workspace_binding_source": binding_source, @@ -258,6 +271,7 @@ def assess_namespace_mutation_workspace( session_lease_worktree: str | None = None, profile_name: str | None = None, current_branch: str | None = None, + configured_canonical_root: str | None = None, ) -> dict: """Evaluate namespace workspace binding before preflight/mutation.""" ctx = resolve_namespace_mutation_context( @@ -268,6 +282,7 @@ def assess_namespace_mutation_workspace( env=env, session_lease_worktree=session_lease_worktree, profile_name=profile_name, + configured_canonical_root=configured_canonical_root, ) mutation_workspace = ctx["workspace_path"] binding_source = ctx["workspace_binding_source"] diff --git a/session_context_binding.py b/session_context_binding.py index e598ea6..4005c48 100644 --- a/session_context_binding.py +++ b/session_context_binding.py @@ -32,6 +32,7 @@ class _SessionContext: expected_username: str | None source: str pid: int + canonical_repository_root: str | None = None def as_dict(self) -> dict[str, Any]: return { @@ -45,6 +46,7 @@ class _SessionContext: "expected_username": self.expected_username, "source": self.source, "pid": self.pid, + "canonical_repository_root": self.canonical_repository_root, } @@ -143,6 +145,7 @@ def bind_session_context( role_kind: str | None = None, expected_username: str | None = None, source: str = "bind", + canonical_repository_root: str | None = None, ) -> dict[str, Any]: """Atomically bind/re-bind context (the explicit activation path).""" with _SESSION_CONTEXT_LOCK: @@ -156,6 +159,7 @@ def bind_session_context( role_kind=role_kind, expected_username=expected_username, source=source, + canonical_repository_root=canonical_repository_root, ) @@ -170,6 +174,7 @@ def _bind_session_context_unlocked( role_kind: str | None, expected_username: str | None, source: str, + canonical_repository_root: str | None = None, ) -> dict[str, Any]: """Store a complete immutable context while the caller holds the lock.""" global _SESSION_CONTEXT @@ -184,6 +189,7 @@ def _bind_session_context_unlocked( expected_username=(expected_username or "").strip() or None, source=source, pid=os.getpid(), + canonical_repository_root=(canonical_repository_root or "").strip() or None, ) return _SESSION_CONTEXT.as_dict() @@ -199,6 +205,7 @@ def seed_session_context_if_unbound( role_kind: str | None = None, expected_username: str | None = None, source: str = "seed", + canonical_repository_root: str | None = None, ) -> dict[str, Any]: """Atomically bind only when this process has no current context. @@ -219,6 +226,7 @@ def seed_session_context_if_unbound( role_kind=role_kind, expected_username=expected_username, source=source, + canonical_repository_root=canonical_repository_root, ) return _SESSION_CONTEXT.as_dict() @@ -232,6 +240,7 @@ def assess_session_context( repository: str | None = None, org: str | None = None, expected_username: str | None = None, + canonical_repository_root: str | None = None, require_bound: bool = False, require_complete: bool = False, ) -> dict[str, Any]: @@ -272,6 +281,7 @@ def assess_session_context( live_identity = (identity or "").strip() or None live_repo = (repository or "").strip() or None live_org = (org or "").strip() or None + live_canonical = (canonical_repository_root or "").strip() or None if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"): reasons.append( @@ -307,6 +317,13 @@ def assess_session_context( f"org drift: live '{live_org}' != bound " f"'{ctx.get('org')}' (fail closed)" ) + bound_canonical = ctx.get("canonical_repository_root") + if bound_canonical and live_canonical and live_canonical != bound_canonical: + reasons.append( + f"canonical repository root drift: live '{live_canonical}' != bound " + f"'{bound_canonical}' (forged or conflicting cross-repository " + "binding, fail closed)" + ) expected = expected_username or ctx.get("expected_username") if expected and live_identity and live_identity != expected: @@ -580,6 +597,7 @@ def mutation_context_audit_fields( "session_org": data.get("org"), "session_role_kind": data.get("role_kind"), "session_context_source": data.get("source"), + "session_canonical_repository_root": data.get("canonical_repository_root"), } diff --git a/tests/test_issue_706_canonical_repository_root.py b/tests/test_issue_706_canonical_repository_root.py new file mode 100644 index 0000000..a51004f --- /dev/null +++ b/tests/test_issue_706_canonical_repository_root.py @@ -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", "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)) + + +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()