Gitea-Tools/stable_branch_push_guard.py

"""Fail-closed guard against direct pushes to stable branches (#671).

MCP workflow sessions must never publish to a stable branch (``master``,
``main``, ``dev``, ...) directly. Stable-branch updates land only through
sanctioned Gitea merge tooling or an explicitly authorized reconciler path.

Incident origin (#670): a bare direct-to-master commit ``2fa97c26`` landed on
``prgs/master`` without PR/review provenance, and a PR #654 merger run
attempted ``git push prgs master`` (audited as a no-op, but the *intent* is the
hazard). Partial protections already existed (``author_proofs.PROTECTED_BRANCHES``,
``root_checkout_guard``, branch-push proofs) but did not detect shell push
equivalents, mark the session contaminated after such an attempt, or fail
closed on subsequent review/merge/close/completion mutations.

This module is the detection + contamination-policy core. Like
``author_proofs`` and ``root_checkout_guard`` it is **pure**: callers gather
the raw facts (the proposed command line, git state, the durable contamination
marker) and pass them in, so the same logic serves prompts, MCP gates, and
tests. Nothing here performs git or network calls or reads durable state; the
server wires these helpers to ``mcp_session_state`` and ``verify_preflight_purity``.

Design rules honoured (from the #671 acceptance criteria):

* Detect ``git push <remote> master`` shell equivalents, including refspecs
  (``HEAD:master``, ``+refs/heads/x:refs/heads/master``), ``--force``,
  ``--dry-run``/no-op intent, and delete refspecs (``:master``).
* Never false-block a feature-branch push (``git push prgs fix/issue-671-...``).
* Never treat ``git fetch`` / ``git pull --ff-only`` as a push.
* Never treat sanctioned ``gitea_merge_pr`` / Gitea API merge as a push.
* Fail closed on ambiguous targets that *may* resolve to a stable branch, but
  surface ambiguity as its own signal rather than silently blocking feature work.
* Contamination must not be clearable by the same worker session — only a
  reconciler (audit) role may clear or bypass the gate.
"""

from __future__ import annotations

import re
from typing import Any, Iterable

from author_proofs import PROTECTED_BRANCHES

# Single source of truth for the stable branch set: the same protected set the
# author branch-identity proofs already enforce (#177), so the two guards can
# never drift apart.
STABLE_BRANCHES = PROTECTED_BRANCHES

# Mutations that must fail closed once the session is contaminated by a direct
# stable-branch push attempt (#671 AC4: review, merge, close, completion).
# ``comment_issue`` / ``lock_issue`` / ``create_issue`` are deliberately NOT
# gated so a contaminated worker can still post the durable audit comment and
# hand off. Only a reconciler (audit) role bypasses this set.
CONTAMINATION_GATED_TASKS = frozenset({
    "create_pr",
    "commit_files",
    "gitea_commit_files",
    "close_pr",
    "close_issue",
    "review_pr",
    "approve_pr",
    "request_changes_pr",
    "submit_pr_review",
    "merge_pr",
    "delete_branch",
    "complete_issue",
})

CONTAMINATION_KIND = "stable_branch_push"

REMEDIATION = (
    "Direct stable-branch publication is forbidden for worker sessions. Stop, "
    "leave the stable branch untouched, and route the change through sanctioned "
    "Gitea merge tooling (gitea_merge_pr) or an explicitly authorized reconciler "
    "audit. This session is workflow-contaminated until a reconciler audits it."
)

# ── command tokenising ────────────────────────────────────────────────────────

# Split a compound command line into individual simple commands on shell
# separators so ``a && git push prgs master`` is analysed segment by segment.
_SEGMENT_SPLIT_RE = re.compile(r"(?:\|\||&&|\||;|\n)")

_GIT_PUSH_RE = re.compile(r"\bgit\b[\w\s\-]*?\bpush\b", re.IGNORECASE)
_GIT_FETCH_RE = re.compile(r"\bgit\b[\w\s\-]*?\b(?:fetch|pull)\b", re.IGNORECASE)

# Flags that take a value argument in ``git push`` (so the following token is
# consumed as the flag's value, not a refspec).
_VALUE_FLAGS = frozenset({
    "--repo",
    "-o",
    "--push-option",
    "--receive-pack",
    "--exec",
})

_FORCE_FLAGS = frozenset({"-f", "--force"})
_DRY_RUN_FLAGS = frozenset({"-n", "--dry-run"})
_DELETE_FLAGS = frozenset({"-d", "--delete"})


