Gitea-Tools/issue_lock_provenance.py

"""Issue-lock provenance and external-state disclosure (#447).

Sanctioned locks are written only by ``gitea_lock_issue`` (or adoption recovery
#442). Manual seeding of ``/tmp/gitea_issue_lock.json`` is unsafe and must be
blocked at PR creation unless explicit operator override proof is recorded.
"""

from __future__ import annotations

import os
import re
from datetime import datetime, timezone

ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")

SOURCE_LOCK_ISSUE = "gitea_lock_issue"
SOURCE_LOCK_ADOPTION = "gitea_lock_issue_adoption"
SOURCE_OPERATOR_OVERRIDE = "operator_override"

SANCTIONED_LOCK_SOURCES = frozenset({
    SOURCE_LOCK_ISSUE,
    SOURCE_LOCK_ADOPTION,
    SOURCE_OPERATOR_OVERRIDE,
})

_OPERATOR_OVERRIDE_ENV = "GITEA_ISSUE_LOCK_OPERATOR_OVERRIDE"

_ISSUE_LOCK_PATH_RE = re.compile(
    r"(?:/tmp/)?gitea_issue_lock\.json",
    re.IGNORECASE,
)
_LOCK_SEED_RE = re.compile(
    r"(?:seed(?:ed|ing)?|restor(?:e|ed|ing)|wrote|written|write|programmatically|"
    r"hand[- ]forg|manual(?:ly)?).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
)
_LOCK_REMOVE_RE = re.compile(
    r"(?:\brm\b|remove|deleted?|unlink).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
)
_LOCK_READ_RE = re.compile(
    r"(?:read|loaded?|parsed?).{0,80}gitea_issue_lock",
    re.IGNORECASE | re.DOTALL,
)
_EXTERNAL_NONE_RE = re.compile(
    r"external[- ]state mutations\s*:\s*none\b",
    re.IGNORECASE,
)
_EXTERNAL_FIELD_RE = re.compile(
    r"external[- ]state mutations\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_CLEANUP_ONLY_RE = re.compile(
    r"cleanup mutations\s*:\s*(?:none|lock removed|removed issue lock)",
    re.IGNORECASE,
)
_PR_CREATED_RE = re.compile(
    r"(?:\bgitea_create_pr\b|PR\s*#\s*\d+\s+created|created\s+PR\s*#|opened\s+PR\s*#|"
    r"PR\s+creation\s+(?:succeeded|complete))",
    re.IGNORECASE,
)
_REVIEW_APPROVE_RE = re.compile(
    r"(?:submitted\s+(?:['\"]approve['\"]|approve\s+review)|"
    r"review decision\s*:\s*approve|approved\s+PR\s*#|gitea_review_pr.*approve)",
    re.IGNORECASE,
)
_OVERRIDE_PROOF_RE = re.compile(
    r"operator[- ]override\s+proof\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def build_sanctioned_lock_provenance(
    *,
    tool: str,
    source: str = SOURCE_LOCK_ISSUE,
    claimant: dict | None = None,
    adoption: dict | None = None,
) -> dict:
    """Return provenance metadata stored with a sanctioned lock write."""
    entry = {
        "source": source,
        "written_at": _utc_now_iso(),
        "written_by_tool": tool,
        "lock_file_path": ISSUE_LOCK_FILE,
    }
    if claimant:
        entry["claimant"] = claimant
    if adoption:
        entry["adoption"] = adoption
    return entry


def operator_override_requested() -> bool:
    return os.environ.get(_OPERATOR_OVERRIDE_ENV, "").strip().lower() in {
        "1",
        "true",
        "yes",
    }


def build_operator_override_provenance(*, reason: str, claimant: dict | None = None) -> dict:
    text = (reason or "").strip()
    if not text:
        raise ValueError(
            "operator override requires a non-empty override reason (fail closed)"
        )
    entry = build_sanctioned_lock_provenance(
        tool="operator_override",
        source=SOURCE_OPERATOR_OVERRIDE,
        claimant=claimant,
    )
    entry["override_reason"] = text
    return entry


