Gitea-Tools/review_quarantine.py

"""Controller quarantine of contaminated formal reviews (#695).

Quarantine records are durable under the MCP session-state root and are only
writable when the caller holds a **native** MCP transport runtime. Untrusted
local scripts that import this module cannot establish a valid quarantine
(writes fail closed). Live server-side gates (eligibility, merge, feedback)
honor quarantine by review_id + PR + head.

Forensic evidence (Gitea review objects, lease comments) is never deleted.
"""

from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from typing import Any

import mcp_daemon_guard
import mcp_session_state

KIND_QUARANTINE = "review_quarantine"
CONFIRMATION_PREFIX = "QUARANTINE CONTAMINATED REVIEW"


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def quarantine_state_path(
    *,
    remote: str,
    org: str,
    repo: str,
    pr_number: int,
    review_id: int,
) -> str:
    # Dedicated subdir under session state root (mode 0o700).
    root = os.path.join(mcp_session_state.default_state_dir(), "quarantine")
    os.makedirs(root, mode=0o700, exist_ok=True)
    # Explicit multi-segment key so remote/org/repo/pr/review cannot collide.
    segs = [
        KIND_QUARANTINE,
        mcp_session_state._sanitize_segment(remote),
        mcp_session_state._sanitize_segment(org),
        mcp_session_state._sanitize_segment(repo),
        f"pr{int(pr_number)}",
        f"rev{int(review_id)}",
    ]
    return os.path.join(root, "-".join(segs) + ".json")


def build_quarantine_record(
    *,
    remote: str,
    org: str,
    repo: str,
    pr_number: int,
    review_id: int,
    reviewed_head_sha: str,
    reason: str,
    actor_username: str | None,
    profile_name: str | None,
    incident_issue: int | None = None,
    forensic_comment_ids: list[int] | None = None,
) -> dict[str, Any]:
    provenance = mcp_daemon_guard.mutation_provenance_fields()
    return {
        "kind": KIND_QUARANTINE,
        "issue_ref": "#695",
        "remote": remote,
        "org": org,
        "repo": repo,
        "pr_number": int(pr_number),
        "review_id": int(review_id),
        "reviewed_head_sha": (reviewed_head_sha or "").strip().lower(),
        "reason": (reason or "").strip(),
        "actor_username": actor_username,
        "profile_name": profile_name,
        "incident_issue": incident_issue,
        "forensic_comment_ids": list(forensic_comment_ids or []),
        "created_at": _now(),
        "native_provenance": provenance,
        "merge_authorization": "void",
        "retain_forensic_evidence": True,
    }


def assess_quarantine_write(
    *,
    confirmation: str,
    pr_number: int,
    review_id: int,
    reason: str,
    native_required: bool = True,
) -> dict[str, Any]:
    """Pure assessment of whether a quarantine write may proceed."""
    reasons: list[str] = []
    expected = f"{CONFIRMATION_PREFIX} {int(review_id)} PR {int(pr_number)}"
    if (confirmation or "").strip() != expected:
        reasons.append(
            f"confirmation must equal exactly '{expected}' (fail closed, #695)"
        )
    if not (reason or "").strip():
        reasons.append("quarantine reason is required (fail closed, #695)")
    if native_required and not mcp_daemon_guard.is_native_mcp_transport():
        if not mcp_daemon_guard.is_pytest_runtime():
            reasons.append(
                "quarantine write requires native MCP transport; untrusted "
                "local code cannot create quarantine records (#695)"
            )
    return {
        "allowed": not reasons,
        "expected_confirmation": expected,
        "reasons": reasons,
    }