def _clean(value: str | None) -> str:
    return (value or "").strip()


def _strip_ref_prefix(ref: str) -> str:
    ref = ref.strip()
    for prefix in ("refs/heads/", "heads/"):
        if ref.startswith(prefix):
            return ref[len(prefix):]
    return ref


def is_stable_ref(ref: str | None) -> bool:
    """True when ``ref`` names a configured stable branch."""
    name = _strip_ref_prefix(_clean(ref))
    return bool(name) and name in STABLE_BRANCHES


# ── redaction ─────────────────────────────────────────────────────────────────

_URL_USERINFO_RE = re.compile(r"(https?://)[^/\s:@]+(?::[^/\s@]+)?@", re.IGNORECASE)
_SECRET_ASSIGN_RE = re.compile(
    r"\b((?:GITEA_)?(?:TOKEN|PASSWORD|PASS|PAT|SECRET|API_KEY|AUTH))\s*=\s*\S+",
    re.IGNORECASE,
)
_BEARER_RE = re.compile(r"\b(Bearer|token)\s+[A-Za-z0-9._\-]{8,}", re.IGNORECASE)


def redact_command(command: str | None) -> str:
    """Strip credentials/URLs from a command line before logging it (#671).

    Redacts URL userinfo (``https://user:tok@host`` → ``https://***@host``),
    ``TOKEN=...`` style assignments, and bearer/token headers. Leaves the
    structural parts (``git push <remote> master``) intact for audit value.
    """
    text = _clean(command)
    if not text:
        return ""
    text = _URL_USERINFO_RE.sub(r"\1***@", text)
    text = _SECRET_ASSIGN_RE.sub(lambda m: f"{m.group(1)}=***", text)
    text = _BEARER_RE.sub(lambda m: f"{m.group(1)} ***", text)
    return text


# ── push classification ───────────────────────────────────────────────────────

def _analyse_push_segment(segment: str) -> dict[str, Any]:
    """Classify one ``git push ...`` command segment."""
    tokens = segment.split()
    # Drop everything up to and including the ``push`` verb.
    try:
        push_idx = next(
            i for i, tok in enumerate(tokens)
            if tok.lower() == "push"
        )
    except StopIteration:
        push_idx = -1
    args = tokens[push_idx + 1:] if push_idx >= 0 else []

    is_force = False
    is_dry_run = False
    is_delete = False
    positionals: list[str] = []

    skip_next = False
    for tok in args:
        if skip_next:
            skip_next = False
            continue
        if tok in _FORCE_FLAGS or tok.startswith("--force-with-lease") or tok.startswith("--force-if-includes"):
            is_force = True
            continue
        if tok in _DRY_RUN_FLAGS:
            is_dry_run = True
            continue
        if tok in _DELETE_FLAGS:
            is_delete = True
            continue
        if tok in _VALUE_FLAGS:
            skip_next = True
            continue
        if tok.startswith("--") and "=" in tok:
            # e.g. --repo=... : self-contained, not a refspec.
            continue
        if tok.startswith("-"):
            # Unknown/combined short flag; ignore for ref detection.
            continue
        positionals.append(tok)

    # First positional after push is the remote (when any positional exists);
    # the rest are refspecs / branch names.
    remote = positionals[0] if positionals else None
    refspecs = positionals[1:]

    stable_refs: list[str] = []
    force_to_stable = False
    delete_stable = False
    for spec in refspecs:
        force_spec = spec.startswith("+")
        body = spec[1:] if force_spec else spec
        if ":" in body:
            src, _, dst = body.partition(":")
            dest = dst
            if src == "" and dst:
                # ``:master`` — delete refspec.
                is_delete = True
        else:
            dest = body
        if is_stable_ref(dest):
            stable_refs.append(_strip_ref_prefix(dest))
            if force_spec or is_force:
                force_to_stable = True
            if is_delete:
                delete_stable = True

    targets_stable = bool(stable_refs)
    # Ambiguous: ``git push <remote>`` (or bare ``git push``) with no refspec —
    # resolves to the current branch's upstream, which we cannot see from the
    # command alone. Flag it, but do not assert stable (would false-block
    # feature-branch pushes).
    ambiguous = not refspecs

    return {
        "is_git_push": True,
        "remote": remote,
        "refspecs": refspecs,
        "targets_stable": targets_stable,
        "stable_refs": stable_refs,
        "is_force": is_force or force_to_stable,
        "is_dry_run": is_dry_run,
        "is_delete": is_delete or delete_stable,
        "ambiguous_target": ambiguous,
    }


