224 lines
7.8 KiB
Python
224 lines
7.8 KiB
Python
"""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)
|
|
|
|
|
|
# #673: Canonical pattern for identifying review worktrees under branches/.
|
|
REVIEW_WORKTREE_RE = re.compile(
|
|
r"branches/(?:review-pr\d+[\w/-]*|merge-simulation-pr\d+|review-[\w-]+)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def is_review_worktree_path(path: str) -> bool:
|
|
"""True when the path belongs to a reviewer or simulation worktree."""
|
|
normalized = (path or "").replace("\\", "/")
|
|
return bool(REVIEW_WORKTREE_RE.search(normalized))
|
|
|
|
|
|
|
|
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": []} |