def write_quarantine_record(record: dict[str, Any]) -> dict[str, Any]:
    """Persist quarantine record; fail closed outside native/pytest runtime."""
    if not mcp_daemon_guard.is_native_mcp_transport():
        if not mcp_daemon_guard.is_pytest_runtime():
            raise mcp_daemon_guard.UnsanctionedRuntimeError(
                "quarantine write blocked: non-native runtime (#695)"
            )
    path = quarantine_state_path(
        remote=str(record["remote"]),
        org=str(record["org"]),
        repo=str(record["repo"]),
        pr_number=int(record["pr_number"]),
        review_id=int(record["review_id"]),
    )
    # Refuse overwriting with weaker provenance from untrusted caller by
    # requiring native fields present.
    if not (record.get("native_provenance") or {}).get("native_mcp_transport"):
        if not mcp_daemon_guard.is_pytest_runtime():
            raise mcp_daemon_guard.UnsanctionedRuntimeError(
                "quarantine record missing native provenance (#695)"
            )
    tmp = path + ".tmp"
    data = json.dumps(record, indent=2, sort_keys=True)
    with open(tmp, "w", encoding="utf-8") as fh:
        fh.write(data)
        fh.flush()
        os.fsync(fh.fileno())
    os.replace(tmp, path)
    os.chmod(path, 0o600)
    return {"path": path, "written": True, "record": record}


def load_quarantine_record(
    *,
    remote: str,
    org: str,
    repo: str,
    pr_number: int,
    review_id: int,
) -> dict[str, Any] | None:
    path = quarantine_state_path(
        remote=remote,
        org=org,
        repo=repo,
        pr_number=pr_number,
        review_id=review_id,
    )
    if not os.path.isfile(path):
        return None
    try:
        with open(path, encoding="utf-8") as fh:
            data = json.load(fh)
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(data, dict):
        return None
    return data


def is_review_quarantined(
    *,
    remote: str,
    org: str,
    repo: str,
    pr_number: int,
    review_id: int | None,
    reviewed_head_sha: str | None = None,
) -> dict[str, Any]:
    """Whether a formal review is quarantined for merge authorization (#695)."""
    if review_id is None:
        return {"quarantined": False, "record": None, "reasons": []}
    rec = load_quarantine_record(
        remote=remote,
        org=org,
        repo=repo,
        pr_number=pr_number,
        review_id=int(review_id),
    )
    if not rec:
        return {"quarantined": False, "record": None, "reasons": []}
    reasons = [
        f"formal review_id {review_id} on PR #{pr_number} is quarantined "
        f"(#695; incident issue {rec.get('incident_issue')}); "
        "merge authorization is void; forensic evidence retained"
    ]
    want_head = (reviewed_head_sha or "").strip().lower()
    rec_head = (rec.get("reviewed_head_sha") or "").strip().lower()
    if want_head and rec_head and want_head != rec_head:
        # Still quarantined by review_id; note head mismatch for operators.
        reasons.append(
            f"quarantine head {rec_head[:12]}… differs from assessed head "
            f"{want_head[:12]}… (still void by review_id)"
        )
    return {"quarantined": True, "record": rec, "reasons": reasons}


def filter_approvals_for_merge(
    *,
    remote: str,
    org: str,
    repo: str,
    pr_number: int,
    current_head_sha: str | None,
    reviews: list[dict],
) -> dict[str, Any]:
    """Split reviews into merge-usable vs quarantined contaminated approvals."""
    current = (current_head_sha or "").strip().lower()
    usable: list[dict] = []
    quarantined: list[dict] = []
    for rev in reviews or []:
        if not isinstance(rev, dict):
            continue
        verdict = (rev.get("verdict") or rev.get("state") or "").upper()
        if verdict not in {"APPROVED", "APPROVE"}:
            continue
        if rev.get("dismissed"):
            continue
        rid = rev.get("review_id") or rev.get("id")
        head = (
            rev.get("reviewed_head_sha")
            or rev.get("commit_id")
            or rev.get("head_sha")
            or ""
        ).strip().lower()
        q = is_review_quarantined(
            remote=remote,
            org=org,
            repo=repo,
            pr_number=pr_number,
            review_id=int(rid) if rid is not None else None,
            reviewed_head_sha=head or current,
        )
        entry = {**rev, "review_id": rid, "reviewed_head_sha": head}
        if q["quarantined"]:
            entry["quarantined"] = True
            entry["quarantine_reasons"] = q["reasons"]
            quarantined.append(entry)
        else:
            entry["quarantined"] = False
            usable.append(entry)
    usable_at_head = [
        e
        for e in usable
        if current and (e.get("reviewed_head_sha") or "") == current
    ]
    return {
        "usable_approvals": usable,
        "usable_approvals_at_current_head": usable_at_head,
        "quarantined_approvals": quarantined,
        "approval_visible_for_merge": bool(usable_at_head),
        "has_quarantined_approval_at_head": any(
            current
            and (e.get("reviewed_head_sha") or "") == current
            for e in quarantined
        ),
    }