def classify_push_command(command: str | None) -> dict[str, Any]:
    """Classify a shell command line for direct stable-branch push intent (#671).

    Returns a dict describing whether the command is a ``git push`` targeting a
    stable branch. Fetch/pull are never pushes. A sanctioned Gitea merge (which
    is not a ``git push`` at all) classifies as non-push. Dry-run and no-op
    pushes still count as *intent* and are reported as contamination
    (``proves_intent``) so a ``--dry-run`` cannot be used to probe the gate.
    """
    text = _clean(command)
    result: dict[str, Any] = {
        "command": text,
        "redacted_command": redact_command(text),
        "is_git_push": False,
        "is_fetch_or_pull": False,
        "targets_stable": False,
        "stable_refs": [],
        "is_force": False,
        "is_dry_run": False,
        "is_delete": False,
        "ambiguous_target": False,
        "contamination": False,
        "proves_intent": False,
        "reasons": [],
    }
    if not text:
        return result

    stable_refs: list[str] = []
    saw_push = False
    for segment in _SEGMENT_SPLIT_RE.split(text):
        seg = segment.strip()
        if not seg:
            continue
        if _GIT_FETCH_RE.search(seg) and not _GIT_PUSH_RE.search(seg):
            result["is_fetch_or_pull"] = True
            continue
        if not _GIT_PUSH_RE.search(seg):
            continue
        saw_push = True
        analysis = _analyse_push_segment(seg)
        result["is_git_push"] = True
        result["is_force"] = result["is_force"] or analysis["is_force"]
        result["is_dry_run"] = result["is_dry_run"] or analysis["is_dry_run"]
        result["is_delete"] = result["is_delete"] or analysis["is_delete"]
        result["ambiguous_target"] = result["ambiguous_target"] or analysis["ambiguous_target"]
        stable_refs.extend(analysis["stable_refs"])

    result["stable_refs"] = sorted(set(stable_refs))
    result["targets_stable"] = bool(stable_refs)

    reasons: list[str] = []
    if result["targets_stable"]:
        refs = ", ".join(result["stable_refs"])
        verb = "delete" if result["is_delete"] else ("force-push" if result["is_force"] else "push")
        qualifier = " (dry-run/no-op still proves intent)" if result["is_dry_run"] else ""
        reasons.append(
            f"direct stable-branch {verb} detected targeting: {refs}{qualifier}; "
            "worker sessions must never publish stable branches directly"
        )
        result["contamination"] = True
        result["proves_intent"] = True
    elif saw_push and result["ambiguous_target"]:
        # Bare ``git push`` with no refspec: ambiguous. Report, but do not
        # contaminate on the command alone — the root-checkout / branch-state
        # detector resolves whether the current branch is stable.
        reasons.append(
            "ambiguous 'git push' with no refspec: resolve the current branch "
            "before pushing; if it is a stable branch this is forbidden"
        )

    result["reasons"] = reasons
    return result


def detect_stable_push(commands: str | Iterable[str] | None) -> dict[str, Any]:
    """Classify one command or an iterable of commands; contamination if any hit."""
    if commands is None:
        return classify_push_command(None)
    if isinstance(commands, str):
        return classify_push_command(commands)
    worst: dict[str, Any] | None = None
    for cmd in commands:
        res = classify_push_command(cmd)
        if res["contamination"]:
            return res
        if worst is None or (res["is_git_push"] and not worst["is_git_push"]):
            worst = res
    return worst or classify_push_command(None)


# ── root/control checkout local-commit detection ──────────────────────────────

