Gitea-Tools/reviewer_worktree.py

"""Fail-closed reviewer worktree and local-git safety proofs (#233).

Reviewer sessions must never stash, reset, or otherwise manipulate unrelated
local changes from another session. When the active worktree has dirty tracked
files outside the PR scope, the workflow must stop or switch to a disposable
scratch worktree (``scripts/worktree-review``).

Git command policy (#243): reviewers use an allowlist, not a blocklist.
Any ``git`` invocation that does not match ``_READONLY_REVIEWER_GIT`` is
forbidden — including ``checkout HEAD --``, ``checkout .``, ``switch``,
and uncommon ``stash`` subcommands that older blocklists missed.
"""

from __future__ import annotations

import re
import shlex

# Read-only git operations reviewers may use for validation (#243 allowlist).
_READONLY_REVIEWER_GIT = re.compile(
    r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
    r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|"
    r"worktree\s+list|worktree\s+add)",
    re.IGNORECASE,
)

_GIT_INVOCATION = re.compile(r"\bgit\b", re.IGNORECASE)


def parse_dirty_tracked_files(porcelain: str) -> list[str]:
    """Return tracked paths with local modifications from ``git status --porcelain``.

    Untracked entries (``??``) are ignored — they do not block reviewer work
    when a scratch worktree is used, and authors may have unrelated untracked
    files without implying reviewer interference.
    """
    paths: list[str] = []
    for line in (porcelain or "").splitlines():
        if not line or len(line) < 4:
            continue
        if line.startswith("??"):
            continue
        path = line[3:].strip()
        if " -> " in path:
            path = path.split(" -> ", 1)[1].strip()
        if path:
            paths.append(path)
    return paths


def files_outside_pr_scope(
    dirty_files: list[str] | None,
    pr_scope_files: list[str] | None,
) -> list[str]:
    """Dirty tracked files not explained by the PR diff file set."""
    dirty = [p for p in (dirty_files or []) if p]
    scope = {p for p in (pr_scope_files or []) if p}
    if not dirty:
        return []
    if not scope:
        return list(dirty)
    return [path for path in dirty if path not in scope]


def _is_git_command(command: str) -> bool:
    return bool(_GIT_INVOCATION.search((command or "").strip()))


def is_readonly_reviewer_git_command(command: str) -> bool:
    """True when the command is an explicitly allowed read-only git operation."""
    text = (command or "").strip()
    if not text or not _is_git_command(text):
        return False
    return bool(_READONLY_REVIEWER_GIT.search(text))


def is_forbidden_reviewer_git_command(command: str) -> bool:
    """True when a git command is not on the reviewer readonly allowlist."""
    text = (command or "").strip()
    if not text or not _is_git_command(text):
        return False
    return not is_readonly_reviewer_git_command(text)


def assess_reviewer_git_command_log(commands: list[str] | None) -> dict:
    """Fail closed when reviewer shell history includes forbidden git mutations."""
    forbidden = [
        cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd)
    ]
    if forbidden:
        return {
            "proven": False,
            "block": True,
            "forbidden_commands": forbidden,
            "reasons": [
                "reviewer workflow executed forbidden local git mutation: "
                f"{cmd!r}"
                for cmd in forbidden
            ],
            "safe_next_action": (
                "stop; report worktree interference; do not stash/reset/checkout "
                "unrelated files — use scripts/worktree-review instead"
            ),
        }
    return {
        "proven": True,
        "block": False,
        "forbidden_commands": [],
        "reasons": [],
        "safe_next_action": "proceed",
    }


def assess_reviewer_worktree_proof(proof: dict | None) -> dict:
    """Evaluate reviewer worktree safety before checkout/diff/validation/review.

    *proof* keys:
    - ``worktree_path`` (required)
    - ``porcelain_status`` or ``dirty_files``
    - ``pr_scope_files`` (paths in the PR diff)
    - ``scratch_used`` (bool)
    - ``scratch_path`` (when scratch_used)
    - ``git_commands`` (shell commands executed this session)
    - ``unrelated_mutations_claimed`` (bool) — stash/reset/drop reported
    """
    proof = dict(proof or {})
    reasons: list[str] = []
    worktree_path = (proof.get("worktree_path") or "").strip()
    if not worktree_path:
        reasons.append("reviewer worktree path not reported; fail closed")

    if proof.get("dirty_files") is not None:
        dirty_files = list(proof.get("dirty_files") or [])
    else:
        dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "")

    pr_scope = list(proof.get("pr_scope_files") or [])
    unrelated = files_outside_pr_scope(dirty_files, pr_scope)
    scratch_used = bool(proof.get("scratch_used"))
    scratch_path = (proof.get("scratch_path") or "").strip()

    is_dirty = bool(dirty_files)
    unrelated_dirty = bool(unrelated)

    if unrelated_dirty and not scratch_used:
        reasons.append(
            "worktree has dirty tracked files outside PR scope "
            f"({', '.join(unrelated)}); stop or use a scratch worktree"
        )
    if scratch_used and not scratch_path:
        reasons.append(
            "scratch worktree was used but scratch_path was not reported"
        )
    if proof.get("unrelated_mutations_claimed"):
        reasons.append(
            "reviewer reported stash/reset/checkout cleanup of unrelated "
            "local changes; this is forbidden"
        )

    command_assessment = assess_reviewer_git_command_log(
        list(proof.get("git_commands") or [])
    )
    if command_assessment["block"]:
        reasons.extend(command_assessment["reasons"])

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "worktree_path": worktree_path or None,
        "is_dirty": is_dirty,
        "dirty_files": dirty_files,
        "unrelated_dirty_files": unrelated,
        "scratch_used": scratch_used,
        "scratch_path": scratch_path or None,
        "unrelated_mutations_avoided": not bool(
            proof.get("unrelated_mutations_claimed")
            or command_assessment.get("forbidden_commands")
        ),
        "safe_next_action": (
            "proceed"
            if proven
            else command_assessment.get("safe_next_action")
            or "stop; use scripts/worktree-review or report dirty worktree"
        ),
        "forbidden_commands": command_assessment.get("forbidden_commands", []),
    }


def assess_author_worktree_continuity(proof: dict | None) -> dict:
    """Authors may keep dirty feature worktrees; reviewers may not manipulate them.

    This helper only proves the task role is author when dirty unrelated files
    exist — it does not grant reviewers an exception.
    """
    proof = dict(proof or {})
    role = (proof.get("task_role") or "").strip().lower()
    dirty_files = list(proof.get("dirty_files") or [])
    if role == "author" and dirty_files:
        return {
            "allowed": True,
            "reasons": [
                "author task may continue with dirty tracked files in its own "
                "worktree; reviewer interference rules do not apply"
            ],
        }
    if role == "reviewer" and dirty_files:
        return assess_reviewer_worktree_proof(proof)
    return {"allowed": True, "reasons": []}