Gitea-Tools/reviewer_fallback.py

"""Reviewer local-fallback detection for normal PR review workflows (#324).

Normal reviewer runs must use MCP tools. Reading profile secret files or
running local Gitea helper scripts while MCP is available is a fail-closed
violation. Explicit recovery mode may use local fallback only with full proof.
"""

from __future__ import annotations

import re
from typing import Any

LOCAL_GITEA_SCRIPT_NAMES = (
    "list_prs.py",
    "view_pr.py",
    "create_pr.py",
    "merge_pr.py",
    "edit_pr.py",
    "list_issues.py",
    "delete_branch.py",
    "create_issue.py",
    "close_issue.py",
    "review_pr.py",
    "mark_issue.py",
    "mirror_refs.sh",
)

PROFILE_SECRET_MARKERS = (
    "profiles.json",
    ".config/gitea-tools/profiles",
    "gitea_auth.py",
    "gitea_config.py",
    "keychain",
    "credential fill",
    "token store",
)

_RECOVERY_MODE_RE = re.compile(
    r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|"
    r"mcp tools unavailable|no mcp path)\b",
    re.IGNORECASE,
)
_FALLBACK_CLASSIFICATION_RE = re.compile(
    r"\b(?:local fallback|fallback mode|used local gitea|ran local script)\b",
    re.IGNORECASE,
)
_MCP_TOOL_USE_RE = re.compile(
    r"\b(?:gitea_list_prs|gitea_view_pr|gitea_review_pr|gitea_merge_pr|"
    r"gitea_resolve_task_capability|gitea_whoami|mcp tool)\b",
    re.IGNORECASE,
)
_LOCAL_SCRIPT_RE = re.compile(
    r"(?:^|[\s\"'`/])(?:" + "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES) + r")",
    re.IGNORECASE,
)
_PROFILE_ACCESS_RE = re.compile(
    r"(?:read|open|inspect|cat|view|access(?:ed)?|loaded?)\s+(?:file\s+)?[`'\"]?"
    r"[^`'\"]*profiles\.json|profiles\.json[`'\"]?\s+(?:read|opened|inspected|accessed)|"
    r"~/?\.config/gitea-tools/profiles",
    re.IGNORECASE,
)
_LOCAL_ENV_SCRIPT_RE = re.compile(
    r"GITEA_MCP_(?:CONFIG|PROFILE)=.*\b(?:python\s+)?(?:list_prs|view_pr|create_pr|merge_pr)\.py",
    re.IGNORECASE,
)
_RECOVERY_PROOF_FIELDS = (
    ("identity proof", ("identity proof", "exact identity proof")),
    ("profile proof", ("profile proof", "exact profile proof")),
    ("repo proof", ("repo proof", "exact repo proof")),
    ("capability proof", ("capability proof", "exact capability proof")),
    ("mcp unavailable reason", (
        "why mcp was unavailable",
        "mcp unavailable",
        "mcp unavailability",
    )),
)


def _collect_observed_text(
    report_text: str,
    action_log: list[dict] | None,
) -> str:
    chunks = [report_text or ""]
    for entry in action_log or []:
        for key in ("command", "action", "detail", "path", "script"):
            value = entry.get(key)
            if value:
                chunks.append(str(value))
    return "\n".join(chunks)


def _detect_profile_secret_access(text: str) -> list[str]:
    reasons = []
    lower = text.lower()
    if _PROFILE_ACCESS_RE.search(text):
        reasons.append("report or action log shows profiles.json/profile secret access")
    for marker in PROFILE_SECRET_MARKERS:
        if marker == "profiles.json":
            continue
        if marker in lower and "do not" not in lower and "must not" not in lower:
            if any(verb in lower for verb in ("read ", "open ", "inspect ", "cat ", "loaded ")):
                reasons.append(f"profile secret surface '{marker}' accessed during review")
    return reasons


def _detect_local_script_use(text: str) -> list[str]:
    reasons = []
    match = _LOCAL_SCRIPT_RE.search(text)
    if match:
        reasons.append(
            f"local Gitea helper script invoked ({match.group(0).strip()})"
        )
    if _LOCAL_ENV_SCRIPT_RE.search(text):
        reasons.append(
            "local Gitea script run with GITEA_MCP_CONFIG/GITEA_MCP_PROFILE env overrides"
        )
    return reasons


def _recovery_proof_missing(text: str) -> list[str]:
    lower = text.lower()
    missing = []
    for label, aliases in _RECOVERY_PROOF_FIELDS:
        if not any(alias in lower for alias in aliases):
            missing.append(label)
    if not _FALLBACK_CLASSIFICATION_RE.search(text):
        missing.append("fallback classification")
    if not _RECOVERY_MODE_RE.search(text):
        missing.append("recovery mode declaration")
    return missing


def assess_reviewer_fallback_report(
    report_text: str,
    *,
    action_log: list[dict] | None = None,
    recovery_mode: bool = False,
    mcp_available: bool = True,
    mcp_tools_used: bool | None = None,
) -> dict[str, Any]:
    """Fail closed when normal reviewer workflows use local Gitea fallbacks (#324)."""
    text = _collect_observed_text(report_text, action_log)
    lower = (report_text or "").lower()

    profile_violations = _detect_profile_secret_access(text)
    script_violations = _detect_local_script_use(text)
    violations = profile_violations + script_violations

    if mcp_tools_used is None:
        mcp_tools_used = bool(_MCP_TOOL_USE_RE.search(report_text or ""))
    elif mcp_tools_used is False and _MCP_TOOL_USE_RE.search(report_text or ""):
        mcp_tools_used = True

    reasons: list[str] = []
    recovery_declared = recovery_mode or bool(_RECOVERY_MODE_RE.search(report_text or ""))

    if violations and mcp_available and not recovery_declared:
        reasons.extend(violations)
        if mcp_tools_used:
            reasons.append(
                "MCP tools were available and used; local fallback/profile access is forbidden"
            )
        else:
            reasons.append(
                "MCP path is available; normal review must not use local Gitea fallbacks"
            )

    if violations and recovery_declared:
        missing = _recovery_proof_missing(report_text or "")
        if missing:
            reasons.append(
                "recovery-mode fallback missing proof fields: "
                + ", ".join(missing)
            )

    if (
        not violations
        and recovery_declared
        and _FALLBACK_CLASSIFICATION_RE.search(report_text or "")
        and not recovery_mode
    ):
        missing = _recovery_proof_missing(report_text or "")
        if missing:
            reasons.append(
                "report claims local fallback but recovery proof is incomplete: "
                + ", ".join(missing)
            )

    proven = not reasons
    blocked = bool(reasons) and not recovery_declared or any(
        "recovery-mode fallback missing" in r or "recovery proof is incomplete" in r
        for r in reasons
    )
    return {
        "proven": proven,
        "block": blocked or (bool(reasons) and mcp_available and not recovery_declared),
        "downgraded": bool(reasons) and not blocked,
        "recovery_mode": recovery_declared,
        "mcp_available": mcp_available,
        "mcp_tools_used": mcp_tools_used,
        "profile_violations": profile_violations,
        "script_violations": script_violations,
        "reasons": reasons,
        "safe_next_action": (
            "use MCP reviewer tools only; do not read profiles.json or run local Gitea scripts"
            if reasons and not recovery_declared
            else (
                "complete recovery-mode fallback proof before claiming local fallback"
                if reasons
                else "proceed with MCP tools"
            )
        ),
    }