def format_quarantine_audit_comment(record: dict[str, Any]) -> str:
    """Append-only forensic audit comment body (does not delete evidence)."""
    lines = [
        "## Contaminated formal review quarantine (#695)",
        "",
        "Status: **QUARANTINED — merge authorization VOID**",
        "",
        f"- review_id: `{record.get('review_id')}`",
        f"- pr: `#{record.get('pr_number')}`",
        f"- reviewed_head_sha: `{record.get('reviewed_head_sha')}`",
        f"- actor: `{record.get('actor_username')}`",
        f"- profile: `{record.get('profile_name')}`",
        f"- incident_issue: `#{record.get('incident_issue')}`",
        f"- reason: {record.get('reason')}",
        f"- created_at: `{record.get('created_at')}`",
        f"- native_transport: "
        f"`{(record.get('native_provenance') or {}).get('native_mcp_transport')}`",
        f"- native_token_fingerprint: "
        f"`{(record.get('native_provenance') or {}).get('native_token_fingerprint')}`",
        f"- forensic_comment_ids retained: "
        f"`{record.get('forensic_comment_ids')}`",
        "",
        "This record does **not** delete Gitea reviews or historical comments. "
        "Fresh native-MCP re-review is required before merge.",
    ]
    return "\n".join(lines)


def assess_untrusted_canonical_approval_claim(body: str) -> list[str]:
    """Reasons to reject canonical comments that claim approved/merge-ready.

    Used when the comment asserts approval without native review proof (#695 AC7).
    False "official workflow" claims that cite offline/import paths are always
    rejected even when a NATIVE_REVIEW_PROOF line is present.
    """
    text = body or ""
    lower = text.lower()
    claims_approved = (
        "state:\napproved" in lower
        or "state: approved" in lower
        or "who_is_next:\nmerger" in lower
        or "who_is_next: merger" in lower
        or "merge_ready: true" in lower
        or "merge_ready:\ntrue" in lower
        or "ready-to-merge" in lower
    )
    if not claims_approved:
        return []
    # Reject explicit offline/import "official workflow" spoofing (#695 AC9).
    offline_spoof = (
        "offline_mcp" in lower
        or "offline import" in lower
        or "direct import" in lower
        or "import gitea_mcp_server" in lower
        or "run_quarantine.py" in lower
        or "offline_mcp_helper" in lower
        or "offline_mcp_runner" in lower
    )
    has_proof = (
        "NATIVE_REVIEW_PROOF:" in text
        or "native_review_proof:" in lower
    )
    if offline_spoof:
        return [
            "canonical approval claim cites offline/import/helper path; "
            "NATIVE_REVIEW_PROOF rejected (#695 AC7/AC9); stop after native "
            "MCP failure — do not construct offline mutation fallbacks"
        ]
    if has_proof:
        return []
    return [
        "canonical comment claims approved/merge-ready state without "
        "NATIVE_REVIEW_PROOF from a native MCP review mutation (#695 AC7); "
        "contaminated or untrusted approvals cannot certify merger handoff"
    ]