"""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") # Fallback integration-branch names, probed only when the remote publishes no # ``refs/remotes//HEAD`` symbolic ref (#983). Mirrors the stable base # branches recognised elsewhere in the workflow (``stacked_pr_support``). The # fallback is deliberately *not* ordered-first-wins: when more than one of these # refs exists and git records no default, the integration branch is genuinely # ambiguous and resolution fails closed instead of guessing. INTEGRATION_BRANCH_CANDIDATES: tuple[str, ...] = ("master", "main", "dev") # Reason codes for base-ref derivation outcomes (#983), so callers and tests can # assert the refusal cause instead of string-matching prose. BASE_REF_SOURCE_REMOTE_HEAD = "remote_head_symref" BASE_REF_SOURCE_UNIQUE_CANDIDATE = "unique_integration_branch_ref" DENY_NO_IDENTITY_REMOTE = "no_identity_remote" DENY_AMBIGUOUS_REMOTE = "ambiguous_identity_remote" DENY_AMBIGUOUS_BASE_BRANCH = "ambiguous_integration_branch" DENY_NO_BASE_BRANCH = "no_integration_branch_ref" # Repository-authority modes (#973 B10). Exactly two values are supported. # ``mode`` selects how repository authority is established, so an unrecognised # value must never be normalised onto one of these: aliasing a trusted mode is # precisely the defect. Omitting the argument keeps the documented safe default, # ``validation``. MODE_VALIDATION = "validation" MODE_DERIVATION = "derivation" SUPPORTED_MODES: tuple[str, ...] = (MODE_VALIDATION, MODE_DERIVATION) # Reason code emitted when an unsupported mode is refused. Mirrors the existing # ``webui.sanctioned_restart.DENY_UNKNOWN_MODE`` convention so callers and tests # can assert the refusal cause rather than string-matching prose. DENY_UNKNOWN_MODE = "unknown_mode" def unsupported_mode_reason(mode: object) -> str | None: """Precise rejection reason for *mode*, or None when *mode* is supported. Only the two documented string values are accepted, compared exactly — no stripping, no case folding — so misspellings and whitespace variants are refused rather than coerced. An empty string, ``None``, and any non-string are all *explicitly supplied* unsupported values and are refused on the same footing; none of them is normalised to a supported mode. Omitting the argument entirely never reaches here with an unsupported value because the parameter default is ``"validation"``. """ if isinstance(mode, str) and mode in SUPPORTED_MODES: return None supported = ", ".join(repr(m) for m in SUPPORTED_MODES) if not isinstance(mode, str): return ( f"unsupported repository-authority mode {mode!r} of type " f"{type(mode).__name__}: only {supported} are supported; the mode " "is refused before any repository assessment and no repository " "identity was resolved through it (fail closed)" ) return ( f"unsupported repository-authority mode {mode!r}: only {supported} are " "supported; the mode is refused before any repository assessment and no " "repository identity was resolved through it (fail closed)" ) 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 _identity_remote_candidates(path: str, remote: str | None) -> list[str]: """Ordered remote names to probe for identity at *path*. Names are used verbatim — never case-folded. Git config subsection names are case-sensitive, so a repository whose remote is ``MDCPS`` is reached only by the exact string ``MDCPS``; the lowercase entry in :data:`_IDENTITY_REMOTE_CANDIDATES` simply does not resolve, and the exact name arrives from ``git remote`` below (#983). """ 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", path, "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) return ordered def _configured_remote_identities(path: str, remote: str | None) -> list[tuple[str, str]]: """``(remote_name, owner/repository)`` for every probe name that resolves. Remote names are returned exactly as configured so downstream tracking refs (``refs/remotes//``) address the real ref (#983). """ found: list[tuple[str, str]] = [] for name in _identity_remote_candidates(path, remote): try: url = subprocess.run( ["git", "-C", path, "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: found.append((name, f"{parsed[0]}/{parsed[1]}")) return found def resolve_identity_remote( path: str, *, remote: str | None = None ) -> tuple[str | None, str | None]: """``(remote_name, owner/repository)`` for the remote that proves identity. The remote *name* is the piece historically thrown away by :func:`repository_identity_slug`, even though resolving the slug already required discovering it. Cross-repository base-ref derivation needs that name to build ``refs/remotes//``, so it is now returned rather than discarded (#983). Returns ``(None, None)`` when no remote URL is parseable (identity cannot be proven). """ text = (path or "").strip() if not text: return None, None for name, slug in _configured_remote_identities(text, remote): return name, slug return None, 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). """ return resolve_identity_remote(path, remote=remote)[1] def _base_ref_result( *, proven: bool, reasons: list[str], remote: str | None = None, branch: str | None = None, repository_slug: str | None = None, source: str | None = None, reason_code: str | None = None, ) -> dict: """Build the base-ref derivation payload. ``tracking_refs`` is the ordered probe tuple downstream guards hand to ``git rev-parse``: the ``/`` shorthand first, then the fully qualified ``refs/remotes//``. It is empty whenever the derivation is not ``proven``, so an unresolved target can never be probed against some other repository's ref. """ tracking_ref = f"refs/remotes/{remote}/{branch}" if proven and remote and branch else None tracking_refs: tuple[str, ...] = ( (f"{remote}/{branch}", tracking_ref) if tracking_ref else () ) return { "proven": proven, "block": not proven, "remote": remote, "branch": branch, "repository_slug": repository_slug, "tracking_ref": tracking_ref, "tracking_refs": tracking_refs, "source": source, "reason_code": reason_code, "reasons": list(reasons), } def _ref_exists(path: str, ref: str) -> bool: """Whether *ref* resolves in the checkout at *path*.""" try: res = subprocess.run( ["git", "-C", path, "rev-parse", "--verify", "--quiet", ref], capture_output=True, text=True, check=False, ) except Exception: return False return res.returncode == 0 and bool((res.stdout or "").strip()) def resolve_target_base_ref(path: str, *, remote: str | None = None) -> dict: """Derive the authoritative integration base ref for the checkout at *path*. This is the single resolved target shared by cross-repository mutation gating and parity reporting, so the two can never disagree about which commit a checkout is supposed to match (#983). Resolution order: 1. The identity remote — the remote that already proves repository identity via :func:`resolve_identity_remote`, with its **exact configured case** preserved (``MDCPS`` stays ``MDCPS``). 2. The integration branch, from ``refs/remotes//HEAD`` — git's own record of that remote's default branch, written by ``clone`` and ``remote set-head``. This is authoritative repository state, which is why no new configuration field is required. 3. Only if the remote publishes no such symbolic ref, exactly one of :data:`INTEGRATION_BRANCH_CANDIDATES` present as a remote-tracking ref. Fails closed — ``proven`` False, empty ``tracking_refs``, and a ``reason_code`` — when identity is unprovable, when distinct remotes claim different repositories, when several candidate branches exist with no recorded default, or when no candidate exists at all. Nothing here invents a branch, writes a ref, or falls back to another repository's base. """ text = (path or "").strip() if not text: return _base_ref_result( proven=False, reasons=["no repository path supplied for base-ref derivation (fail closed)"], reason_code=DENY_NO_IDENTITY_REMOTE, ) identities = _configured_remote_identities(text, remote) if not identities: return _base_ref_result( proven=False, reasons=[ f"no git remote at '{text}' yields a parseable repository identity, so " "the integration base ref cannot be derived (fail closed)" ], reason_code=DENY_NO_IDENTITY_REMOTE, ) # Ambiguity only matters when the caller named no remote: if distinct remotes # describe different repositories there is no single authoritative target, # and picking the first would silently gate against the wrong repository. if not (remote or "").strip(): distinct = {slug for _, slug in identities} if len(distinct) > 1: listed = ", ".join(f"{name} -> {slug}" for name, slug in identities) return _base_ref_result( proven=False, reasons=[ f"ambiguous repository identity at '{text}': remotes resolve to " f"different repositories ({listed}); no single authoritative " "integration base ref can be derived (fail closed)" ], reason_code=DENY_AMBIGUOUS_REMOTE, ) remote_name, slug = identities[0] symref = None try: res = subprocess.run( ["git", "-C", text, "symbolic-ref", "--quiet", f"refs/remotes/{remote_name}/HEAD"], capture_output=True, text=True, check=False, ) if res.returncode == 0: symref = (res.stdout or "").strip() or None except Exception: symref = None prefix = f"refs/remotes/{remote_name}/" if symref and symref.startswith(prefix): branch = symref[len(prefix):].strip() if branch and branch != "HEAD": return _base_ref_result( proven=True, reasons=[], remote=remote_name, branch=branch, repository_slug=slug, source=BASE_REF_SOURCE_REMOTE_HEAD, ) present = [ candidate for candidate in INTEGRATION_BRANCH_CANDIDATES if _ref_exists(text, f"{prefix}{candidate}") ] if len(present) == 1: return _base_ref_result( proven=True, reasons=[], remote=remote_name, branch=present[0], repository_slug=slug, source=BASE_REF_SOURCE_UNIQUE_CANDIDATE, ) if len(present) > 1: return _base_ref_result( proven=False, reasons=[ f"remote '{remote_name}' publishes no '{prefix}HEAD' default and " f"several integration branches exist ({', '.join(present)}); the " "integration base ref is ambiguous (fail closed)" ], reason_code=DENY_AMBIGUOUS_BASE_BRANCH, ) return _base_ref_result( proven=False, reasons=[ f"remote '{remote_name}' publishes no '{prefix}HEAD' default and none of " f"{'/'.join(INTEGRATION_BRANCH_CANDIDATES)} exists as a remote-tracking " f"ref under '{prefix}'; the integration base ref cannot be derived " "(fail closed; no fetch is performed here)" ], reason_code=DENY_NO_BASE_BRANCH, ) 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, mode: str = "validation", ) -> 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``. *mode* accepts exactly the two values in :data:`SUPPORTED_MODES`: - ``"validation"`` (default): an independently trusted expected repository slug is known (or required). Candidate root's observed identity must match. Unprovable or missing expected identity fails closed when *require_binding* is True. - ``"derivation"``: caller is deriving the canonical repository identity. No expected slug exists yet by design. Derivation succeeds if the configured root exists, is a git repository, and carries a resolvable git remote identity. Every other explicitly supplied value — unknown strings, misspellings, the empty string, ``None``, and non-strings — is refused with ``proven`` False, ``block`` True, and ``reason_code`` :data:`DENY_UNKNOWN_MODE` (#973 B10). Omitting *mode* entirely keeps the documented ``"validation"`` default. """ # #973 B10: refuse an unsupported repository-authority mode as the very first # act, before any candidate-root existence check, path or symlink resolution, # git top-level discovery, remote-URL or repository-identity discovery, and # before any expected-versus-observed comparison or validation/derivation # behaviour. An unsupported mode previously fell into the catch-all ``else`` # on the configured-root path (silently receiving validation semantics) and # skipped the identity comparison entirely on the single-repository default # path (strictly weaker than validation), so matching identities could return # ``proven`` True. No repository identity may be resolved through a mode the # contract does not define. mode_reason = unsupported_mode_reason(mode) if mode_reason is not None: return _assessment( proven=False, reasons=[mode_reason], configured=bool((configured_value or "").strip()), canonical_repo_root=None, resolved_slug=None, source=source, reason_code=DENY_UNKNOWN_MODE, ) 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 resolved_slug = repository_identity_slug(derived, remote=remote) reasons: list[str] = [] # #973 B10: the allowlist above guarantees *mode* is one of the two # supported values here, so an unsupported value can no longer skip this # identity comparison and end up strictly weaker than validation. if mode == MODE_VALIDATION and expected_slug: expected = expected_slug.strip() if resolved_slug and resolved_slug.lower() != expected.lower(): reasons.append( f"canonical repository root identity mismatch: '{derived}' resolves " f"to repository '{resolved_slug}' but expected repository identity " f"is '{expected}' (forged or conflicting binding, fail closed)" ) elif not resolved_slug and require_binding: reasons.append( f"canonical repository root '{derived}' has no resolvable git " f"remote identity to confirm authorization for '{expected}' " "(fail closed)" ) return _assessment( proven=not reasons, reasons=reasons, configured=False, canonical_repo_root=derived, resolved_slug=resolved_slug, 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] = [] if mode == MODE_DERIVATION: if not resolved_slug: reasons.append( f"configured canonical repository root '{toplevel}' has no resolvable " "git remote identity (fail closed)" ) else: # #973 B10: MODE_VALIDATION only. This arm is no longer a catch-all — the # allowlist above admits no third value, so an unsupported mode can no # longer silently receive validation semantics here. 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)" ) elif require_binding: reasons.append( f"canonical repository root '{toplevel}' has configured value " f"'{configured_value}' but authoritative expected repository identity " "is unprovable or missing (fail closed)" ) return _assessment( proven=not reasons, 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 | None, resolved_slug: str | None, source: str | None, reason_code: str | None = None, ) -> dict: """Build the assessment payload. ``canonical_repo_root`` is None only when the assessment refused to resolve one at all — today exactly the unsupported-mode refusal (#973 B10), which must not derive a trusted repository identity through an undefined mode. ``reason_code`` is a machine-checkable refusal cause; None for every ordinary (non-coded) outcome. """ return { "proven": proven, "block": not proven, "reasons": list(reasons), "configured": configured, "canonical_repo_root": canonical_repo_root, "resolved_slug": resolved_slug, "source": source, "reason_code": reason_code, }