def assess_lock_file_for_create_pr(lock_data: dict | None) -> dict:
    """Fail closed when lock file lacks sanctioned provenance (#447)."""
    data = lock_data if isinstance(lock_data, dict) else {}
    reasons: list[str] = []
    provenance = data.get("lock_provenance")
    if not isinstance(provenance, dict):
        reasons.append(
            "issue lock file lacks sanctioned lock_provenance; manual seeding is "
            "not a normal recovery path — call gitea_lock_issue or use #442 adoption"
        )
        return _provenance_result(False, reasons, provenance)

    source = str(provenance.get("source") or "").strip()
    if source not in SANCTIONED_LOCK_SOURCES:
        reasons.append(
            f"issue lock provenance source '{source or '(missing)'}' is not sanctioned"
        )

    if source == SOURCE_OPERATOR_OVERRIDE and not str(
        provenance.get("override_reason") or ""
    ).strip():
        reasons.append(
            "operator_override lock provenance requires override_reason proof"
        )

    if not data.get("work_lease"):
        reasons.append("issue lock file missing work_lease metadata")

    if not str(provenance.get("written_by_tool") or "").strip():
        reasons.append("issue lock provenance missing written_by_tool")

    proven = not reasons
    return _provenance_result(proven, reasons, provenance)


def _provenance_result(proven: bool, reasons: list[str], provenance: dict | None) -> dict:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "lock_provenance": provenance,
    }


def format_lock_provenance_error(assessment: dict) -> str:
    reasons = "; ".join(assessment.get("reasons") or ["unknown lock provenance violation"])
    return f"Issue lock provenance guard (#447): {reasons} (fail closed)"


def _lock_activity_detected(text: str) -> dict[str, bool]:
    body = text or ""
    return {
        "seed_or_restore": bool(_LOCK_SEED_RE.search(body)),
        "remove": bool(_LOCK_REMOVE_RE.search(body)),
        "read": bool(_LOCK_READ_RE.search(body)),
    }


def _external_state_discloses_lock(text: str) -> bool:
    match = _EXTERNAL_FIELD_RE.search(text or "")
    if not match:
        return False
    value = (match.group(1) or "").strip().lower()
    if value in {"", "none", "n/a"}:
        return False
    return "lock" in value or "gitea_issue_lock" in value or "issue-lock" in value


def assess_issue_lock_external_state_report(report_text: str) -> dict:
    """Require explicit external-state disclosure for issue-lock mutations (#447)."""
    text = report_text or ""
    activity = _lock_activity_detected(text)
    if not any(activity.values()):
        return {"proven": True, "block": False, "reasons": [], "activity": activity}

    reasons: list[str] = []
    disclosed = _external_state_discloses_lock(text)

    if activity["seed_or_restore"] and _EXTERNAL_NONE_RE.search(text):
        reasons.append(
            "report mentions seeding/restoring gitea_issue_lock.json but claims "
            "External-state mutations: none"
        )
    elif activity["seed_or_restore"] and not disclosed:
        reasons.append(
            "report mentions issue-lock file activity but External-state mutations "
            "does not disclose read/write of gitea_issue_lock.json"
        )

    if activity["remove"]:
        if _EXTERNAL_NONE_RE.search(text):
            reasons.append(
                "report mentions removing gitea_issue_lock.json but claims "
                "External-state mutations: none"
            )
        elif not disclosed and _CLEANUP_ONLY_RE.search(text):
            reasons.append(
                "report removes issue lock but classifies it as cleanup only; "
                "record under External-state mutations"
            )
        elif not disclosed:
            reasons.append(
                "report mentions deleting issue lock without External-state "
                "mutation disclosure"
            )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "activity": activity,
    }


def assess_manual_lock_pr_without_override(report_text: str) -> dict:
    """Block reports that created a PR via manual lock seed without override proof."""
    text = report_text or ""
    seeded = bool(_LOCK_SEED_RE.search(text))
    created = bool(_PR_CREATED_RE.search(text))
    if not (seeded and created):
        return {"proven": True, "block": False, "reasons": []}

    if _OVERRIDE_PROOF_RE.search(text):
        return {"proven": True, "block": False, "reasons": []}

    return {
        "proven": False,
        "block": True,
        "reasons": [
            "report created/opened a PR after manual issue-lock seeding without "
            "operator override proof"
        ],
    }


def assess_author_reviewer_same_run_report(report_text: str) -> dict:
    """Reviewer handoff must not create and approve the same PR in one run (#447)."""
    text = report_text or ""
    if not (_PR_CREATED_RE.search(text) and _REVIEW_APPROVE_RE.search(text)):
        return {"proven": True, "block": False, "reasons": []}
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "report mixes author-side PR creation and reviewer approval in one "
            "final handoff; split author and reviewer sessions"
        ],
    }