def assess_root_checkout_local_commit(
    *,
    current_branch: str | None,
    head_sha: str | None,
    remote_master_sha: str | None,
    is_under_branches: bool,
    ahead_count: int | None = None,
) -> dict[str, Any]:
    """Detect a local commit on the root/control checkout not on a feature branch.

    #671 AC2. An isolated ``branches/...`` worktree is exempt (that is where
    feature work belongs). On the control checkout, a commit is illegitimate
    when the checkout sits on a stable branch (or detached) and its HEAD has
    advanced past the tracking ``prgs/master`` — i.e. a commit was made that is
    not carried by an issue feature branch/PR.

    Positive evidence only: missing state reports ``unknown`` rather than
    asserting contamination, but an unreadable *branch* on the control checkout
    (detached HEAD) with an advanced HEAD is still flagged.
    """
    branch = _clean(current_branch)
    head = _clean(head_sha).lower()
    master = _clean(remote_master_sha).lower()

    if is_under_branches:
        return {
            "contamination": False,
            "unknown": False,
            "reasons": [],
            "detail": "isolated branches/ worktree — feature commits are expected here",
        }

    reasons: list[str] = []
    unknown = False

    on_stable = (not branch) or (branch in STABLE_BRANCHES)
    if not on_stable:
        # Control checkout is on some non-stable branch: out of scope for this
        # detector (root_checkout_guard #475 handles that contamination class).
        return {
            "contamination": False,
            "unknown": False,
            "reasons": [],
            "detail": f"control checkout on non-stable branch '{branch}' — not this detector's class",
        }

    advanced = False
    if ahead_count is not None and ahead_count > 0:
        advanced = True
    if head and master and head != master:
        advanced = True
    if (not head or not master) and ahead_count is None:
        unknown = True

    if advanced:
        where = f"branch '{branch}'" if branch else "detached HEAD"
        reasons.append(
            f"control checkout ({where}) has local commits not on an issue "
            "feature branch (HEAD advanced past prgs/master); worker sessions "
            "must commit only on issue branches under branches/"
        )

    return {
        "contamination": bool(reasons),
        "unknown": unknown and not reasons,
        "reasons": reasons,
        "current_branch": branch or None,
        "head_sha": head or None,
        "remote_master_sha": master or None,
    }


# ── contamination record + gate ───────────────────────────────────────────────

def build_contamination_record(
    *,
    reason_class: str,
    command_redacted: str | None = None,
    session_id: str | None = None,
    remote: str | None = None,
    ref: str | None = None,
    role: str | None = None,
    detail: str | None = None,
) -> dict[str, Any]:
    """Build the durable contamination marker payload (redacted, audit-safe).

    ``reason_class`` is one of ``stable_branch_push`` / ``root_checkout_commit``.
    The command is stored already-redacted; callers pass the raw line through
    :func:`redact_command` (or this builder redacts a raw ``command`` for them).
    """
    return {
        "kind": CONTAMINATION_KIND,
        "reason_class": _clean(reason_class) or "stable_branch_push",
        "command_summary": redact_command(command_redacted),
        "session_id": _clean(session_id) or None,
        "remote": _clean(remote) or None,
        "ref": _clean(ref) or None,
        "role": _clean(role) or None,
        "detail": _clean(detail) or None,
        "cleared_by_reconciler": False,
    }


def assess_contamination_gate(
    marker: dict[str, Any] | None,
    *,
    task: str | None,
    actual_role: str | None,
) -> dict[str, Any]:
    """Fail closed on gated mutations while a contamination marker is live (#671 AC4).

    * No marker → allowed.
    * Reconciler (audit) role → allowed (the sanctioned path to inspect/clear).
    * Marker present + ``task`` in :data:`CONTAMINATION_GATED_TASKS` → blocked.
    * Marker present + non-gated task (e.g. ``comment_issue``) → allowed, so the
      worker can still post the durable audit/handoff comment.
    """
    if not marker or marker.get("cleared_by_reconciler"):
        return {"block": False, "reasons": [], "task": task}

    role = _clean(actual_role).lower()
    if role == "reconciler":
        return {
            "block": False,
            "reasons": [],
            "task": task,
            "detail": "reconciler audit path is exempt from the contamination gate",
        }

    task_name = _clean(task)
    if task_name and task_name in CONTAMINATION_GATED_TASKS:
        summary = marker.get("command_summary") or marker.get("detail") or "(no summary)"
        reason_class = marker.get("reason_class") or "stable_branch_push"
        return {
            "block": True,
            "reasons": [
                f"session is workflow-contaminated ({reason_class}): {summary}. "
                f"'{task_name}' is blocked until a reconciler audits and clears "
                "the contamination. " + REMEDIATION
            ],
            "task": task_name,
        }

    return {"block": False, "reasons": [], "task": task_name or None}


def format_contamination_gate_error(gate: dict[str, Any]) -> str:
    """Single RuntimeError message for MCP mutation gates."""
    reasons = "; ".join(gate.get("reasons") or ["session workflow-contaminated"])
    return f"Stable-branch contamination gate (#671): {reasons}"