Gitea-Tools/remote_repo_guard.py

"""Remote/repo mismatch guard (#530).

Bare ``remote`` names resolve to a default ``org``/``repo`` via the ``REMOTES``
table in :mod:`gitea_auth`. For the ``prgs`` instance the hardcoded default repo
is ``Timesheet``, but the tools in this project operate on
``Scaled-Tech-Consulting/Gitea-Tools``. When a session runs inside a Gitea-Tools
worktree and calls a tool with a bare ``remote=prgs`` (no explicit ``org``/``repo``),
the resolved target silently points at the wrong repository, producing false 404s
and risking mutation of a different repo.

This module provides a pure assessment that compares the MCP-resolved ``org/repo``
against the local git remote URL and fails closed on a genuine mismatch, unless the
caller supplied explicit ``org``/``repo`` (in which case their intent is authoritative)
or the local remote URL is unavailable (best-effort corroboration only).
"""

from __future__ import annotations

import re

REMEDIATION = (
    "Pass explicit org= and repo= matching the local git remote on tools that "
    "accept those parameters (including mcp_get_control_plane_guide), "
    "e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools. "
    "Do not pass org/repo to tools whose schema does not list them."
)

# https://host/org/repo.git  or  git@host:org/repo.git
_REMOTE_URL_SLUG_RE = re.compile(
    r"(?:[:/])(?P<org>[^/]+)/(?P<repo>[^/]+?)(?:\.git)?/*$"
)


def parse_org_repo_from_remote_url(remote_url: str | None) -> tuple[str, str] | None:
    """Best-effort parse of org/repo from a git remote URL."""
    url = (remote_url or "").strip()
    if not url:
        return None
    match = _REMOTE_URL_SLUG_RE.search(url)
    if not match:
        return None
    org = (match.group("org") or "").strip()
    repo = (match.group("repo") or "").strip()
    if not org or not repo:
        return None
    return org, repo


def assess_remote_repo_match(
    *,
    remote: str,
    resolved_org: str,
    resolved_repo: str,
    local_remote_url: str | None,
    org_explicit: bool,
    repo_explicit: bool,
    for_mutation: bool = False,
    primary_org: str | None = None,
    primary_repo: str | None = None,
) -> dict:
    """Fail closed when the resolved org/repo disagrees with the local git remote.

    The guard enforces two protection levels:

    1. Cross-Project Mutation Boundary (#707):
       When ``for_mutation`` is True (codebase mutation operation: branch, commit, PR,
       merge, branch deletion), the target repository (``resolved_org/resolved_repo``)
       MUST match the primary authorized project context (``primary_org/primary_repo``
       or parsed from ``local_remote_url``). Any attempt to mutate a different project
       fails closed, even if explicit org/repo were passed. Metadata operations (such
       as creating issues or commenting on issues) across projects remain allowed.

    2. Workspace Mismatch Guard (#530):
       When ``for_mutation`` is False (or targets match), bare remotes must resolve
       to an org/repo slug present in the local git remote URL. When explicit org/repo
       are supplied for non-mutation operations, the caller's intent is authoritative.
    """
    reasons: list[str] = []

    eff_primary_org = primary_org
    eff_primary_repo = primary_repo
    if not eff_primary_org or not eff_primary_repo:
        parsed_primary = parse_org_repo_from_remote_url(local_remote_url)
        if parsed_primary:
            eff_primary_org = eff_primary_org or parsed_primary[0]
            eff_primary_repo = eff_primary_repo or parsed_primary[1]

    # #707: Enforce cross-project codebase mutation boundary if for_mutation is True
    if for_mutation and eff_primary_org and eff_primary_repo:
        if (
            resolved_org.lower() != eff_primary_org.lower()
            or resolved_repo.lower() != eff_primary_repo.lower()
        ):
            reasons.append(
                f"Cross-project mutation guard (#707): Attempted codebase mutation targeting "
                f"'{resolved_org}/{resolved_repo}' outside of primary authorized project "
                f"context '{eff_primary_org}/{eff_primary_repo}'"
            )
            return {
                "proven": False,
                "block": True,
                "cross_project_mutation_block": True,
                "reasons": reasons,
                "remote": remote,
                "resolved_org": resolved_org,
                "resolved_repo": resolved_repo,
                "primary_org": eff_primary_org,
                "primary_repo": eff_primary_repo,
                "local_remote_url": local_remote_url,
                "remediation": f"Cross-project codebase work is forbidden. Create an issue in the target repository ('{resolved_org}/{resolved_repo}') instead.",
            }

    if org_explicit and repo_explicit:
        return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)

    url = (local_remote_url or "").strip()
    if not url:
        return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)

    expected_slug = f"{resolved_org}/{resolved_repo}".lower()
    if expected_slug in url.lower():
        return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)

    reasons.append(
        f"MCP-resolved repository '{resolved_org}/{resolved_repo}' for remote "
        f"'{remote}' does not match the local git remote URL '{url}'"
    )
    return _assessment(False, reasons, remote, resolved_org, resolved_repo, local_remote_url)


def format_remote_repo_guard_error(assessment: dict) -> str:
    """Single RuntimeError message for the MCP resolver gate."""
    if assessment.get("cross_project_mutation_block"):
        resolved = f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}"
        primary = f"{assessment.get('primary_org')}/{assessment.get('primary_repo')}"
        return (
            f"Cross-project mutation guard (#707): Attempted codebase mutation targeting '{resolved}' "
            f"outside of primary authorized project context '{primary}'. "
            f"Cross-project codebase work (creating branches, committing files, creating PRs) is forbidden; "
            f"create an issue in the target repository ('{resolved}') instead."
        )

    reasons = "; ".join(
        assessment.get("reasons") or ["remote/repo resolution mismatch"]
    )
    resolved = (
        f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}"
    )
    local = assessment.get("local_remote_url") or "(unknown)"
    return (
        f"Remote/repo guard (#530): {reasons}. "
        f"Resolved target: {resolved}; local git remote: {local}. "
        f"{REMEDIATION}"
    )


def _assessment(
    proven: bool,
    reasons: list[str],
    remote: str,
    resolved_org: str,
    resolved_repo: str,
    local_remote_url: str | None,
) -> dict:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "remote": remote,
        "resolved_org": resolved_org,
        "resolved_repo": resolved_repo,
        "local_remote_url": local_remote_url,
        "remediation": REMEDIATION,
    }