Gitea-Tools/canonical_repository_root.py

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

# 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 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,
    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,
    }