Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: common anti-stomp preflight before every mutation tool (Closes #604) #680

Open
jcwalker3 wants to merge 2 commits from feat/issue-604-anti-stomp-preflight into master
pull from: feat/issue-604-anti-stomp-preflight
merge into: :master
Conversation 9 Commits 2 Files Changed 4
+2142 -3
4 changed files with 2142 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
anti_stomp_preflight.py
+807
View File
@@ -0,0 +1,807 @@
"""Common anti-stomp preflight for every MCP mutation tool (#604).
Even with leases, mutation tools need a **shared** preflight that prevents
stale sessions, wrong worktrees, wrong repos, old prompts, terminal locks,
foreign leases, contaminated approvals, and root-checkout mutations from
slipping through.
This module is the pure assessment core. Callers (MCP mutation entrypoints)
gather live facts and pass them in — nothing here performs git, network, or
durable-state I/O. Existing happy-path guards remain authoritative; this
module composes them into one typed, fail-closed result.
Required checks (issue #604):
* repo/org verification
* profile/role verification
* root checkout clean and not used for mutation (when role requires branches/)
* worktree under branches when required
* active lease ownership (when lease is required for the mutation)
* terminal lock status (when applicable)
* expected head SHA (when pinned)
* stale runtime status
* workflow hash (when a workflow load is required)
* source contamination status
* no manual state/mtime/source-file bypass
Failure returns a typed blocker and an exact next action. There is **no**
agent-facing bypass flag.
"""
from __future__ import annotations
from typing import Any
import author_mutation_worktree
import master_parity_gate
import remote_repo_guard
import root_checkout_guard
import stable_branch_push_guard
# ── public constants ──────────────────────────────────────────────────────────
# Typed blocker kinds returned in the structured result.
BLOCKER_WRONG_REPO = "wrong_repo"
BLOCKER_WRONG_ROLE = "wrong_role"
BLOCKER_ROOT_CHECKOUT = "root_checkout_mutation"
BLOCKER_WRONG_WORKTREE = "wrong_worktree"
BLOCKER_FOREIGN_LEASE = "foreign_lease"
BLOCKER_TERMINAL_LOCK = "terminal_lock"
BLOCKER_HEAD_SHA = "head_sha_mismatch"
BLOCKER_STALE_RUNTIME = "stale_runtime"
BLOCKER_WORKFLOW_HASH = "workflow_hash"
BLOCKER_SOURCE_CONTAMINATION = "source_contamination"
BLOCKER_MANUAL_BYPASS = "manual_bypass"
BLOCKER_KINDS = frozenset({
BLOCKER_WRONG_REPO,
BLOCKER_WRONG_ROLE,
BLOCKER_ROOT_CHECKOUT,
BLOCKER_WRONG_WORKTREE,
BLOCKER_FOREIGN_LEASE,
BLOCKER_TERMINAL_LOCK,
BLOCKER_HEAD_SHA,
BLOCKER_STALE_RUNTIME,
BLOCKER_WORKFLOW_HASH,
BLOCKER_SOURCE_CONTAMINATION,
BLOCKER_MANUAL_BYPASS,
})
# Mutation tasks that must invoke the shared anti-stomp preflight before
# acting (issue #604 AC1). Every member must appear as a live task= kwarg
# to verify_preflight_purity / _run_anti_stomp_preflight (or the review
# anti_task map) in gitea_mcp_server.py. Inventory↔wiring tests fail if
# a declared task is not wired.
MUTATION_TASKS = frozenset({
"create_issue",
"comment_issue",
"close_issue",
"mark_issue",
"lock_issue",
"set_issue_labels",
"create_label",
"create_pr",
"close_pr",
"edit_pr",
"commit_files",
"gitea_commit_files",
"delete_branch",
"cleanup_merged_pr_branch",
"cleanup_stale_claims",
"reconcile_merged_cleanups",
"reconcile_already_landed_pr",
"reconcile_close_superseded_pr",
"post_heartbeat",
"acquire_reviewer_pr_lease",
"gitea_acquire_reviewer_pr_lease",
"adopt_merger_pr_lease",
"review_pr",
"submit_pr_review",
"approve_pr",
"request_changes_pr",
"merge_pr",
})
# Intentionally excluded from MUTATION_TASKS. Each has a dedicated
# fail-closed gate that preserves decision-lock ownership, workflow-hash,
# head-SHA, and repository consistency. Do not re-add without either
# wiring shared preflight or updating this rationale (#604 Blocker B).
DEDICATED_GATE_MUTATIONS: dict[str, str] = {
"mark_final_review_decision": (
"Local review-decision lock only. Enforced by session lock ownership, "
"workflow-hash, expected head SHA, PR work lease, eligibility, and "
"terminal-lock gates. No Gitea review POST; shared anti-stomp would "
"duplicate without side-effect-ordering value."
),
"save_review_draft": (
"Local session-state draft save with live PR head/base consistency "
"checks. Not a Gitea review/merge mutation; dedicated head-SHA and "
"worktree resolution gates apply."
),
"resume_review_draft": (
"Live-state resume with terminal lock, lease ownership, head/base SHA, "
"and parity checks. When submit=True, delegates to gitea_submit_pr_review "
"which runs shared anti-stomp before the Gitea review POST."
),
"cleanup_stale_review_decision_lock": (
"Moot-lock cleanup with identity match, live PR merged/closed proof, "
"and reviewer capability gate (#594). Distinct from shared anti-stomp."
),
"gitea_cleanup_stale_review_decision_lock": (
"Alias of cleanup_stale_review_decision_lock; dedicated #594 path."
),
"heartbeat_reviewer_pr_lease": (
"Lease heartbeat posts via verify_preflight_purity(task='review_pr') "
"plus in-session lease ownership checks; not a separate mutation class."
),
"release_reviewer_pr_lease": (
"Lease release uses workspace binding + ownership checks; posts only "
"when the session owns the active lease."
),
}
# Roles that must mutate from a branches/ worktree (not the control checkout).
_BRANCHES_REQUIRED_ROLES = frozenset({"author"})
# Reviewer/merger mutations that require an owned lease + head pinning when
# the caller supplies those facts.
_LEASE_AWARE_TASKS = frozenset({
"review_pr",
"submit_pr_review",
"approve_pr",
"request_changes_pr",
"merge_pr",
"acquire_reviewer_pr_lease",
"gitea_acquire_reviewer_pr_lease",
"adopt_merger_pr_lease",
})
_NEXT_ACTIONS: dict[str, str] = {
BLOCKER_WRONG_REPO: (
"Pass explicit org= and repo= matching the local git remote "
"(e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools) and retry."
),
BLOCKER_WRONG_ROLE: (
"Call gitea_resolve_task_capability, switch to the required "
"namespace/profile, re-verify with gitea_whoami, then retry."
),
BLOCKER_ROOT_CHECKOUT: (
"Leave the root control checkout on clean master; create or switch "
"to a session-owned worktree under branches/ and retry the mutation "
"with worktree_path set."
),
BLOCKER_WRONG_WORKTREE: (
"Create or bind a session-owned worktree under branches/, set "
"worktree_path (or GITEA_ACTIVE_WORKTREE), and retry."
),
BLOCKER_FOREIGN_LEASE: (
"Stop. Do not stomp a foreign lease. Wait for expiry, request "
"takeover through the sanctioned path, or hand off to the owner."
),
BLOCKER_TERMINAL_LOCK: (
"Stop. A terminal review-decision lock is active for this head. "
"Do not re-approve/merge; follow the #332/#620 recovery path."
),
BLOCKER_HEAD_SHA: (
"Re-fetch the live PR head with gitea_view_pr, re-pin "
"expected_head_sha to the current head, re-validate, then retry."
),
BLOCKER_STALE_RUNTIME: (
"Restart the Gitea MCP server so it reloads master's capability "
"gates, re-run gitea_whoami + gitea_resolve_task_capability, then retry."
),
BLOCKER_WORKFLOW_HASH: (
"Reload the canonical workflow via gitea_load_review_workflow, then "
"rerun the full review-merge workflow from inventory (no approve/merge replay)."
),
BLOCKER_SOURCE_CONTAMINATION: (
"Stop. Session is source-contaminated. Hand off to a reconciler for "
"audit/clear; do not clear markers by deleting session-state files."
),
BLOCKER_MANUAL_BYPASS: (
"Stop. Manual state/mtime/source-file bypass is forbidden. Use "
"sanctioned MCP tools only; never delete or rewrite session-state "
"or lock files by hand."
),
}
def is_mutation_task(task: str | None) -> bool:
"""True when *task* is in the #604 mutation set (normalized)."""
name = (task or "").strip().lower().removeprefix("gitea_")
if not name:
return False
if name in MUTATION_TASKS:
return True
# Accept gitea_ prefixed membership for aliases already in the set.
return f"gitea_{name}" in MUTATION_TASKS
def roles_compatible(active_role: str | None, required_role: str | None) -> bool:
"""Whether *active_role* may perform a task that maps to *required_role*.
Exact match always passes. Reconcilers may run author-class mutations
(comment/close/cleanup) that the capability map stamps as ``author`` —
matching the long-standing reconciler exemption for control-checkout
work. No other cross-role substitutions are allowed.
Capability-authorized multi-role tasks (e.g. reviewer holding
``gitea.issue.comment`` for ``comment_issue``) are handled by
:func:`authorization_compatible`, not by expanding this role matrix.
"""
active = (active_role or "").strip().lower()
required = (required_role or "").strip().lower()
if not active or not required:
return True
if active == required:
return True
if active == "reconciler" and required in {"author", "reconciler"}:
return True
return False
def authorization_compatible(
active_role: str | None,
required_role: str | None,
*,
required_permission: str | None = None,
allowed_operations: Any = None,
) -> bool:
"""Whether the active session may run a task given role *and* capability.
Order:
1. :func:`roles_compatible` (exact role or narrow reconciler→author).
2. Capability possession: when *required_permission* is non-empty and
present in *allowed_operations*, allow even if the nominal task role
differs (e.g. reviewer/merger with ``gitea.issue.comment`` on
``comment_issue`` / ``mark_issue`` / ``set_issue_labels`` /
``lock_issue``).
Fail closed otherwise. This is **not** a broad reviewer→author or
merger→author role rewrite: without the specific permission the check
still denies. Missing *allowed_operations* when a permission is required
also fails closed (callers must supply the live profile op list).
"""
if roles_compatible(active_role, required_role):
return True
perm = (required_permission or "").strip()
if not perm:
return False
if allowed_operations is None:
return False
try:
ops = {str(o).strip() for o in allowed_operations if o is not None}
except TypeError:
return False
return perm in ops
def _blocker(
kind: str,
reasons: list[str],
*,
exact_next_action: str | None = None,
detail: dict | None = None,
) -> dict[str, Any]:
if kind not in BLOCKER_KINDS:
raise ValueError(f"unknown anti-stomp blocker kind: {kind!r}")
return {
"kind": kind,
"reasons": list(reasons),
"exact_next_action": exact_next_action or _NEXT_ACTIONS[kind],
"detail": dict(detail or {}),
}
def _allowed_result(checks: dict[str, Any]) -> dict[str, Any]:
return {
"allowed": True,
"block": False,
"blockers": [],
"reasons": [],
"exact_next_action": "proceed",
"blocker_kind": None,
"checks": checks,
}
def _blocked_result(
blockers: list[dict[str, Any]],
checks: dict[str, Any],
) -> dict[str, Any]:
primary = blockers[0]
all_reasons: list[str] = []
for b in blockers:
for r in b.get("reasons") or []:
if r not in all_reasons:
all_reasons.append(r)
return {
"allowed": False,
"block": True,
"blockers": blockers,
"reasons": all_reasons,
"exact_next_action": primary["exact_next_action"],
"blocker_kind": primary["kind"],
"checks": checks,
}
def assess_anti_stomp_preflight(
*,
task: str | None = None,
# repo/org
remote: str | None = None,
resolved_org: str | None = None,
resolved_repo: str | None = None,
local_remote_url: str | None = None,
org_explicit: bool = False,
repo_explicit: bool = False,
check_repo: bool = True,
# profile/role + capability
profile_name: str | None = None,
profile_role: str | None = None,
required_role: str | None = None,
required_permission: str | None = None,
allowed_operations: Any = None,
check_role: bool = True,
# root checkout + worktree
workspace_path: str | None = None,
project_root: str | None = None,
current_branch: str | None = None,
root_head_sha: str | None = None,
root_porcelain: str | None = None,
remote_master_sha: str | None = None,
check_root_checkout: bool = True,
check_worktree: bool = True,
# stale runtime (master parity)
startup_head: str | None = None,
current_code_head: str | None = None,
check_stale_runtime: bool = True,
# lease ownership
lease_required: bool = False,
foreign_lease: bool | None = None,
lease_owner_session: str | None = None,
active_session_id: str | None = None,
lease_owner_identity: str | None = None,
active_identity: str | None = None,
lease_reasons: list[str] | None = None,
# terminal lock
terminal_lock_blocks: bool | None = None,
terminal_lock_reasons: list[str] | None = None,
# expected head SHA
require_head_sha: bool = False,
expected_head_sha: str | None = None,
live_head_sha: str | None = None,
# workflow hash
workflow_hash_valid: bool | None = None,
workflow_hash_reasons: list[str] | None = None,
# source contamination (stable-branch push / approval contamination)
source_contaminated: bool | None = None,
contamination_reasons: list[str] | None = None,
# manual bypass
manual_bypass_attempted: bool = False,
manual_bypass_reasons: list[str] | None = None,
) -> dict[str, Any]:
"""Fail-closed common anti-stomp assessment (pure).
Only checks for which facts are supplied (or which are required by the
mutation category) are evaluated. Unsupplied optional facts are recorded
as ``skipped`` so callers can see coverage without inventing defaults.
Returns a structured dict with ``allowed``/``block``, typed ``blockers``,
aggregated ``reasons``, ``exact_next_action``, and per-check ``checks``.
"""
task_name = (task or "").strip()
role = (profile_role or "").strip().lower() or None
req_role = (required_role or "").strip().lower() or None
req_perm = (required_permission or "").strip() or None
checks: dict[str, Any] = {
"task": task_name or None,
"profile_name": profile_name,
"profile_role": role,
"required_role": req_role,
"required_permission": req_perm,
}
blockers: list[dict[str, Any]] = []
# ── manual bypass (always first; never skippable when attempted) ─────────
if manual_bypass_attempted:
reasons = list(manual_bypass_reasons or [
"manual state/mtime/source-file bypass attempt detected (fail closed)"
])
blockers.append(_blocker(BLOCKER_MANUAL_BYPASS, reasons))
checks["manual_bypass"] = {"block": True, "reasons": reasons}
else:
checks["manual_bypass"] = {"block": False, "skipped": False}
# ── repo/org ─────────────────────────────────────────────────────────────
if check_repo and resolved_org is not None and resolved_repo is not None:
repo_assessment = remote_repo_guard.assess_remote_repo_match(
remote=remote or "",
resolved_org=resolved_org,
resolved_repo=resolved_repo,
local_remote_url=local_remote_url,
org_explicit=org_explicit,
repo_explicit=repo_explicit,
)
checks["repo"] = {
"block": bool(repo_assessment.get("block")),
"reasons": list(repo_assessment.get("reasons") or []),
"resolved_org": resolved_org,
"resolved_repo": resolved_repo,
}
if repo_assessment.get("block"):
blockers.append(
_blocker(
BLOCKER_WRONG_REPO,
list(repo_assessment.get("reasons") or ["repo/org mismatch"]),
detail={
"resolved_org": resolved_org,
"resolved_repo": resolved_repo,
"local_remote_url": local_remote_url,
},
)
)
else:
checks["repo"] = {"block": False, "skipped": True}
# ── profile/role + capability ────────────────────────────────────────────
# Role match OR possession of the task's required permission (Blocker A).
# Reviewer/merger may run issue-comment-class tasks when they hold
# gitea.issue.comment; unauthorized escalation without the permission
# still fails closed.
if check_role and req_role and role:
authorized = authorization_compatible(
role,
req_role,
required_permission=req_perm,
allowed_operations=allowed_operations,
)
if not authorized:
perm_clause = (
f", required_permission='{req_perm}'" if req_perm else ""
)
reasons = [
f"active profile role '{role}' is not authorized for task "
f"'{task_name or '(unknown)'}' (required_role='{req_role}'"
f"{perm_clause}; profile={profile_name or '(unknown)'})"
]
checks["role"] = {
"block": True,
"reasons": reasons,
"required_permission": req_perm,
"capability_authorized": False,
}
blockers.append(
_blocker(
BLOCKER_WRONG_ROLE,
reasons,
detail={
"profile_name": profile_name,
"profile_role": role,
"required_role": req_role,
"required_permission": req_perm,
},
)
)
else:
checks["role"] = {
"block": False,
"reasons": [],
"required_permission": req_perm,
"capability_authorized": bool(
req_perm
and allowed_operations is not None
and req_perm in {
str(o).strip() for o in (allowed_operations or [])
if o is not None
}
and not roles_compatible(role, req_role)
),
}
else:
checks["role"] = {"block": False, "skipped": True}
# ── stale runtime (master parity) ────────────────────────────────────────
if check_stale_runtime and (
startup_head is not None or current_code_head is not None
):
parity = master_parity_gate.assess_master_parity(
{"startup_head": startup_head},
current_code_head,
)
stale_reasons = master_parity_gate.parity_block_reasons(parity)
checks["stale_runtime"] = {
"block": bool(stale_reasons),
"reasons": list(stale_reasons),
"startup_head": startup_head,
"current_code_head": current_code_head,
"stale": bool(parity.get("stale")),
}
if stale_reasons:
blockers.append(
_blocker(
BLOCKER_STALE_RUNTIME,
list(stale_reasons),
detail={
"startup_head": startup_head,
"current_code_head": current_code_head,
},
)
)
else:
checks["stale_runtime"] = {"block": False, "skipped": True}
# ── root checkout ────────────────────────────────────────────────────────
if (
check_root_checkout
and workspace_path is not None
and project_root is not None
and root_porcelain is not None
):
root_assessment = root_checkout_guard.assess_root_checkout_guard(
workspace_path=workspace_path,
canonical_repo_root=project_root,
current_branch=current_branch,
head_sha=root_head_sha,
porcelain_status=root_porcelain,
remote_master_sha=remote_master_sha,
resolved_role=req_role or role,
actual_role=role,
)
checks["root_checkout"] = {
"block": bool(root_assessment.get("block")),
"reasons": list(root_assessment.get("reasons") or []),
}
if root_assessment.get("block"):
blockers.append(
_blocker(
BLOCKER_ROOT_CHECKOUT,
list(root_assessment.get("reasons") or [
"control checkout is not clean master"
]),
detail={
"workspace_path": workspace_path,
"project_root": project_root,
"current_branch": current_branch,
},
)
)
else:
checks["root_checkout"] = {"block": False, "skipped": True}
# ── worktree under branches (author) ─────────────────────────────────────
worktree_role = role or req_role
if (
check_worktree
and workspace_path is not None
and project_root is not None
and worktree_role in _BRANCHES_REQUIRED_ROLES
):
wt = author_mutation_worktree.assess_author_mutation_worktree(
workspace_path=workspace_path,
project_root=project_root,
current_branch=current_branch,
)
checks["worktree"] = {
"block": bool(wt.get("block")),
"reasons": list(wt.get("reasons") or []),
"under_branches": wt.get("under_branches"),
}
if wt.get("block"):
blockers.append(
_blocker(
BLOCKER_WRONG_WORKTREE,
list(wt.get("reasons") or [
"mutation worktree is not under branches/"
]),
detail={
"workspace_path": workspace_path,
"project_root": project_root,
},
)
)
else:
checks["worktree"] = {
"block": False,
"skipped": worktree_role not in _BRANCHES_REQUIRED_ROLES
or workspace_path is None,
}
# ── foreign lease ────────────────────────────────────────────────────────
lease_needed = lease_required or (
task_name.removeprefix("gitea_") in {
t.removeprefix("gitea_") for t in _LEASE_AWARE_TASKS
}
and foreign_lease is not None
)
if lease_needed or foreign_lease is True:
is_foreign = bool(foreign_lease)
if foreign_lease is None and lease_required:
# Ownership must be proven when lease_required; missing ownership
# facts fail closed.
owner_ok = (
(not lease_owner_session and not active_session_id)
or (
lease_owner_session
and active_session_id
and lease_owner_session == active_session_id
)
)
identity_ok = (
(not lease_owner_identity and not active_identity)
or (
lease_owner_identity
and active_identity
and lease_owner_identity == active_identity
)
)
if not owner_ok or not identity_ok:
is_foreign = True
reasons = list(lease_reasons or [])
if is_foreign and not reasons:
reasons = [
"active lease is owned by a foreign session or identity "
"(anti-stomp fail closed)"
]
checks["lease"] = {
"block": is_foreign,
"reasons": reasons if is_foreign else [],
"lease_owner_session": lease_owner_session,
"active_session_id": active_session_id,
}
if is_foreign:
blockers.append(
_blocker(BLOCKER_FOREIGN_LEASE, reasons)
)
else:
checks["lease"] = {"block": False, "skipped": True}
# ── terminal lock ────────────────────────────────────────────────────────
if terminal_lock_blocks is not None:
reasons = list(terminal_lock_reasons or [])
if terminal_lock_blocks and not reasons:
reasons = [
"terminal review-decision lock is active for this head "
"(anti-stomp fail closed)"
]
checks["terminal_lock"] = {
"block": bool(terminal_lock_blocks),
"reasons": reasons if terminal_lock_blocks else [],
}
if terminal_lock_blocks:
blockers.append(_blocker(BLOCKER_TERMINAL_LOCK, reasons))
else:
checks["terminal_lock"] = {"block": False, "skipped": True}
# ── expected head SHA ────────────────────────────────────────────────────
if require_head_sha or (
expected_head_sha is not None and live_head_sha is not None
):
exp = (expected_head_sha or "").strip().lower()
live = (live_head_sha or "").strip().lower()
mismatch = False
reasons: list[str] = []
if require_head_sha and not exp:
mismatch = True
reasons.append(
"expected_head_sha is required before this mutation "
"(anti-stomp fail closed)"
)
elif exp and live and exp != live:
mismatch = True
reasons.append(
f"expected_head_sha '{exp}' does not match live PR head "
f"'{live}' (anti-stomp fail closed; stale prompt data)"
)
elif require_head_sha and exp and not live:
mismatch = True
reasons.append(
"live PR head SHA could not be determined to corroborate "
"expected_head_sha (anti-stomp fail closed)"
)
checks["head_sha"] = {
"block": mismatch,
"reasons": reasons,
"expected_head_sha": expected_head_sha,
"live_head_sha": live_head_sha,
}
if mismatch:
blockers.append(_blocker(BLOCKER_HEAD_SHA, reasons))
else:
checks["head_sha"] = {"block": False, "skipped": True}
# ── workflow hash ────────────────────────────────────────────────────────
if workflow_hash_valid is not None:
reasons = list(workflow_hash_reasons or [])
if not workflow_hash_valid and not reasons:
reasons = [
"workflow load proof missing or hash stale "
"(anti-stomp fail closed)"
]
checks["workflow_hash"] = {
"block": not bool(workflow_hash_valid),
"reasons": reasons if not workflow_hash_valid else [],
}
if not workflow_hash_valid:
blockers.append(_blocker(BLOCKER_WORKFLOW_HASH, reasons))
else:
checks["workflow_hash"] = {"block": False, "skipped": True}
# ── source contamination ─────────────────────────────────────────────────
if source_contaminated is not None:
reasons = list(contamination_reasons or [])
if source_contaminated and not reasons:
reasons = [
"session is source-contaminated (stable-branch push or "
"contaminated approval path); anti-stomp fail closed"
]
checks["source_contamination"] = {
"block": bool(source_contaminated),
"reasons": reasons if source_contaminated else [],
}
if source_contaminated:
blockers.append(
_blocker(BLOCKER_SOURCE_CONTAMINATION, reasons)
)
else:
checks["source_contamination"] = {"block": False, "skipped": True}
if blockers:
return _blocked_result(blockers, checks)
return _allowed_result(checks)
def format_anti_stomp_error(assessment: dict[str, Any]) -> str:
"""Single RuntimeError message for MCP mutation gates."""
kind = assessment.get("blocker_kind") or "unknown"
reasons = "; ".join(
assessment.get("reasons")
or ["anti-stomp preflight blocked the mutation"]
)
next_action = assessment.get("exact_next_action") or "stop and diagnose"
return (
f"Anti-stomp preflight (#604) blocked mutation "
f"[{kind}]: {reasons}. "
f"exact_next_action: {next_action}"
)
def block_response(
assessment: dict[str, Any],
**extra_fields: Any,
) -> dict[str, Any]:
"""Structured tool-return payload for a blocked mutation."""
payload = {
"success": False,
"performed": False,
"blocked": True,
"anti_stomp": True,
"blocker_kind": assessment.get("blocker_kind"),
"blockers": list(assessment.get("blockers") or []),
"reasons": list(assessment.get("reasons") or []),
"exact_next_action": assessment.get("exact_next_action"),
"checks": assessment.get("checks") or {},
}
payload.update(extra_fields)
return payload
def contamination_from_stable_marker(
marker: dict | None,
*,
task: str | None,
actual_role: str | None,
) -> tuple[bool, list[str]]:
"""Derive source-contamination facts from a #671 durable marker."""
if not marker:
return False, []
gate = stable_branch_push_guard.assess_contamination_gate(
marker,
task=task,
actual_role=actual_role,
)
if gate.get("block"):
return True, list(gate.get("reasons") or [])
return False, []
gitea_mcp_server.py
+282 -3
View File
@@ -619,18 +619,219 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
)
def _anti_stomp_in_test_mode() -> bool:
"""Whether the #604 anti-stomp gate is skipped under pytest.
Mirrors remote-repo / stable-contamination test bypasses so the existing
suite (bare remotes, mocked APIs) stays green. Set
``GITEA_TEST_FORCE_ANTI_STOMP=1`` to exercise the live gate under tests.
"""
if not _preflight_in_test_mode():
return False
return not bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
def _run_anti_stomp_preflight(
task: str | None,
*,
remote: str | None = None,
worktree_path: str | None = None,
host: str | None = None,
org: str | None = None,
repo: str | None = None,
expected_head_sha: str | None = None,
live_head_sha: str | None = None,
require_head_sha: bool = False,
lease_required: bool = False,
foreign_lease: bool | None = None,
lease_reasons: list[str] | None = None,
terminal_lock_blocks: bool | None = None,
terminal_lock_reasons: list[str] | None = None,
workflow_hash_valid: bool | None = None,
workflow_hash_reasons: list[str] | None = None,
raise_on_block: bool = True,
) -> dict | None:
"""Shared #604 anti-stomp preflight for MCP mutation entrypoints.
Gathers live workspace/parity/role/repo facts and invokes the pure
:func:`anti_stomp_preflight.assess_anti_stomp_preflight` assessor. On
block, raises ``RuntimeError`` (default) or returns a structured
:func:`anti_stomp_preflight.block_response` when *raise_on_block* is
False. Returns ``None`` when the mutation may proceed.
"""
if _anti_stomp_in_test_mode():
return None
# Only enforce when the caller names a known mutation task. Read-only
# paths and legacy verify_preflight_purity calls without a task stay
# on the pre-#604 gate chain (whoami/capability/root/worktree).
if not task or not anti_stomp_preflight.is_mutation_task(task):
return None
profile = get_profile()
profile_name = profile.get("profile_name")
profile_role = _actual_profile_role()
allowed_operations = list(profile.get("allowed_operations") or [])
req_role = None
req_perm = None
if task:
try:
req_role = task_capability_map.required_role(task)
except KeyError:
req_role = _preflight_resolved_role
try:
req_perm = task_capability_map.required_permission(task)
except KeyError:
req_perm = None
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
canonical_root = ctx["canonical_repo_root"]
git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
# Repo/org facts (best-effort; explicit org/repo when provided).
resolved_org = org
resolved_repo = repo
local_remote_url = None
org_explicit = org is not None
repo_explicit = repo is not None
if remote:
try:
local_remote_url = _local_git_remote_url(remote)
except Exception:
local_remote_url = None
if resolved_org is None or resolved_repo is None:
try:
rem = _effective_remote(remote)
if rem in REMOTES:
if resolved_org is None:
resolved_org = REMOTES[rem]["org"]
if resolved_repo is None:
resolved_repo = REMOTES[rem]["repo"]
except Exception:
pass
parity = _current_master_parity()
startup_head = parity.get("startup_head")
current_code_head = parity.get("current_head")
# Source contamination from durable #671 marker (role-aware).
source_contaminated = None
contamination_reasons = None
try:
marker = _load_stable_contamination_marker(remote)
contaminated, cont_reasons = anti_stomp_preflight.contamination_from_stable_marker(
marker,
task=task,
actual_role=profile_role,
)
if marker is not None:
source_contaminated = contaminated
contamination_reasons = cont_reasons
except Exception:
# Fail soft on marker I/O: existing #671 enforcer still runs separately.
source_contaminated = None
# Workflow-hash facts when the task is review/merge oriented.
if workflow_hash_valid is None and task in {
"review_pr",
"submit_pr_review",
"approve_pr",
"request_changes_pr",
"merge_pr",
}:
try:
wf_reasons = _review_workflow_load_gate_reasons()
workflow_hash_valid = not bool(wf_reasons)
workflow_hash_reasons = list(wf_reasons) if wf_reasons else None
except Exception:
pass
# Mirror remote_repo_guard's pytest bypass: under the unit suite bare
# remote=prgs still defaults to Timesheet, and re-checking here would
# break happy-path reconciler/author tests that already rely on the
# #530 gate being opt-in via GITEA_FORCE_REMOTE_REPO_CHECK.
check_repo = True
if "pytest" in sys.modules and not os.environ.get(
"GITEA_FORCE_REMOTE_REPO_CHECK"
):
check_repo = False
assessment = anti_stomp_preflight.assess_anti_stomp_preflight(
task=task,
remote=remote,
resolved_org=resolved_org,
resolved_repo=resolved_repo,
local_remote_url=local_remote_url,
org_explicit=org_explicit,
repo_explicit=repo_explicit,
check_repo=check_repo,
profile_name=profile_name,
profile_role=profile_role,
required_role=req_role or _preflight_resolved_role,
required_permission=req_perm,
allowed_operations=allowed_operations,
workspace_path=workspace,
project_root=canonical_root,
current_branch=git_state.get("current_branch"),
root_head_sha=git_state.get("head_sha"),
root_porcelain=git_state.get("porcelain_status") or "",
remote_master_sha=remote_master_sha,
startup_head=startup_head,
current_code_head=current_code_head,
lease_required=lease_required,
foreign_lease=foreign_lease,
lease_reasons=lease_reasons,
terminal_lock_blocks=terminal_lock_blocks,
terminal_lock_reasons=terminal_lock_reasons,
require_head_sha=require_head_sha,
expected_head_sha=expected_head_sha,
live_head_sha=live_head_sha,
workflow_hash_valid=workflow_hash_valid,
workflow_hash_reasons=workflow_hash_reasons,
source_contaminated=source_contaminated,
contamination_reasons=contamination_reasons,
manual_bypass_attempted=False,
)
if not assessment.get("block"):
return None
if raise_on_block:
raise RuntimeError(anti_stomp_preflight.format_anti_stomp_error(assessment))
return anti_stomp_preflight.block_response(assessment)
def verify_preflight_purity(
remote: str | None = None,
worktree_path: str | None = None,
task: str | None = None,
*,
expected_head_sha: str | None = None,
live_head_sha: str | None = None,
require_head_sha: bool = False,
lease_required: bool = False,
foreign_lease: bool | None = None,
lease_reasons: list[str] | None = None,
terminal_lock_blocks: bool | None = None,
terminal_lock_reasons: list[str] | None = None,
workflow_hash_valid: bool | None = None,
workflow_hash_reasons: list[str] | None = None,
org: str | None = None,
repo: str | None = None,
):
"""Verify that identity and capability were verified prior to session edits."""
"""Verify that identity and capability were verified prior to session edits.
Also runs the shared #604 anti-stomp preflight for mutation tasks (typed
blocker + exact next action). Extended lease/head/terminal facts may be
supplied by review/merge callers.
"""
global _preflight_reviewer_violation_files
in_test = _preflight_in_test_mode()
if in_test and not (
os.environ.get("GITEA_TEST_FORCE_DIRTY")
or os.environ.get("GITEA_TEST_PORCELAIN") is not None
or os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP")
):
return
@@ -717,6 +918,30 @@ def verify_preflight_purity(
_enforce_root_checkout_guard(worktree_path)
_enforce_branches_only_author_mutation(worktree_path)
# #604: common anti-stomp preflight (typed blocker + exact next action).
# Runs after the legacy enforcers so existing happy paths stay green and
# the shared module remains the single fail-closed composition point for
# repo/role/worktree/lease/terminal-lock/head/stale/workflow/contamination.
_run_anti_stomp_preflight(
task,
remote=remote,
worktree_path=worktree_path,
org=org,
repo=repo,
expected_head_sha=expected_head_sha,
live_head_sha=live_head_sha,
require_head_sha=require_head_sha,
lease_required=lease_required,
foreign_lease=foreign_lease,
lease_reasons=lease_reasons,
terminal_lock_blocks=terminal_lock_blocks,
terminal_lock_reasons=terminal_lock_reasons,
workflow_hash_valid=workflow_hash_valid,
workflow_hash_reasons=workflow_hash_reasons,
raise_on_block=True,
)
_clear_preflight_capability_state()
@@ -930,6 +1155,7 @@ import author_mutation_worktree # noqa: E402
import root_checkout_guard # noqa: E402
import stable_branch_push_guard # noqa: E402
import remote_repo_guard # noqa: E402
import anti_stomp_preflight # noqa: E402
import issue_claim_heartbeat # noqa: E402
import issue_work_duplicate_gate # noqa: E402
import issue_workflow_labels # noqa: E402
@@ -3833,6 +4059,30 @@ def _evaluate_pr_review_submission(
result["pr_work_lease"] = lease_block
return result
if live:
# #604: common anti-stomp with live head + lease proof (typed blocker).
anti_task = {
"approve": "approve_pr",
"request_changes": "request_changes_pr",
"comment": "submit_pr_review",
}.get(action, "submit_pr_review")
try:
_run_anti_stomp_preflight(
anti_task,
remote=remote,
worktree_path=worktree_path,
org=org,
repo=repo,
expected_head_sha=pinned_sha,
live_head_sha=actual_sha,
require_head_sha=True,
foreign_lease=False,
terminal_lock_blocks=False,
)
except RuntimeError as exc:
reasons.append(str(exc))
return result
if (body or "").strip():
gate = _canonical_comment_gate(body, context="pr_review")
if gate["blocked"]:
@@ -4787,7 +5037,12 @@ def gitea_edit_pr(
raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
closing = payload.get("state") == "closed"
verify_preflight_purity(remote, task="close_pr" if closing else None)
# Closing uses close_pr; non-closing title/body/base edits use edit_pr so
# the shared #604 anti-stomp inventory stays consistent with wiring.
if closing:
verify_preflight_purity(remote, task="close_pr")
else:
verify_preflight_purity(remote, task="edit_pr")
# PR closure is a first-class capability, distinct from retitling or
# rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
@@ -5269,6 +5524,25 @@ def gitea_merge_pr(
reasons.extend(lease_block.get("reasons") or [])
result["pr_work_lease"] = lease_block
return result
# #604: common anti-stomp with live head + lease proof (typed blocker).
try:
_run_anti_stomp_preflight(
"merge_pr",
remote=remote,
worktree_path=worktree_path,
org=org,
repo=repo,
expected_head_sha=expected_head_sha,
live_head_sha=actual_sha,
require_head_sha=True,
foreign_lease=False,
terminal_lock_blocks=False,
)
except RuntimeError as exc:
reasons.append(str(exc))
return result
if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
reasons.append(
"expected head SHA does not match current PR head (fail closed)"
@@ -7388,7 +7662,11 @@ def gitea_acquire_reviewer_pr_lease(
"permission_report": _permission_block_report("gitea.pr.comment"),
}
_verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
# task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
# anti-stomp for the declared lease-acquire mutation inventory entry.
_verify_role_mutation_workspace(
remote, worktree=worktree, task="acquire_reviewer_pr_lease"
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
profile = get_profile()
@@ -7529,6 +7807,7 @@ def gitea_adopt_merger_pr_lease(
"permission_report": _permission_block_report("gitea.pr.merge"),
}
# task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp.
_verify_role_mutation_workspace(
remote, worktree=worktree, task="adopt_merger_pr_lease"
)
task_capability_map.py
+21
View File
@@ -60,6 +60,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.pr.close",
"role": "author",
},
# Non-closing PR metadata edits (title/body/base). Closing uses close_pr.
"edit_pr": {
"permission": "gitea.pr.create",
"role": "author",
},
"gitea_edit_pr": {
"permission": "gitea.pr.create",
"role": "author",
},
"address_pr_change_requests": {
"permission": "gitea.branch.push",
"role": "author",
@@ -68,10 +77,22 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.pr.review",
"role": "reviewer",
},
"submit_pr_review": {
"permission": "gitea.pr.review",
"role": "reviewer",
},
"merge_pr": {
"permission": "gitea.pr.merge",
"role": "merger",
},
"acquire_reviewer_pr_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"gitea_acquire_reviewer_pr_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"adopt_merger_pr_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
tests/test_anti_stomp_preflight.py
+1032
View File
File diff suppressed because it is too large Load Diff