Compare commits

..
Author SHA1 Message Date
sysadmin a1e5c33bfc feat(guard): native MCP transport binding and contaminated-review quarantine (Closes #695)
Bind mutation/credential paths to a process-local native MCP runtime so env
spoofing, direct imports, and offline helpers cannot reconstruct session gates
after native transport failure. Quarantine contaminated formal reviews under
controller authority and honor quarantine in review feedback, merge eligibility,
merge mutation, and canonical handoff validation. Add regression coverage for
the second (PR #694 / review 427) incident class.
2026-07-13 05:57:42 -04:00
15 changed files with 107 additions and 1643 deletions
+8 -23
View File
@@ -9,37 +9,22 @@ formal reviews then looked identical to native approvals.
## Rule
Mutation auth, keychain fill, and controller quarantine require a **production
native MCP transport runtime** established only by:
1. the **resolved absolute path** of the canonical entrypoint
(`mcp_server.py` / `gitea_mcp_server.py` next to `mcp_daemon_guard.py`), and
2. a live **transport bind** (`bind_native_mcp_transport(transport="stdio")`)
immediately before `mcp.run`.
Basename-only trust (a renamed file called `mcp_server.py`), caller-controlled
flags (there is **no** `allow_test_bootstrap`), environment variables, stack
frame spoofing, or import-only launch are insufficient.
Mutation auth, keychain fill, and controller quarantine require a **native MCP
transport runtime** established only by the official entrypoint.
| Context | Allowed |
|---------|---------|
| Official IDE-native MCP: resolved canonical entrypoint marks + binds stdio, holds process-local runtime token | yes |
| pytest (hermetic unit tests) via `is_pytest_runtime()` | yes for unit gates |
| `install_test_native_runtime()` under pytest (test-mode record) | unit-test transport gates only — **never** production Gitea mutations |
| `allow_test_bootstrap=True` (removed; must not exist) | **no** |
| Renamed runner basename `mcp_server.py` outside package root | **no** |
| Import/launch of real entrypoint without transport bind | **no** |
| Official MCP entrypoint (`mcp_server.py` / `gitea_mcp_server` `__main__`) calls `mark_sanctioned_daemon()` and holds a process-local runtime token | yes |
| pytest (hermetic unit tests) | yes |
| `GITEA_MCP_SANCTIONED_DAEMON=1` alone (no process-local native runtime) | **no** (#695) |
| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set; never authorizes mutations (#695 AC1 / PR #701) |
| Override `GITEA_MCP_SESSION_STATE_DIR` mid-session | **no** — production bind pins state root; redirect cannot forge independent decision locks (#695 AC2 / PR #701) |
| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set in agent sessions |
| `GITEA_ALLOW_KEYCHAIN_CLI=1` in LLM sessions | **no** — human operator only |
| bare `python -c 'import gitea_mcp_server; …'` or offline runners | **no** |
| keychain fill outside native/pytest | **no** |
Native runtime is **process-local**: a random token bound to the daemon PID and
transport phase. It is never reconstructed from environment variables,
session-state files, caller-controlled flags, or importing internals in a
fresh Python process.
Native runtime is **process-local**: a random token bound to the daemon PID.
It is never reconstructed from environment variables, session-state files, or
importing internals in a fresh Python process.
## Contaminated review quarantine (#695 AC8)
+9 -94
View File
@@ -134,22 +134,16 @@ _TARGET_BRANCH_SHA_RE = re.compile(
r"target branch sha\s*:\s*[0-9a-f]{40}",
re.IGNORECASE,
)
# #698: structured proof is rendered in several equivalent shapes —
# `workflow_hash: abc...`, `workflow_hash=abc...`, or JSON
# `"workflow_hash": "abc..."`. Recognize all of them; demanding one exact
# punctuation style rejects legitimate structured workflow-load proof.
_WORKFLOW_LOAD_HELPER_RE = re.compile(
r"workflow[-_ ]load[-_ ]helper[-_ ]result\s*[:=]",
r"workflow[- ]load helper result\s*:",
re.IGNORECASE,
)
_WORKFLOW_LOAD_HASH_RE = re.compile(
r"workflow[-_ ]load[-_ ]helper[-_ ]result[\s\S]{0,400}?"
r"workflow[_ ]hash\"?\s*[:=]\s*\"?[0-9a-f]{12}",
r"workflow[- ]load helper result[\s\S]{0,400}?workflow[_ ]hash\s*:\s*[0-9a-f]{12}",
re.IGNORECASE,
)
_WORKFLOW_LOAD_BOUNDARY_RE = re.compile(
r"workflow[-_ ]load[-_ ]helper[-_ ]result[\s\S]{0,400}?"
r"boundary[_ ]status\"?\s*[:=]\s*\"?(?:clean|violation)",
r"workflow[- ]load helper result[\s\S]{0,400}?boundary[_ ]status\s*:\s*(?:clean|violation)",
re.IGNORECASE,
)
_WORKFLOW_FILE_VIEW_NARRATIVE_RE = re.compile(
@@ -250,59 +244,6 @@ def _normalize_task_kind(task_kind: str | None) -> str:
return _TASK_KIND_ALIASES.get(raw, raw)
def _iter_action_entries(action_log: list | None) -> list[dict]:
"""Yield only structured (dict) action-log entries (#698).
Callers must never crash on malformed entries (strings, numbers, null)
that reach the validator from LLM-composed or partially parsed logs;
:func:`sanitize_action_log` reports them separately.
"""
return [e for e in (action_log or []) if isinstance(e, dict)]
def sanitize_action_log(
action_log: list | None,
) -> tuple[list[dict], list[dict[str, str]]]:
"""Split an action log into structured entries and sanitized findings (#698).
Malformed entries become clear, sanitized ``warning`` findings — the
offending value's content is never echoed back (only its position and
type), so secrets or garbage in a broken log cannot leak into validation
errors, and validation itself proceeds without secondary exceptions.
"""
if action_log is None:
return [], []
if not isinstance(action_log, (list, tuple)):
return [], [
validator_finding(
"shared.action_log_malformed",
"downgrade",
"Action log",
"action_log is not a list of structured entries "
f"(got {type(action_log).__name__}); it was ignored",
"pass action_log as a list of dict entries",
)
]
entries: list[dict] = []
findings: list[dict[str, str]] = []
for index, entry in enumerate(action_log):
if isinstance(entry, dict):
entries.append(entry)
continue
findings.append(
validator_finding(
"shared.action_log_malformed",
"downgrade",
"Action log",
f"action_log entry {index} is not a structured mapping "
f"(got {type(entry).__name__}); the entry was ignored",
"repair the malformed action_log entry or drop it before "
"revalidating",
)
)
return entries, findings
def validator_finding(
rule_id: str,
severity: str,
@@ -480,7 +421,7 @@ def _rule_shared_canonical_comment_post_claim(
rejected_in_report = bool(_CANONICAL_VALIDATION_REJECTED_RE.search(text))
rejected_in_log = False
if action_log:
for entry in _iter_action_entries(action_log):
for entry in action_log:
validation = entry.get("canonical_comment_validation") or {}
if validation.get("allowed") is False:
rejected_in_log = True
@@ -534,13 +475,9 @@ def _rule_reviewer_vague_mutations_none(
action_log: list[dict] | None = None,
mutations_observed: bool = False,
) -> list[dict[str, str]]:
# #698: infer review mutations only from authoritative evidence — an
# entry proves a mutation only when it affirmatively records
# performed=true and was not gated. Read-only diagnostics and pre-API
# rejections (entries without a performed flag) are not mutations.
performed = any(
e.get("performed") is True and not e.get("gated_rejected")
for e in _iter_action_entries(action_log)
e.get("performed") is not False and not e.get("gated_rejected")
for e in (action_log or [])
)
if not (mutations_observed or performed):
return []
@@ -599,7 +536,7 @@ def _rule_reviewer_git_fetch_readonly(
text = report_text or ""
fetch_observed = any(
_GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
for e in _iter_action_entries(action_log)
for e in (action_log or [])
) or _GIT_FETCH_RE.search(text)
if not fetch_observed:
return []
@@ -1040,7 +977,7 @@ def _rule_reviewer_target_branch_freshness(
fields = _handoff_fields(text)
fetch_reported = bool(_GIT_FETCH_RE.search(text)) or any(
_GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
for e in _iter_action_entries(action_log)
for e in (action_log or [])
)
target_sha_reported = bool(_TARGET_BRANCH_SHA_RE.search(text)) or any(
"target branch" in key and "sha" in key and _FULL_SHA_RE.search(value)
@@ -1798,12 +1735,6 @@ def assess_final_report_validator(
checks: dict[str, Any] = {}
findings: list[dict[str, str]] = []
# #698: malformed action_log data must never crash validation with a
# secondary exception; malformed entries surface as sanitized findings.
sanitized_action_log, action_log_findings = sanitize_action_log(action_log)
action_log = sanitized_action_log
findings.extend(action_log_findings)
if normalized_kind == "issue_filing" and issue_filing_lock is not None:
checks["issue_filing"] = assess_issue_filing_final_report(
report_text,
@@ -1832,23 +1763,7 @@ def assess_final_report_validator(
}
for rule in _RULES_BY_TASK.get(normalized_kind, ()):
try:
findings.extend(
_call_rule(rule, report_text, normalized_kind, rule_kwargs)
)
except Exception as exc: # #698: fail closed with a sanitized error
findings.append(
validator_finding(
"shared.validator_rule_error",
"block",
"Validator",
f"validator rule '{getattr(rule, '__name__', 'unknown')}' "
f"failed with {type(exc).__name__} (details withheld; "
"sanitized)",
"file a validator defect with the rule name; do not "
"bypass final-report validation",
)
)
findings.extend(_call_rule(rule, report_text, normalized_kind, rule_kwargs))
grade, blocked, downgraded = _aggregate_grade(findings)
reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings]
+5 -39
View File
@@ -3305,18 +3305,6 @@ def _save_review_decision_lock(data):
payload["profile_identity"] = binding["profile_identity"]
if binding.get("remote") and not payload.get("remote"):
payload["remote"] = binding["remote"]
# #695 AC6: stamp native transport provenance on durable decision locks.
try:
payload.update(
{
k: v
for k, v in mcp_daemon_guard.mutation_provenance_fields().items()
if v is not None
}
)
except Exception:
payload.setdefault("transport", "untrusted")
payload.setdefault("native_mcp_transport", False)
persisted = mcp_session_state.save_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
payload=payload,
@@ -4053,15 +4041,6 @@ def _evaluate_pr_review_submission(
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
return result
if live:
# #695 AC1/AC2: offline direct-import submit (PR #701 run_submit.py) fails closed.
try:
mcp_daemon_guard.assert_sanctioned_mutation_runtime(
"gitea_submit_pr_review"
)
mcp_daemon_guard.assert_no_direct_import_bypass("gitea_submit_pr_review")
except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
reasons.append(str(exc))
return result
ns_gate = _live_namespace_health_gate("review_pr")
if ns_gate:
reasons.extend(ns_gate)
@@ -4249,16 +4228,6 @@ def gitea_mark_final_review_decision(
repo: str | None = None,
) -> dict:
"""Mark validation complete; the final review decision is ready to submit."""
# #695 AC1/AC2: direct import / redirected session state cannot mark final.
try:
mcp_daemon_guard.assert_sanctioned_mutation_runtime(
"gitea_mark_final_review_decision"
)
mcp_daemon_guard.assert_no_direct_import_bypass(
"gitea_mark_final_review_decision"
)
except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
return {"marked_ready": False, "reasons": [str(exc)]}
action = (action or "").strip().lower()
lock = _load_review_decision_lock()
if lock is None:
@@ -12874,10 +12843,9 @@ def gitea_quarantine_contaminated_review(
427 until an independent adversarial reviewer and controller deployment
of this tooling have completed.
"""
# Fail closed outside production native MCP before any mutation assessment.
# Test-mode bootstrap must never reach this production mutation endpoint (#695).
# Fail closed outside native MCP before any mutation assessment.
try:
mcp_daemon_guard.assert_production_mutation_runtime(
mcp_daemon_guard.assert_sanctioned_mutation_runtime(
"gitea_quarantine_contaminated_review"
)
except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
@@ -13083,12 +13051,10 @@ def gitea_quarantine_contaminated_review(
# ── Entry point ───────────────────────────────────────────────────────────────
if __name__ == "__main__":
# #558 / #695: claim the resolved canonical entrypoint, then bind the live
# native MCP transport lifecycle before any tool dispatch. Env vars,
# basename-only stack frames, and import-only launch cannot reconstruct
# native transport; offline imports / standalone scripts fail closed.
# #558 / #695: mark this process as the official native MCP daemon before
# any tool dispatch. Env vars alone cannot reconstruct native transport;
# offline imports / standalone scripts fail closed on mutations.
mcp_daemon_guard.mark_sanctioned_daemon()
mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
# Lock this session's launch profile into the environment so child CLI
# processes (e.g. review_pr.py) can detect and refuse profile
# side-channel overrides (#199).
+34 -339
View File
@@ -6,18 +6,12 @@ Direct ``import gitea_mcp_server`` from a shell bypasses native MCP transport.
- Environment variables alone cannot reconstruct a native session
(``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are
insufficient for mutation gates).
- A process-local runtime record is established only by the resolved canonical
entrypoint path (not basename) **and** the actual native MCP transport
lifecycle (``bind_native_mcp_transport`` before ``mcp.run``). Merely
importing or launching the entrypoint offline does not grant mutation
authority.
- Public caller-controlled flags (including any former
``allow_test_bootstrap``) never establish trusted mutation provenance.
- A process-local runtime record is created only by the official entrypoint
(``mcp_server.py`` calling ``mark_sanctioned_daemon``), bound to PID and a
random secret that never leaves process memory.
- Offline scripts that import internals fail closed on mutations.
- Pytest remains allowed for hermetic unit tests via ``is_pytest_runtime()``.
A separate test-only seam may establish a **test-mode** native record for
unit tests of transport gates; that record cannot authorize production
Gitea mutation endpoints.
- Pytest remains allowed (hermetic tests); optional force flags exist for
provenance regression tests.
Manual deletion of session-state files is never a recovery path.
"""
@@ -29,7 +23,6 @@ import inspect
import os
import secrets
import time
from pathlib import Path
from typing import Any
SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON"
@@ -41,18 +34,6 @@ FORCE_PROVENANCE_FAIL_ENV = "GITEA_TEST_FORCE_UNSANCTIONED"
# Process-local native runtime (never persisted, never read from env alone).
_NATIVE_RUNTIME: dict[str, Any] | None = None
# Production transport identifiers accepted by bind_native_mcp_transport.
_PRODUCTION_TRANSPORTS = frozenset({"stdio"})
_RUNTIME_MODE_PRODUCTION = "production"
_RUNTIME_MODE_TEST = "test"
_PHASE_ENTRYPOINT_CLAIMED = "entrypoint_claimed"
_PHASE_TRANSPORT_BOUND = "transport_bound"
# Default session-state root (mirrors mcp_session_state; kept local to avoid
# import cycles). Used only to pin authority at transport bind (#695 AC2).
_DEFAULT_SESSION_STATE_DIR = os.path.expanduser("~/.cache/gitea-tools/session-state")
SESSION_STATE_DIR_ENV = "GITEA_MCP_SESSION_STATE_DIR"
class UnsanctionedRuntimeError(RuntimeError):
"""Raised when mutation/credential code runs outside a native MCP daemon."""
@@ -72,261 +53,52 @@ def is_pytest_runtime() -> bool:
return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip())
def _package_root() -> Path:
"""Directory that contains the canonical MCP entrypoint modules."""
return Path(__file__).resolve().parent
def canonical_entrypoint_paths() -> frozenset[str]:
"""Resolved absolute paths of official entrypoints (not basenames)."""
root = _package_root()
return frozenset(
{
str((root / "mcp_server.py").resolve()),
str((root / "gitea_mcp_server.py").resolve()),
}
)
def _resolve_path(path: str | None) -> str | None:
if not path:
return None
try:
return str(Path(path).resolve())
except (OSError, RuntimeError, ValueError):
return None
def _caller_official_entrypoint_path() -> str | None:
"""Return the resolved canonical entrypoint path in the call stack, or None.
Basename-only matches (e.g. an attacker file named ``mcp_server.py``
elsewhere) are rejected. The path must equal one of
:func:`canonical_entrypoint_paths`.
"""
canonical = canonical_entrypoint_paths()
for frame in inspect.stack()[1:20]:
resolved = _resolve_path(frame.filename)
if resolved and resolved in canonical:
return resolved
return None
def _caller_is_official_entrypoint() -> bool:
"""True when invoked from a resolved canonical entrypoint path (#695)."""
return _caller_official_entrypoint_path() is not None
"""True when mark_sanctioned_daemon is invoked from mcp_server.py."""
for frame in inspect.stack()[1:12]:
path = (frame.filename or "").replace("\\", "/")
base = path.rsplit("/", 1)[-1]
if base == "mcp_server.py":
return True
return False
def _new_runtime_token() -> tuple[str, str]:
token = secrets.token_hex(32)
fingerprint = hashlib.sha256(token.encode()).hexdigest()[:16]
return token, fingerprint
def mark_sanctioned_daemon(*, allow_test_bootstrap: bool = False) -> dict[str, Any]:
"""Mark this process as the official native MCP daemon (#695).
def mark_sanctioned_daemon() -> dict[str, Any]:
"""Claim the official entrypoint for this process (#695).
This alone does **not** authorize mutations. Callers must subsequently
bind the native MCP transport via :func:`bind_native_mcp_transport`.
Only a stack frame whose **resolved absolute path** is the canonical
``mcp_server.py`` or ``gitea_mcp_server.py`` next to this module may
claim the entrypoint. Basename spoofing is rejected.
There is no public ``allow_test_bootstrap`` argument: caller-controlled
flags must never establish trusted mutation provenance. Hermetic tests
use :func:`install_test_native_runtime` (pytest-only, test mode).
Only the official ``mcp_server.py`` entrypoint (or pytest test bootstrap)
may establish native transport. Setting env vars alone is insufficient.
"""
global _NATIVE_RUNTIME
if is_pytest_runtime():
# Under pytest, production mark is a no-op for transport authority.
# Tests that need a native-transport record use install_test_native_runtime.
return native_runtime_status()
entrypoint_path = _caller_official_entrypoint_path()
if entrypoint_path is None:
raise UnsanctionedRuntimeError(
"mark_sanctioned_daemon rejected: not called from the resolved "
"canonical MCP entrypoint path (#695). Basename-only names "
"(e.g. a renamed runner called mcp_server.py) are insufficient. "
"Offline import / standalone scripts cannot reconstruct native "
"transport. Stop after native MCP failure; do not run offline "
"mutation helpers."
)
token, fingerprint = _new_runtime_token()
if not is_pytest_runtime() and not allow_test_bootstrap:
if not _caller_is_official_entrypoint():
raise UnsanctionedRuntimeError(
"mark_sanctioned_daemon rejected: not called from official "
"mcp_server.py entrypoint (#695). Offline import / standalone "
"scripts cannot reconstruct native transport. Stop after native "
"MCP failure; do not run offline mutation helpers."
)
token = secrets.token_hex(32)
_NATIVE_RUNTIME = {
"token": token,
"token_fingerprint": fingerprint,
"token_fingerprint": hashlib.sha256(token.encode()).hexdigest()[:16],
"pid": os.getpid(),
"started_at": time.time(),
"entrypoint": "mcp_server",
"entrypoint_path": entrypoint_path,
"phase": _PHASE_ENTRYPOINT_CLAIMED,
"transport": None,
"mode": _RUNTIME_MODE_PRODUCTION,
}
# Legacy signal for older probes; alone does not authorize mutations.
os.environ[SANCTIONED_DAEMON_ENV] = "1"
return native_runtime_status()
def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
"""Bind the live native MCP transport lifecycle (#695).
Must be called from the resolved canonical entrypoint immediately before
the real MCP server transport loop (e.g. ``mcp.run(transport=\"stdio\")``).
Requires a prior successful :func:`mark_sanctioned_daemon` claim in this
process. Import-only or offline launch without this bind leaves
:func:`is_native_mcp_transport` false.
"""
global _NATIVE_RUNTIME
transport_name = (transport or "").strip().lower()
if transport_name not in _PRODUCTION_TRANSPORTS:
raise UnsanctionedRuntimeError(
f"bind_native_mcp_transport rejected: transport {transport!r} is "
f"not a production MCP transport (#695). Allowed: "
f"{sorted(_PRODUCTION_TRANSPORTS)}."
)
entrypoint_path = _caller_official_entrypoint_path()
if entrypoint_path is None:
raise UnsanctionedRuntimeError(
"bind_native_mcp_transport rejected: not called from the resolved "
"canonical MCP entrypoint path (#695)."
)
if _NATIVE_RUNTIME is None or int(_NATIVE_RUNTIME.get("pid") or -1) != os.getpid():
raise UnsanctionedRuntimeError(
"bind_native_mcp_transport rejected: no entrypoint claim in this "
"process (#695). Call mark_sanctioned_daemon() from the official "
"entrypoint first."
)
if _NATIVE_RUNTIME.get("mode") != _RUNTIME_MODE_PRODUCTION:
raise UnsanctionedRuntimeError(
"bind_native_mcp_transport rejected: runtime mode is not "
"production (#695)."
)
claimed = (_NATIVE_RUNTIME.get("entrypoint_path") or "").strip()
if claimed and claimed != entrypoint_path:
raise UnsanctionedRuntimeError(
"bind_native_mcp_transport rejected: entrypoint path mismatch "
"between mark and bind (#695)."
)
# Pin session-state root for this server lifetime (#695 AC2 / PR #701).
# Changing GITEA_MCP_SESSION_STATE_DIR after bind must not manufacture a
# second authority domain for decision locks / workflow proofs.
raw_state = (os.environ.get(SESSION_STATE_DIR_ENV) or "").strip()
if not raw_state:
raw_state = _DEFAULT_SESSION_STATE_DIR
try:
pinned_state = str(Path(raw_state).resolve())
except (OSError, RuntimeError, ValueError):
pinned_state = raw_state
_NATIVE_RUNTIME["phase"] = _PHASE_TRANSPORT_BOUND
_NATIVE_RUNTIME["transport"] = transport_name
_NATIVE_RUNTIME["entrypoint_path"] = entrypoint_path
_NATIVE_RUNTIME["bound_at"] = time.time()
_NATIVE_RUNTIME["session_state_dir"] = pinned_state
os.environ[SANCTIONED_DAEMON_ENV] = "1"
return native_runtime_status()
def install_test_native_runtime() -> dict[str, Any]:
"""Pytest-only seam for hermetic native-transport unit tests (#695).
Establishes a **test-mode** process-local record so unit tests can exercise
gates that require ``is_native_mcp_transport()``. This record:
- is rejected outside pytest (including a fresh offline interpreter);
- never uses production mode;
- cannot authorize production Gitea mutation endpoints
(:func:`assert_production_mutation_runtime` / production path of
:func:`assert_sanctioned_mutation_runtime` when not under pytest).
There is no public caller-controlled flag that forges production native
transport.
"""
global _NATIVE_RUNTIME
if not is_pytest_runtime():
raise UnsanctionedRuntimeError(
"install_test_native_runtime rejected: test-mode native runtime "
"is only available under pytest (#695). allow_test_bootstrap and "
"similar caller-controlled flags do not exist and cannot authorize "
"a fresh offline interpreter."
)
token, fingerprint = _new_runtime_token()
_NATIVE_RUNTIME = {
"token": token,
"token_fingerprint": fingerprint,
"pid": os.getpid(),
"started_at": time.time(),
"entrypoint": "test_bootstrap",
"entrypoint_path": None,
"phase": _PHASE_TRANSPORT_BOUND,
"transport": "test",
"mode": _RUNTIME_MODE_TEST,
"bound_at": time.time(),
}
return native_runtime_status()
def clear_native_runtime_for_tests() -> None:
"""Test helper: drop native runtime (does not clear env)."""
global _NATIVE_RUNTIME
_NATIVE_RUNTIME = None
def pinned_session_state_dir() -> str | None:
"""Session-state root pinned for this production transport lifetime (#695 AC2).
When production native transport is bound, durable session proofs must use
this directory only. Env overrides of ``GITEA_MCP_SESSION_STATE_DIR`` after
bind are ignored so redirected dirs (e.g. ``.mcp_session_701``) cannot
manufacture independent decision-lock authority (PR #701 recurrence).
"""
if not is_production_native_mcp_transport():
return None
pinned = (_NATIVE_RUNTIME or {}).get("session_state_dir")
text = (str(pinned) if pinned is not None else "").strip()
return text or None
def direct_import_env_enabled() -> bool:
"""True when the legacy direct-import opt-in env is set (never authorizes)."""
return (os.environ.get(ALLOW_DIRECT_IMPORT_ENV) or "").strip().lower() in {
"1",
"true",
"yes",
}
def assert_no_direct_import_bypass(context: str = "mutation") -> None:
"""Fail closed when GITEA_ALLOW_DIRECT_MCP_IMPORT is used for mutations (#695 AC1).
The env flag is never a sanctioned recovery path for LLM/agent sessions.
Under pytest hermetic tests this is a no-op so unit tests can set the flag
to prove it does not grant authority.
"""
if is_pytest_runtime():
return
if not direct_import_env_enabled():
return
raise UnsanctionedRuntimeError(
f"{ALLOW_DIRECT_IMPORT_ENV} does not authorize {context} (#695 AC1). "
"Direct import of gitea_mcp_server mutation tools is forbidden. "
"Stop after native MCP failure; reconnect the official MCP daemon. "
"Do not set direct-import flags, offline runners, or redirected "
f"{SESSION_STATE_DIR_ENV} directories to reconstruct gates."
)
def is_native_mcp_transport() -> bool:
"""True when this process holds a transport-bound native runtime (#695)."""
"""True when this process holds a live native MCP runtime record (#695)."""
if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
"1",
"true",
@@ -339,23 +111,12 @@ def is_native_mcp_transport() -> bool:
return False
if not (_NATIVE_RUNTIME.get("token") or "").strip():
return False
if _NATIVE_RUNTIME.get("phase") != _PHASE_TRANSPORT_BOUND:
return False
if not (_NATIVE_RUNTIME.get("transport") or "").strip():
return False
return True
def is_production_native_mcp_transport() -> bool:
"""True only for production-mode, transport-bound native runtime."""
if not is_native_mcp_transport():
return False
return (_NATIVE_RUNTIME or {}).get("mode") == _RUNTIME_MODE_PRODUCTION
def is_sanctioned_mcp_daemon() -> bool:
"""Backward-compatible name; #695 requires native transport, not env alone."""
if is_production_native_mcp_transport():
if is_native_mcp_transport():
return True
if is_pytest_runtime():
return True
@@ -365,45 +126,10 @@ def is_sanctioned_mcp_daemon() -> bool:
return False
def assert_production_mutation_runtime(context: str = "mutation") -> None:
"""Fail closed unless production native MCP transport is bound (#695).
Test-mode bootstrap records and pytest-only hermetic allowances do **not**
satisfy this gate. Use for production Gitea mutation endpoints that must
never be reachable via test bootstrap.
"""
if is_production_native_mcp_transport():
return
mode = (_NATIVE_RUNTIME or {}).get("mode")
if mode == _RUNTIME_MODE_TEST:
raise UnsanctionedRuntimeError(
f"Test-mode native runtime cannot authorize production {context} "
"(#695). install_test_native_runtime / former allow_test_bootstrap "
"must never reach real Gitea mutation endpoints."
)
assert_sanctioned_mutation_runtime(context)
def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
"""Fail closed when mutation code runs outside native MCP transport (#695).
Under pytest, hermetic unit tests are allowed (profile/permission tests).
Outside pytest, requires production-mode transport-bound native runtime.
Test-mode records do not authorize non-pytest production mutations.
``GITEA_ALLOW_DIRECT_MCP_IMPORT`` never authorizes mutations (#695 AC1).
"""
if is_pytest_runtime():
"""Fail closed when mutation code runs outside native MCP transport (#695)."""
if is_sanctioned_mcp_daemon():
return
# AC1: direct-import env is never a mutation recovery path (PR #701).
assert_no_direct_import_bypass(context)
if is_production_native_mcp_transport():
return
mode = (_NATIVE_RUNTIME or {}).get("mode")
if mode == _RUNTIME_MODE_TEST:
raise UnsanctionedRuntimeError(
f"Test-mode native runtime cannot authorize production {context} "
"(#695). Test bootstrap cannot reach production mutation endpoints."
)
env_spoof = (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {
"1",
"true",
@@ -413,26 +139,14 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
if env_spoof:
extra = (
f" Note: {SANCTIONED_DAEMON_ENV} alone is not sufficient (#695); "
"native transport requires the official MCP entrypoint and a live "
"transport bind."
)
phase = (_NATIVE_RUNTIME or {}).get("phase")
if phase == _PHASE_ENTRYPOINT_CLAIMED:
extra = (
(extra + " ") if extra else " "
) + (
"Entrypoint was claimed but native MCP transport was never bound "
"(#695); offline launch/import of the real entrypoint does not "
"grant mutation authority."
"native transport requires the official MCP entrypoint."
)
raise UnsanctionedRuntimeError(
f"Unsanctioned / non-native runtime blocked {context} (#695). "
"Do not import gitea_mcp_server or call mutation helpers from a raw "
"shell, offline runner, or ad-hoc script after native MCP failure. "
"Stop and reconnect the official MCP daemon (mcp_server.py) over "
"native transport. "
f"Do not set {ALLOW_DIRECT_IMPORT_ENV}, override "
f"{SESSION_STATE_DIR_ENV}, or use raw token env vars in LLM "
"Stop and reconnect the official MCP daemon (mcp_server.py). "
f"Do not set {ALLOW_DIRECT_IMPORT_ENV} or raw token env vars in LLM "
f"sessions.{extra}"
)
@@ -467,23 +181,14 @@ def native_runtime_status() -> dict[str, Any]:
rt = _NATIVE_RUNTIME or {}
return {
"native_mcp_transport": is_native_mcp_transport(),
"production_native_mcp_transport": is_production_native_mcp_transport(),
"pytest": is_pytest_runtime(),
"pid": rt.get("pid"),
"token_fingerprint": rt.get("token_fingerprint"),
"started_at": rt.get("started_at"),
"entrypoint": rt.get("entrypoint"),
"entrypoint_path": rt.get("entrypoint_path"),
"phase": rt.get("phase"),
"transport": rt.get("transport"),
"mode": rt.get("mode"),
"session_state_dir": pinned_session_state_dir() or rt.get("session_state_dir"),
"session_state_dir_pinned": pinned_session_state_dir() is not None,
"direct_import_env_set": direct_import_env_enabled(),
"env_sanctioned_alone_insufficient": True,
"sanctioned_env": SANCTIONED_DAEMON_ENV,
"allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV,
"session_state_dir_env": SESSION_STATE_DIR_ENV,
"allow_keychain_cli_env": ALLOW_KEYCHAIN_CLI_ENV,
}
@@ -498,20 +203,10 @@ def runtime_status() -> dict[str, Any]:
def mutation_provenance_fields() -> dict[str, Any]:
"""Fields to attach to live mutation / review audit records (#695 AC6)."""
st = native_runtime_status()
transport = "native_mcp" if st["native_mcp_transport"] else "untrusted"
if st.get("mode") == _RUNTIME_MODE_TEST and st["native_mcp_transport"]:
transport = "test_native_mcp"
return {
"transport": transport,
"transport": "native_mcp" if st["native_mcp_transport"] else "untrusted",
"native_mcp_transport": bool(st["native_mcp_transport"]),
"production_native_mcp_transport": bool(
st.get("production_native_mcp_transport")
),
"native_runtime_pid": st.get("pid"),
"native_token_fingerprint": st.get("token_fingerprint"),
"entrypoint": st.get("entrypoint"),
"phase": st.get("phase"),
"mode": st.get("mode"),
"session_state_dir": st.get("session_state_dir"),
"session_state_dir_pinned": bool(st.get("session_state_dir_pinned")),
}
+3 -5
View File
@@ -41,17 +41,15 @@ def check_conflict_markers():
check_conflict_markers()
# #558 / #695: claim the official entrypoint before loading mutation modules.
# This alone does NOT authorize mutations — gitea_mcp_server binds the live
# native MCP transport (stdio) immediately before mcp.run. Import-only or
# offline launch without that bind fails closed on mutations.
# #558: official entrypoint marks the process as the sanctioned MCP daemon
# before loading mutation modules (blocks raw shell import bypasses).
try:
import mcp_daemon_guard
mcp_daemon_guard.mark_sanctioned_daemon()
except Exception:
# Guard import failures must not hide conflict-marker infra_stop above;
# gitea_mcp_server main also marks + binds when run as __main__.
# gitea_mcp_server main also marks sanctioned when run as __main__.
pass
# Execute the actual server logic via exec in this namespace.
-50
View File
@@ -44,34 +44,6 @@ SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
def default_state_dir() -> str:
"""Resolve the durable session-state root.
When production native MCP transport is bound (#695 AC2), the directory
pinned at transport bind is authoritative: later overrides of
``GITEA_MCP_SESSION_STATE_DIR`` cannot manufacture a second authority
domain (PR #701 recurrence: ``.mcp_session_701`` evasion of cross-PR
decision locks).
"""
try:
import mcp_daemon_guard
pinned = mcp_daemon_guard.pinned_session_state_dir()
if pinned:
return pinned
except Exception:
# Fail open to env/default only when guard is unavailable (e.g. partial
# import during bootstrap). Mutation gates still fail closed separately.
pass
raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip()
return raw or DEFAULT_STATE_DIR
def env_session_state_dir_unpinned() -> str:
"""Raw env/default session-state dir ignoring production transport pin.
Intended for diagnostics and tests that assert pin behavior — not for
mutation-sensitive durable proofs under a bound native daemon.
"""
raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip()
return raw or DEFAULT_STATE_DIR
@@ -390,26 +362,6 @@ def save_state(
body["org"] = key_org
if key_repo is not None:
body["repo"] = key_repo
# Stamp session-state authority used for this write (#695 AC2 / AC6).
body.setdefault("session_state_dir", root)
try:
import mcp_daemon_guard
prov = mcp_daemon_guard.mutation_provenance_fields()
body.setdefault(
"native_token_fingerprint", prov.get("native_token_fingerprint")
)
body.setdefault(
"native_mcp_transport", bool(prov.get("native_mcp_transport"))
)
body.setdefault(
"production_native_mcp_transport",
bool(prov.get("production_native_mcp_transport")),
)
body.setdefault("transport", prov.get("transport"))
except Exception:
body.setdefault("native_mcp_transport", False)
body.setdefault("transport", "untrusted")
envelope = {
"kind": kind,
@@ -421,8 +373,6 @@ def save_state(
"recorded_at": body["recorded_at"],
"updated_at": body["updated_at"],
"writer_pid": body["writer_pid"],
"session_state_dir": body.get("session_state_dir"),
"transport": body.get("transport"),
"payload": body,
}
_write_json(path, envelope)
+8 -27
View File
@@ -86,22 +86,6 @@ _WRONG_BRANCH_RE = re.compile(
r"deleted branch (?:does not match|!=|differs from) (?:merged )?pr head",
re.IGNORECASE,
)
# #698: canonical reviewer lease lifecycle operations are NOT post-merge
# cleanup. Releasing a reviewer PR lease (or posting its terminal
# phase=released marker) happens after every review — merged or not — and
# must never trigger the post-merge branch/worktree cleanup checklist.
_LEASE_LIFECYCLE_RE = re.compile(
r"(?:gitea_release_reviewer_pr_lease|gitea_abandon_workflow_lease|"
r"release[d]? (?:the )?(?:reviewer|workflow) (?:pr )?lease|"
r"reviewer (?:pr )?lease release[d]?|"
r"lease (?:marker|comment).{0,40}phase\s*[:=]\s*released|"
r"phase\s*[:=]\s*released)",
re.IGNORECASE,
)
_CLEANUP_MUTATIONS_VALUE_RE = re.compile(
r"cleanup mutations\s*:\s*([^\n]+)",
re.IGNORECASE,
)
def _claims_remote_delete(text: str) -> bool:
@@ -220,19 +204,16 @@ def assess_post_merge_cleanup_proof(
for field in _worktree_cleanup_fields_present(text)
)
if (remote_delete or worktree_remove) and not (remote_delete or worktree_remove):
pass
if not remote_delete and not worktree_remove:
value_match = _CLEANUP_MUTATIONS_VALUE_RE.search(text)
value = (value_match.group(1).strip() if value_match else "")
value_lower = value.lower()
substantive = bool(value) and value_lower not in {
"none", "n/a", "not applicable",
}
# #698: reviewer lease release / terminal lease markers are lease
# lifecycle, not post-merge cleanup — no checklist owed.
lease_lifecycle_only = substantive and bool(
_LEASE_LIFECYCLE_RE.search(value)
cleanup_mutations = re.search(
r"cleanup mutations\s*:\s*(?!none\b)\S",
text,
re.IGNORECASE,
)
if substantive and not lease_lifecycle_only:
if cleanup_mutations:
reasons.append(
"cleanup mutations reported without post-merge cleanup proof checklist"
)
+6 -67
View File
@@ -403,37 +403,10 @@ _REVIEWER_ACTIVE_RE = re.compile(
r"whether any reviewer was active\s*:\s*(yes|no|true|false)",
re.IGNORECASE,
)
# #698 phase detection for phase-specific head proofs.
_NO_REVIEWED_HEAD_RE = re.compile(
r"(?:reviewed head sha|candidate head sha)\s*:\s*none\b",
re.IGNORECASE,
)
_VERDICT_RECORDED_RE = re.compile(
r"review decision\s*:\s*(?:approve[d]?|request[_ ]changes)\b"
r"|review_status\s*:\s*(?:approved|request_changes)\b"
r"|terminal review mutation\s*:\s*(?!none\b)\S",
re.IGNORECASE,
)
_MERGE_ATTEMPTED_RE = re.compile(
r"merge result\s*:\s*(?:merged|success|performed|failed|attempted)\b"
r"|merge mutations\s*:\s*(?!none\b|not applicable\b)\S",
re.IGNORECASE,
)
_VALIDATION_STARTED_RE = re.compile(
r"validation\s*:\s*(?!none\b|not run\b|not applicable\b|not started\b)"
r"[^\n]*(?:pass|fail|ran|executed|\d+\s+passed)",
re.IGNORECASE,
)
def assess_reviewer_stale_head_final_report(report_text: str) -> dict[str, Any]:
"""Final-report proof for reviewed vs live head SHAs (#399 AC 6).
#698: head proofs are phase-specific. A legitimately blocked run that
never began validation (no reviewed head, no formal verdict, no merge)
owes none of them; approval-time and merge-time live-head proofs are
owed only once the corresponding phase actually begins.
"""
"""Final-report proof for reviewed vs live head SHAs (#399 AC 6)."""
text = report_text or ""
reasons: list[str] = []
reviewed = _normalize_sha(_REVIEWED_HEAD_RE.search(text).group(1) if _REVIEWED_HEAD_RE.search(text) else None)
@@ -449,49 +422,15 @@ def assess_reviewer_stale_head_final_report(report_text: str) -> dict[str, Any]:
)
push_during = _PUSH_DURING_VALIDATION_RE.search(text)
# Phase detection from the report's own claims.
no_head_stated = bool(_NO_REVIEWED_HEAD_RE.search(text))
verdict_recorded = bool(_VERDICT_RECORDED_RE.search(text))
merge_attempted = bool(_MERGE_ATTEMPTED_RE.search(text))
validation_started = bool(reviewed) or bool(_VALIDATION_STARTED_RE.search(text))
blocked_before_validation = (
no_head_stated
and not reviewed
and not verdict_recorded
and not merge_attempted
and not validation_started
)
if blocked_before_validation:
return {
"proven": True,
"block": False,
"reasons": [],
"reviewed_head_sha": None,
"live_head_sha_before_approval": None,
"live_head_sha_before_merge": None,
"push_during_validation": (
push_during.group(1).lower() if push_during else None
),
"phase": "blocked_before_validation",
}
if not reviewed and not no_head_stated:
# The head must always be STATED — either a SHA or an explicit
# 'none'. Silence is not a phase claim and fails closed.
reasons.append(
"reviewed head SHA not stated in final report "
"(state the SHA or an explicit 'none')"
)
elif not reviewed and (validation_started or verdict_recorded or merge_attempted):
if not reviewed:
reasons.append("reviewed head SHA not stated in final report")
if verdict_recorded and not live_approval:
if not live_approval:
reasons.append("final live head SHA before approval not stated")
if merge_attempted and not live_merge:
if not live_merge:
reasons.append("final live head SHA before merge not stated")
if validation_started and not push_during:
if not push_during:
reasons.append("whether push occurred during validation not stated")
if reviewed and live_approval and reviewed != live_approval:
elif reviewed and live_approval and reviewed != live_approval:
reasons.append("live head before approval differs from reviewed head SHA")
elif reviewed and live_merge and reviewed != live_merge:
reasons.append("live head before merge differs from reviewed head SHA")
+1 -6
View File
@@ -25,13 +25,8 @@ _REVIEWED_HEAD_RE = re.compile(
r"(?:pinned reviewed head|reviewed head sha)\s*:\s*([0-9a-f]{7,40})",
re.IGNORECASE,
)
# #698: validation pass proof appears in several legitimate shapes —
# "Validation: pass", "Validation: focused 50 passed; full 2665 passed",
# or structured "validation_status: pass". Accept pass evidence anywhere in
# the Validation field's value, not only as its first token.
_VALIDATION_PASS_RE = re.compile(
r"validation(?:_status)?\s*:[^\n]{0,300}?"
r"(?:\bpass(?:ed)?\b|\bstrong\b|\bok\b|\bgreen\b|\d+\s+passed)",
r"validation\s*:\s*(?:pass|passed|strong|ok|green)",
re.IGNORECASE,
)
_MERGED_CLAIM_RE = re.compile(
+9 -36
View File
@@ -858,16 +858,9 @@ _WALKTHROUGH_ARTIFACT_RE = re.compile(r"walkthrough\.md", re.I)
def _performed_file_mutations(action_log: list[dict] | None) -> list[dict]:
"""Return performed local file mutations, excluding gated rejections.
Non-dict entries (malformed JSON, LLM mistakes) are ignored instead of
raising ``AttributeError`` (#698): a malformed ledger entry can never be
authoritative mutation evidence.
"""
"""Return performed local file mutations, excluding gated rejections."""
performed: list[dict] = []
for entry in action_log or []:
if not isinstance(entry, dict):
continue
if entry.get("gated_rejected") or entry.get("performed") is False:
continue
action = (entry.get("action") or "").strip().lower()
@@ -2235,31 +2228,24 @@ HANDOFF_REVIEW_MUTATION_FIELDS = (
)
HANDOFF_ROLE_FIELDS = {
# #698: the review/merger required-field sets must stay aligned with the
# canonical schema (skills/llm-project-workflow/schemas/
# review-merge-final-report.md). The schema explicitly FORBIDS the legacy
# fields 'Pinned reviewed head', 'Scratch worktree used', and 'Workspace
# mutations' — a validator must never demand a field the schema bans.
"review": (
("Selected PR", ("selected pr",)),
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
("Reviewed head SHA", ("reviewed head sha", "candidate head sha")),
("Review worktree path", ("review worktree path", "worktree path",
"starting worktree path")),
("Review worktree dirty", ("review worktree dirty", "worktree dirty",
"whether worktree was dirty")),
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
("Worktree path", ("worktree path", "starting worktree path")),
("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
("Scratch worktree used", ("scratch worktree used", "scratch clone used",
"scratch worktree")),
("Unrelated local mutations", ("unrelated local mutations",
"unrelated files modified",
"file edits by reviewer")),
"unrelated files modified")),
("Review decision", ("review decision", "decision")),
("Merge result", ("merge result",)),
("Linked issue status", ("linked issue status", "linked issue")),
("Cleanup status", ("cleanup status", "cleanup")),
("Safe next action", ("safe next action", "next")),
) + HANDOFF_REVIEW_MUTATION_FIELDS,
"merger": (
("Selected PR", ("selected pr",)),
("Reviewed head SHA", ("reviewed head sha", "candidate head sha")),
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
("Active profile", ("active profile",)),
("Role kind", ("role kind",)),
("Merge capability source", ("merge capability source",)),
@@ -2447,22 +2433,9 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
# Issue #320: reviewer and merger handoffs use the precise mutation categories
# in HANDOFF_REVIEW_MUTATION_FIELDS instead of the legacy ambiguous
# "Workspace mutations" field, which is rejected below.
# Issue #698: the canonical review-merge schema has no 'Mutations',
# 'Next', 'Issue/PR', 'Branch/SHA', or 'Files changed' fields — their
# content lives in the precise mutation categories, 'Safe next
# action', 'Selected PR'/'Linked issue', head-SHA fields, and 'Files
# reviewed'. Requiring the legacy names rejects canonical reports.
_non_canonical_for_review = {
"Workspace mutations",
"Mutations",
"Next",
"Issue/PR",
"Branch/SHA",
"Files changed",
}
required = [
field for field in required
if field[0] not in _non_canonical_for_review
if field[0] != "Workspace mutations"
]
if any(label.startswith("workspace mutations") for label in labels):
return {
-11
View File
@@ -58,17 +58,6 @@ def _reset_mutation_authority(monkeypatch):
_fallback: str = state_dir,
_env_key: str = mcp_session_state.STATE_DIR_ENV,
) -> str:
# #695 AC2: when production native transport has pinned a session
# state root, that pin is authoritative even under test isolation
# (PR #701 redirected-state regression).
try:
import mcp_daemon_guard
pinned = mcp_daemon_guard.pinned_session_state_dir()
if pinned:
return pinned
except Exception:
pass
raw = (os.environ.get(_env_key) or "").strip()
return raw or _fallback
@@ -1,18 +1,14 @@
"""Regression tests for Issue #695 — second incident (PR #694 / review 427).
Reproduces offline import, env-only runtime spoof, exposed-token invocation,
direct imports, basename entrypoint spoof, allow_test_bootstrap forgery,
standalone quarantine attempts, and false official workflow canonical claims.
Gates must fail closed.
direct imports, locally generated runtime keys, standalone quarantine attempts,
and false official workflow canonical claims. Gates must fail closed.
"""
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import textwrap
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -24,26 +20,6 @@ import canonical_comment_validator as ccv
HEAD_694 = "1844e298809373be19a526fd39b7d8b0669eb5bd"
HEAD_OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
REPO_ROOT = Path(__file__).resolve().parent.parent
def _run_offline_snippet(snippet: str, *, env_extra: dict[str, str] | None = None) -> subprocess.CompletedProcess:
"""Execute snippet in a fresh interpreter (no pytest modules)."""
env = os.environ.copy()
env.pop("PYTEST_CURRENT_TEST", None)
env.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "")
if env_extra:
env.update(env_extra)
return subprocess.run(
[sys.executable, "-c", snippet],
capture_output=True,
text=True,
cwd=str(REPO_ROOT),
env=env,
timeout=30,
check=False,
)
class TestNativeTransportBinding(unittest.TestCase):
@@ -63,53 +39,35 @@ class TestNativeTransportBinding(unittest.TestCase):
mcp_daemon_guard.assert_sanctioned_mutation_runtime("offline_import")
msg = str(ctx.exception)
self.assertIn("#695", msg)
# FORCE disables pytest allowance; direct-import env is rejected first
# (AC1). Without ALLOW_DIRECT, env-alone also yields "not sufficient".
lowered = msg.lower()
self.assertTrue(
"direct" in lowered
or "not sufficient" in lowered
or "allow_direct" in lowered
or "gitea_allow_direct" in lowered,
msg,
)
self.assertIn("not sufficient", msg.lower() + " " + msg)
def test_direct_import_mark_rejected_outside_entrypoint(self):
mcp_daemon_guard.clear_native_runtime_for_tests()
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
mcp_daemon_guard.mark_sanctioned_daemon()
self.assertIn("canonical", str(ctx.exception).lower())
self.assertIn("mcp_server.py", str(ctx.exception))
self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
def test_locally_generated_runtime_key_without_entrypoint_rejected(self):
"""Spoofing process-local fields via mark outside entrypoint fails."""
mcp_daemon_guard.clear_native_runtime_for_tests()
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
# Even if a caller tries allow_test_bootstrap under force-unsanctioned
# pytest path is also forced off — only real entrypoint may mark.
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
mcp_daemon_guard.mark_sanctioned_daemon()
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=False)
def test_test_native_runtime_for_hermetic_tests_not_production(self):
def test_test_bootstrap_establishes_native_for_hermetic_tests(self):
mcp_daemon_guard.clear_native_runtime_for_tests()
st = mcp_daemon_guard.install_test_native_runtime()
st = mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
self.assertTrue(st["native_mcp_transport"])
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport())
mcp_daemon_guard.assert_sanctioned_mutation_runtime("test-bootstrap")
fields = mcp_daemon_guard.mutation_provenance_fields()
self.assertEqual(fields["transport"], "test_native_mcp")
self.assertEqual(fields["transport"], "native_mcp")
self.assertTrue(fields["native_mcp_transport"])
self.assertFalse(fields["production_native_mcp_transport"])
self.assertIsNotNone(fields["native_token_fingerprint"])
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
mcp_daemon_guard.assert_production_mutation_runtime("prod-endpoint")
self.assertIn("Test-mode", str(ctx.exception))
def test_no_allow_test_bootstrap_parameter_on_mark(self):
import inspect
sig = inspect.signature(mcp_daemon_guard.mark_sanctioned_daemon)
self.assertNotIn("allow_test_bootstrap", sig.parameters)
def test_exposed_token_env_never_grants_native(self):
"""Raw / exposed token env vars must never reconstruct native transport."""
@@ -138,249 +96,6 @@ class TestNativeTransportBinding(unittest.TestCase):
os.environ.pop(key, None)
class TestAC9BypassRegressions(unittest.TestCase):
"""AC9: empirically reproduced offline bypasses must fail closed (#695)."""
def tearDown(self) -> None:
mcp_daemon_guard.clear_native_runtime_for_tests()
os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
def test_allow_test_bootstrap_cannot_authorize_fresh_offline_interpreter(self):
"""Finding 1: former allow_test_bootstrap forge is gone and rejected offline."""
snippet = textwrap.dedent(
"""
import inspect
import mcp_daemon_guard as g
sig = inspect.signature(g.mark_sanctioned_daemon)
assert "allow_test_bootstrap" not in sig.parameters, "bootstrap flag must not exist"
try:
g.mark_sanctioned_daemon(allow_test_bootstrap=True)
except TypeError:
pass
else:
raise SystemExit("mark_sanctioned_daemon accepted allow_test_bootstrap")
try:
g.install_test_native_runtime()
except g.UnsanctionedRuntimeError as exc:
assert "pytest" in str(exc).lower() or "#695" in str(exc)
else:
raise SystemExit("install_test_native_runtime authorized offline interpreter")
assert g.is_native_mcp_transport() is False
try:
g.assert_sanctioned_mutation_runtime("offline-bootstrap")
except g.UnsanctionedRuntimeError:
pass
else:
raise SystemExit("mutation runtime authorized after offline bootstrap attempt")
print("OK")
"""
)
proc = _run_offline_snippet(snippet)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("OK", proc.stdout)
def test_renamed_runner_named_mcp_server_py_rejected(self):
"""Finding 2: basename-only entrypoint trust is insufficient."""
with tempfile.TemporaryDirectory() as tmp:
attacker = Path(tmp) / "mcp_server.py"
attacker.write_text(
textwrap.dedent(
"""
import mcp_daemon_guard as g
try:
g.mark_sanctioned_daemon()
except g.UnsanctionedRuntimeError as exc:
print("REJECTED:" + str(exc))
raise SystemExit(0)
print("AUTHORIZED")
raise SystemExit(1)
"""
),
encoding="utf-8",
)
env = os.environ.copy()
env.pop("PYTEST_CURRENT_TEST", None)
env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "")
proc = subprocess.run(
[sys.executable, str(attacker)],
capture_output=True,
text=True,
cwd=tmp,
env=env,
timeout=30,
check=False,
)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("REJECTED:", proc.stdout)
self.assertIn("canonical", proc.stdout.lower())
self.assertNotIn("AUTHORIZED", proc.stdout)
def test_direct_launch_import_canonical_entrypoint_without_transport_rejected(self):
"""Merely importing/launching real entrypoint offline must not authorize."""
snippet = textwrap.dedent(
f"""
import importlib.util
import mcp_daemon_guard as g
# Simulate claim-only phase (no transport bind).
g.clear_native_runtime_for_tests()
# Direct mark from non-entrypoint must fail.
try:
g.mark_sanctioned_daemon()
except g.UnsanctionedRuntimeError:
pass
assert g.is_native_mcp_transport() is False
# Even if someone forges entrypoint_claimed without transport bind:
g._NATIVE_RUNTIME = {{
"token": "x" * 64,
"token_fingerprint": "deadbeefdeadbeef",
"pid": __import__("os").getpid(),
"started_at": 0,
"entrypoint": "mcp_server",
"entrypoint_path": {str(REPO_ROOT / "mcp_server.py")!r},
"phase": "entrypoint_claimed",
"transport": None,
"mode": "production",
}}
assert g.is_native_mcp_transport() is False
try:
g.assert_sanctioned_mutation_runtime("import-only")
except g.UnsanctionedRuntimeError as exc:
assert "transport" in str(exc).lower() or "entrypoint" in str(exc).lower() or "#695" in str(exc)
else:
raise SystemExit("import-only claim authorized mutation")
print("OK")
"""
)
proc = _run_offline_snippet(snippet)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("OK", proc.stdout)
def test_spoofed_pytest_env_stack_path_rejected(self):
"""Spoofed pytest/env/call-stack/path evidence must not authorize offline."""
snippet = textwrap.dedent(
f"""
import os
import mcp_daemon_guard as g
os.environ["PYTEST_CURRENT_TEST"] = "spoofed::test"
os.environ[g.SANCTIONED_DAEMON_ENV] = "1"
os.environ[g.ALLOW_DIRECT_IMPORT_ENV] = "1"
# Fresh interpreter has no pytest module; PYTEST_CURRENT_TEST alone
# might still trip is_pytest_runtime — force-unsanctioned is not set.
# But install_test_native_runtime requires real pytest path; if
# PYTEST_CURRENT_TEST alone grants is_pytest_runtime, production
# mutation still requires production transport outside true pytest.
if g.is_pytest_runtime():
# Env-only pytest spoof: test install may succeed, but production
# mutation gate must still reject test mode.
g.install_test_native_runtime()
assert g.is_production_native_mcp_transport() is False
try:
g.assert_production_mutation_runtime("spoofed-pytest")
except g.UnsanctionedRuntimeError:
pass
else:
raise SystemExit("production mutation accepted test-mode under spoofed pytest env")
else:
try:
g.install_test_native_runtime()
except g.UnsanctionedRuntimeError:
pass
else:
raise SystemExit("test install without pytest evidence")
# Basename path spoof via inspect is covered elsewhere; env alone:
g.clear_native_runtime_for_tests()
os.environ.pop("PYTEST_CURRENT_TEST", None)
assert g.is_native_mcp_transport() is False
try:
g.assert_sanctioned_mutation_runtime("env-path-spoof")
except g.UnsanctionedRuntimeError:
pass
else:
raise SystemExit("env/path spoof authorized mutation")
print("OK")
"""
)
proc = _run_offline_snippet(snippet)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("OK", proc.stdout)
def test_test_bootstrap_cannot_reach_production_mutation_endpoints(self):
"""Under pytest, test-mode runtime cannot satisfy production mutation gate."""
mcp_daemon_guard.clear_native_runtime_for_tests()
mcp_daemon_guard.install_test_native_runtime()
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
mcp_daemon_guard.assert_production_mutation_runtime(
"gitea_quarantine_contaminated_review"
)
msg = str(ctx.exception)
self.assertIn("Test-mode", msg)
self.assertIn("#695", msg)
# Offline: install_test_native_runtime must not authorize production mutations.
snippet = textwrap.dedent(
"""
import mcp_daemon_guard as g
try:
g.install_test_native_runtime()
except g.UnsanctionedRuntimeError:
pass
assert g.is_production_native_mcp_transport() is False
try:
g.assert_production_mutation_runtime("gitea_quarantine_contaminated_review")
except g.UnsanctionedRuntimeError:
pass
else:
raise SystemExit("production mutation authorized offline")
try:
g.assert_sanctioned_mutation_runtime("gitea_mutation")
except g.UnsanctionedRuntimeError:
pass
else:
raise SystemExit("sanctioned mutation authorized offline")
print("OK")
"""
)
proc = _run_offline_snippet(snippet)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("OK", proc.stdout)
def test_legitimate_native_transport_bind_succeeds(self):
"""Canonical entrypoint path + stdio bind establishes production native."""
mcp_daemon_guard.clear_native_runtime_for_tests()
# Simulate production path under force-unsanctioned (no pytest allowance)
# by calling internal claim/bind with patched caller path.
canonical = str((REPO_ROOT / "mcp_server.py").resolve())
def _fake_caller():
return canonical
with patch.object(
mcp_daemon_guard, "_caller_official_entrypoint_path", side_effect=_fake_caller
):
# Force non-pytest path for mark/bind logic.
with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
st1 = mcp_daemon_guard.mark_sanctioned_daemon()
self.assertFalse(st1["native_mcp_transport"])
self.assertEqual(st1["phase"], "entrypoint_claimed")
st2 = mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
self.assertTrue(st2["native_mcp_transport"])
self.assertTrue(st2["production_native_mcp_transport"])
self.assertEqual(st2["transport"], "stdio")
self.assertEqual(st2["mode"], "production")
mcp_daemon_guard.assert_sanctioned_mutation_runtime("native-ide")
mcp_daemon_guard.assert_production_mutation_runtime("native-ide")
fields = mcp_daemon_guard.mutation_provenance_fields()
self.assertEqual(fields["transport"], "native_mcp")
self.assertTrue(fields["production_native_mcp_transport"])
def test_canonical_entrypoint_paths_are_resolved_absolute(self):
paths = mcp_daemon_guard.canonical_entrypoint_paths()
self.assertTrue(any(p.endswith("mcp_server.py") for p in paths))
for p in paths:
self.assertTrue(os.path.isabs(p), p)
self.assertEqual(p, str(Path(p).resolve()))
class TestQuarantineWriteNativeOnly(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
@@ -418,12 +133,7 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase):
forensic_comment_ids=[10883, 10886],
)
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
# Force non-pytest and non-native for write path.
with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
with patch.object(
mcp_daemon_guard, "is_native_mcp_transport", return_value=False
):
review_quarantine.write_quarantine_record(record)
review_quarantine.write_quarantine_record(record)
def test_confirmation_must_match_exactly(self):
assessment = review_quarantine.assess_quarantine_write(
@@ -460,7 +170,7 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase):
"review_quarantine.mcp_session_state.default_state_dir",
return_value=self._tmp.name,
):
mcp_daemon_guard.install_test_native_runtime()
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
record = review_quarantine.build_quarantine_record(
remote="prgs",
org="org",
@@ -595,7 +305,7 @@ class TestFeedbackQuarantineIntegration(unittest.TestCase):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
mcp_daemon_guard.clear_native_runtime_for_tests()
mcp_daemon_guard.install_test_native_runtime()
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
def tearDown(self) -> None:
mcp_daemon_guard.clear_native_runtime_for_tests()
@@ -686,163 +396,5 @@ class TestDocsStopAfterNativeFailure(unittest.TestCase):
self.assertIn("GITEA_ALLOW_DIRECT_MCP_IMPORT", doc)
class TestPR701DirectImportSessionOverrideSequence(unittest.TestCase):
"""AC9 regression: exact PR #701 contaminated-approval sequence must fail closed.
Observed attack:
- GITEA_ALLOW_DIRECT_MCP_IMPORT=1
- import mutation tools from gitea_mcp_server
- GITEA_MCP_SESSION_STATE_DIR=.mcp_session_701 (cross-PR lock evasion)
- mark_final + submit_pr_review
"""
def tearDown(self) -> None:
mcp_daemon_guard.clear_native_runtime_for_tests()
os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
os.environ.pop(mcp_daemon_guard.SANCTIONED_DAEMON_ENV, None)
os.environ.pop(mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV, None)
os.environ.pop(mcp_daemon_guard.SESSION_STATE_DIR_ENV, None)
def test_offline_run_submit_sequence_fails_closed(self):
"""Fresh interpreter: direct import + state-dir override cannot mark/submit."""
with tempfile.TemporaryDirectory() as tmp:
redirect = str(Path(tmp) / ".mcp_session_701")
snippet = textwrap.dedent(
f"""
import os
import sys
os.environ["GITEA_ALLOW_DIRECT_MCP_IMPORT"] = "1"
os.environ["GITEA_MCP_SESSION_STATE_DIR"] = {redirect!r}
os.environ["GITEA_MCP_PROFILE"] = "prgs-reviewer"
# No pytest modules in this subprocess.
import mcp_daemon_guard as g
assert g.is_native_mcp_transport() is False
assert g.is_production_native_mcp_transport() is False
try:
g.assert_sanctioned_mutation_runtime("run_submit_mark")
except g.UnsanctionedRuntimeError as exc:
msg = str(exc)
assert "GITEA_ALLOW_DIRECT_MCP_IMPORT" in msg or "#695" in msg
else:
raise SystemExit("direct-import env authorized mutation runtime")
try:
g.mark_sanctioned_daemon()
except g.UnsanctionedRuntimeError:
pass
else:
raise SystemExit("mark_sanctioned_daemon authorized offline import")
# Simulate decision-lock write into redirected dir only — must not
# establish native authority.
import mcp_session_state as ss
wrote = ss.save_state(
kind=ss.KIND_DECISION_LOCK,
payload={{
"final_review_decision_ready": True,
"ready_pr_number": 701,
"ready_action": "approve",
"ready_expected_head_sha": "6b675f5c834b41f9d74e8a54294ff44dddf28ae4",
"session_profile": "prgs-reviewer",
"session_profile_lock": "prgs-reviewer",
"remote": "prgs",
}},
profile_identity="prgs-reviewer",
state_dir={redirect!r},
)
assert wrote is not None
assert g.is_native_mcp_transport() is False
try:
g.assert_no_direct_import_bypass("gitea_submit_pr_review")
except g.UnsanctionedRuntimeError:
pass
else:
raise SystemExit("direct-import bypass accepted for submit")
print("OK")
"""
)
proc = _run_offline_snippet(snippet)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("OK", proc.stdout)
def test_session_state_dir_pin_ignores_post_bind_redirect(self):
"""AC2: after production bind, env STATE_DIR override is ignored."""
mcp_daemon_guard.clear_native_runtime_for_tests()
import mcp_session_state
with tempfile.TemporaryDirectory() as tmp:
legitimate = str(Path(tmp) / "legitimate-state")
rogue = str(Path(tmp) / ".mcp_session_701")
os.makedirs(legitimate, mode=0o700, exist_ok=True)
os.makedirs(rogue, mode=0o700, exist_ok=True)
os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = legitimate
canonical = str((REPO_ROOT / "mcp_server.py").resolve())
def _fake_caller():
return canonical
with patch.object(
mcp_daemon_guard,
"_caller_official_entrypoint_path",
side_effect=_fake_caller,
):
with patch.object(
mcp_daemon_guard, "is_pytest_runtime", return_value=False
):
mcp_daemon_guard.mark_sanctioned_daemon()
mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
pinned = mcp_daemon_guard.pinned_session_state_dir()
self.assertEqual(pinned, str(Path(legitimate).resolve()))
# Attacker redirects env after bind (PR #701).
os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = rogue
self.assertEqual(
mcp_daemon_guard.pinned_session_state_dir(),
str(Path(legitimate).resolve()),
)
self.assertEqual(
mcp_session_state.default_state_dir(),
str(Path(legitimate).resolve()),
)
self.assertNotEqual(
mcp_session_state.default_state_dir(),
str(Path(rogue).resolve()),
)
# Unpinned env view still sees rogue (diagnostics only).
unpinned = mcp_session_state.env_session_state_dir_unpinned()
self.assertTrue(
unpinned == rogue
or Path(unpinned).resolve() == Path(rogue).resolve(),
unpinned,
)
def test_direct_import_env_does_not_authorize_under_force_unsanctioned(self):
mcp_daemon_guard.clear_native_runtime_for_tests()
os.environ[mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV] = "1"
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
self.assertTrue(mcp_daemon_guard.direct_import_env_enabled())
# Under pytest, assert_no_direct_import_bypass is a no-op; FORCE path
# still blocks is_native / assert_sanctioned via force-unsanctioned.
self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
mcp_daemon_guard.assert_sanctioned_mutation_runtime("direct-import")
def test_quarantine_voids_merge_approval_for_contaminated_review(self):
"""AC6AC8: quarantined APPROVED does not satisfy merge approval head."""
entry = {
"verdict": "APPROVED",
"dismissed": False,
"reviewed_head_sha": "6b675f5c834b41f9d74e8a54294ff44dddf28ae4",
"review_id": 431,
"submitted_at": "2026-07-13T23:52:34Z",
"quarantined": True,
}
result = merge_approval_gate.assess_merge_approval_head(
current_head_sha="6b675f5c834b41f9d74e8a54294ff44dddf28ae4",
latest_by_reviewer={"sysadmin": entry},
quarantined_review_ids={431},
)
self.assertFalse(result["approval_at_current_head"])
self.assertEqual(result["quarantined_approvals_at_current_head"], 1)
self.assertIn("quarantined", (result["stale_approval_block_reason"] or ""))
if __name__ == "__main__":
unittest.main()
@@ -1,461 +0,0 @@
"""Regression tests for #698: final-report validator vs canonical schema.
Covers the original #698 lead plus the independent reproduction recorded
during the PR #703 formal review (issue #698 comment 11246):
1. non-dict ``action_log`` entries must fail structured, never crash;
2. the validator must not demand legacy fields the canonical schema forbids
(``Pinned reviewed head``, ``Scratch worktree used``, ``Worktree path``,
``Worktree dirty``, ``Mutations``, ``Next``);
3. a legitimately blocked report (``Candidate head SHA: none``, no formal
verdict) must not owe approval/merge live-head proofs;
4. canonical reviewer lease release must not be misclassified as post-merge
cleanup;
5. structured workflow-load and validation proof must be recognized;
6. review mutations are inferred only from authoritative evidence.
"""
from __future__ import annotations
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import final_report_validator as frv # noqa: E402
import post_merge_cleanup_proof as pmcp # noqa: E402
import pr_work_lease as pwl # noqa: E402
import review_proofs as rp # noqa: E402
from review_final_report_schema import ( # noqa: E402
assess_review_final_report_schema,
)
REVIEWED_HEAD = "a" * 40
LIVE_HEAD = "a" * 40
def _pr703_style_report(
*,
decision: str = "request_changes",
cleanup_mutations: str = (
"released reviewer PR lease via gitea_release_reviewer_pr_lease "
"(terminal lease marker phase=released posted)"
),
) -> str:
"""Canonical-schema report modeled on the PR #703 formal review handoff."""
return f"""
Formal review completed with a REQUEST_CHANGES verdict submitted and read
back via the native review API.
## Controller Handoff
- Task: review-merge-pr
- Repo: Scaled-Tech-Consulting/Gitea-Tools
- Role: reviewer
- Identity: sysadmin / prgs-reviewer
- Active profile: prgs-reviewer
- Runtime context: neutral workspace binding
- Selected PR: 703
- Linked issue: #702 open
- Eligibility class: reviewable
- Queue ordering policy: oldest eligible first
- Inventory pagination proof: has_more=false, total_count=8
- Earlier PRs skipped: none
- Candidate head SHA: {REVIEWED_HEAD}
- Reviewed head SHA: {REVIEWED_HEAD}
- Target branch: master
- Target branch SHA: {"2" * 40}
- Already-landed gate: not landed
- Author-safety result: pass (author differs from reviewer)
- Prior request-changes state: none
- Review worktree used: true
- Review worktree path: branches/review-pr-703-independent
- Review worktree inside branches: true
- Review worktree HEAD state: detached at pinned head
- Review worktree dirty before validation: clean
- Review worktree dirty after validation: clean
- Baseline worktree used: false
- Baseline worktree path: none
- Files reviewed: 4
- Validation: focused 50 passed; related 94 passed; full 2665 passed, 6 skipped
- Official validation integrity status: intact
- Terminal review mutation: one REQUEST_CHANGES review submitted and read back
- Review decision: {decision}
- Merge preflight: not run
- Merge result: none
- Linked issue status: open (live fetch proof: gitea_view_issue)
- Main checkout branch: master
- Main checkout dirty state: clean
- Main checkout updated: false
- File edits by reviewer: none
- Worktree/index mutations: none
- Git ref mutations: git fetch prgs (recorded)
- MCP/Gitea mutations: review submission and lease comments only
- Review mutations: one formal REQUEST_CHANGES verdict
- Merge mutations: none
- Cleanup mutations: {cleanup_mutations}
- External-state mutations: none
- Read-only diagnostics: gitea_view_pr, gitea_get_pr_review_feedback
- Blockers: findings F1-F6 recorded on the PR thread
- Current status: review complete; author remediation required
- Safe next action: author addresses findings and pushes a new head
- Safety statement: no merge attempted; no self-review; no root-checkout edits
- Workflow-load helper result: workflow_hash=da045d1e1f1f boundary_status=clean
- Live head SHA before approval: {LIVE_HEAD}
- Pushes occurred during validation: no
"""
def _blocked_preflight_report() -> str:
"""Blocked-run report modeled on the #702 comment 11164 reproduction."""
return """
Fresh review preflight stopped before any worktree or validation work.
## Controller Handoff
- Task: review-merge-pr
- Repo: Scaled-Tech-Consulting/Gitea-Tools
- Role: reviewer
- Identity: sysadmin / prgs-reviewer
- Active profile: prgs-reviewer
- Runtime context: stale workspace binding detected
- Selected PR: 701
- Linked issue: #699 open
- Eligibility class: blocked-before-validation
- Queue ordering policy: oldest eligible first
- Inventory pagination proof: has_more=false, total_count=8
- Earlier PRs skipped: none
- Candidate head SHA: none
- Reviewed head SHA: none
- Target branch: master
- Target branch SHA: none
- Already-landed gate: not run
- Author-safety result: not run
- Prior request-changes state: none
- Review worktree used: false
- Review worktree path: none
- Review worktree inside branches: not applicable
- Review worktree HEAD state: not applicable
- Review worktree dirty before validation: not applicable
- Review worktree dirty after validation: not applicable
- Baseline worktree used: false
- Baseline worktree path: none
- Files reviewed: 0
- Validation: not run
- Official validation integrity status: not applicable
- Terminal review mutation: none
- Review decision: none
- Merge preflight: not run
- Merge result: none
- Linked issue status: open (live fetch proof: gitea_view_issue)
- Main checkout branch: master
- Main checkout dirty state: clean
- Main checkout updated: false
- File edits by reviewer: none
- Worktree/index mutations: none
- Git ref mutations: none
- MCP/Gitea mutations: none
- Review mutations: none
- Merge mutations: none
- Cleanup mutations: none
- External-state mutations: none
- Read-only diagnostics: gitea_view_pr, gitea_get_runtime_context
- Blockers: runtime bound to a foreign task worktree; mutation prohibited
- Current status: stopped before validation began
- Next actor: operator
- Next action: repair the runtime workspace binding, then rerun the full
review workflow in a fresh reviewer session
- Next prompt: Act as REVIEWER for PR 701 after the operator repairs the
runtime binding; acquire the lease before any validation.
- Safe next action: operator repairs runtime binding, then a fresh reviewer
reruns the full workflow
- Safety statement: no lease acquired; no verdict recorded; no source edits
- Workflow-load helper result: workflow_hash=da045d1e1f1f boundary_status=clean
"""
class TestActionLogRobustness(unittest.TestCase):
"""#698 original lead: non-dict action_log must not crash validation."""
MALFORMED = [
"git fetch prgs",
42,
None,
{"action": "edit", "path": "x.py", "performed": True, "tracked": True},
]
def test_assess_final_report_validator_survives_malformed_entries(self):
result = frv.assess_final_report_validator(
_pr703_style_report(),
"review_pr",
action_log=self.MALFORMED,
)
self.assertIsInstance(result, dict)
rule_ids = {f["rule_id"] for f in result["findings"]}
self.assertIn("shared.action_log_malformed", rule_ids)
def test_malformed_entry_errors_are_sanitized(self):
_entries, findings = frv.sanitize_action_log(["secret-token-abc123"])
self.assertEqual(len(findings), 1)
reason = findings[0]["reason"]
self.assertNotIn("secret-token-abc123", reason)
self.assertIn("str", reason)
self.assertIn("entry 0", reason)
def test_non_list_action_log_is_reported_not_raised(self):
entries, findings = frv.sanitize_action_log("not-a-list")
self.assertEqual(entries, [])
self.assertEqual(len(findings), 1)
self.assertIn("not a list", findings[0]["reason"])
def test_performed_file_mutations_skips_non_dict_entries(self):
performed = rp._performed_file_mutations(
["oops", {"action": "edited", "path": "a.py"}]
)
self.assertEqual(len(performed), 1)
self.assertEqual(performed[0]["path"], "a.py")
def test_schema_entrypoint_survives_string_only_log(self):
result = assess_review_final_report_schema(
_pr703_style_report(),
action_log=["just a string", "another string"],
)
self.assertIsInstance(result, dict)
class TestLegacyFieldRequirementsRemoved(unittest.TestCase):
"""#698: prohibited legacy fields must not be REQUIRED of reports."""
PROHIBITED = (
"Pinned reviewed head",
"Scratch worktree used",
"Worktree path",
"Worktree dirty",
"Workspace mutations",
"Mutations",
"Next",
"Issue/PR",
"Branch/SHA",
"Files changed",
)
def test_review_role_field_table_has_no_prohibited_requirements(self):
names = [name for name, _ in rp.HANDOFF_ROLE_FIELDS["review"]]
for prohibited in ("Pinned reviewed head", "Scratch worktree used",
"Worktree path", "Worktree dirty"):
self.assertNotIn(prohibited, names)
def test_merger_role_field_table_has_no_pinned_reviewed_head(self):
names = [name for name, _ in rp.HANDOFF_ROLE_FIELDS["merger"]]
self.assertNotIn("Pinned reviewed head", names)
def test_canonical_report_missing_fields_never_include_prohibited(self):
result = rp.assess_controller_handoff(
_pr703_style_report(), role="review"
)
for prohibited in self.PROHIBITED:
self.assertNotIn(prohibited, result.get("missing_fields") or [])
def test_canonical_pr703_report_satisfies_required_fields(self):
result = rp.assess_controller_handoff(
_pr703_style_report(), role="review"
)
self.assertEqual(result.get("missing_fields") or [], [])
self.assertEqual(result.get("verdict"), "complete")
class TestBlockedReportAccepted(unittest.TestCase):
"""#698: blocked run with no reviewed head / verdict is legitimate."""
def test_stale_head_proof_waived_before_validation(self):
result = pwl.assess_reviewer_stale_head_final_report(
_blocked_preflight_report()
)
self.assertTrue(result["proven"])
self.assertEqual(result.get("phase"), "blocked_before_validation")
def test_blocked_report_passes_schema_validation(self):
result = assess_review_final_report_schema(_blocked_preflight_report())
blocking = [
f for f in result["findings"] if f["severity"] == "block"
]
self.assertEqual(blocking, [], blocking)
def test_verdict_phase_still_demands_approval_head_proof(self):
report = _blocked_preflight_report().replace(
"- Review decision: none",
"- Review decision: approve",
).replace(
"- Candidate head SHA: none",
f"- Candidate head SHA: {REVIEWED_HEAD}",
)
result = pwl.assess_reviewer_stale_head_final_report(report)
self.assertFalse(result["proven"])
joined = " ".join(result["reasons"])
self.assertIn("before approval", joined)
def test_merge_phase_still_demands_merge_head_proof(self):
report = _pr703_style_report().replace(
"- Merge result: none",
"- Merge result: merged",
)
result = pwl.assess_reviewer_stale_head_final_report(report)
self.assertFalse(result["proven"])
self.assertIn(
"final live head SHA before merge not stated",
result["reasons"],
)
def test_validation_phase_demands_push_disclosure(self):
report = _pr703_style_report().replace(
"- Pushes occurred during validation: no\n", ""
)
result = pwl.assess_reviewer_stale_head_final_report(report)
self.assertFalse(result["proven"])
self.assertIn(
"whether push occurred during validation not stated",
result["reasons"],
)
class TestLeaseReleaseVsPostMergeCleanup(unittest.TestCase):
"""#698 (PR #703 review reproduction): lease release is not cleanup."""
def test_lease_release_cleanup_mutations_do_not_demand_checklist(self):
result = pmcp.assess_post_merge_cleanup_proof(_pr703_style_report())
self.assertFalse(result["block"], result["reasons"])
def test_release_tool_name_alone_is_recognized(self):
report = _pr703_style_report(
cleanup_mutations="gitea_release_reviewer_pr_lease comment 11244"
)
result = pmcp.assess_post_merge_cleanup_proof(report)
self.assertFalse(result["block"], result["reasons"])
def test_substantive_non_lease_cleanup_still_demands_checklist(self):
report = _pr703_style_report(
cleanup_mutations="deleted stale scratch directory manually"
)
result = pmcp.assess_post_merge_cleanup_proof(report)
self.assertTrue(result["block"])
def test_remote_branch_delete_claims_still_demand_full_proof(self):
report = _pr703_style_report(
cleanup_mutations="gitea_delete_branch removed the remote branch"
)
result = pmcp.assess_post_merge_cleanup_proof(report)
self.assertTrue(result["block"])
self.assertTrue(
any("remote branch deletion missing" in r for r in result["reasons"])
)
def test_full_schema_run_accepts_lease_release_report(self):
result = assess_review_final_report_schema(_pr703_style_report())
lease_cleanup_blocks = [
f for f in result["findings"]
if f["rule_id"] == "reviewer.post_merge_cleanup_proof"
]
self.assertEqual(lease_cleanup_blocks, [], lease_cleanup_blocks)
class TestStructuredProofRecognition(unittest.TestCase):
"""#698: structured workflow-load and validation proof must be accepted."""
def test_key_value_workflow_proof_recognized(self):
findings = frv._rule_reviewer_workflow_load_boundary(
_pr703_style_report()
)
self.assertEqual(findings, [], findings)
def test_colon_form_workflow_proof_still_recognized(self):
report = _pr703_style_report().replace(
"- Workflow-load helper result: workflow_hash=da045d1e1f1f "
"boundary_status=clean",
"- Workflow-load helper result: workflow_hash: da045d1e1f1f, "
"boundary_status: clean",
)
findings = frv._rule_reviewer_workflow_load_boundary(report)
self.assertEqual(findings, [], findings)
def test_incomplete_structured_proof_still_blocks(self):
report = _pr703_style_report().replace(
"workflow_hash=da045d1e1f1f boundary_status=clean",
"workflow_hash=da045d1e1f1f",
)
findings = frv._rule_reviewer_workflow_load_boundary(report)
self.assertTrue(findings)
self.assertIn("boundary_status", findings[0]["reason"])
def test_validation_counts_accepted_as_pass_proof(self):
# "Validation: focused 50 passed; ..." must satisfy the reviewed-head
# validation-proof rule (PR #703 reproduction).
result = assess_review_final_report_schema(_pr703_style_report())
head_blocks = [
f for f in result["findings"]
if f["rule_id"] == "reviewer.reviewed_head_without_validation"
]
self.assertEqual(head_blocks, [], head_blocks)
class TestAuthoritativeMutationInference(unittest.TestCase):
"""#698: review mutations inferred only from authoritative evidence."""
def test_read_only_entries_do_not_imply_mutations(self):
report = "## Controller Handoff\n- Mutations: none\n"
findings = frv._rule_reviewer_vague_mutations_none(
report,
action_log=[
{"action": "gitea_view_pr"},
{"action": "gitea_get_pr_review_feedback", "performed": False},
],
)
self.assertEqual(findings, [], findings)
def test_performed_mutation_still_blocks_vague_none(self):
report = "## Controller Handoff\n- Mutations: none\n"
findings = frv._rule_reviewer_vague_mutations_none(
report,
action_log=[{"action": "edit", "path": "a.py", "performed": True}],
)
self.assertTrue(findings)
def test_gated_rejection_is_not_a_mutation(self):
report = "## Controller Handoff\n- Mutations: none\n"
findings = frv._rule_reviewer_vague_mutations_none(
report,
action_log=[
{"action": "edit", "path": "a.py", "performed": True,
"gated_rejected": True},
],
)
self.assertEqual(findings, [], findings)
class TestValidatorRuleErrorContainment(unittest.TestCase):
"""#698: a defective rule fails closed with a sanitized error."""
def test_rule_exception_becomes_sanitized_block_finding(self):
def _boom(report_text):
raise ValueError("raw secret detail that must not leak")
original = frv._RULES_BY_TASK["review_pr"]
frv._RULES_BY_TASK["review_pr"] = [_boom]
try:
result = frv.assess_final_report_validator(
"report body", "review_pr"
)
finally:
frv._RULES_BY_TASK["review_pr"] = original
self.assertTrue(result["blocked"])
finding = next(
f for f in result["findings"]
if f["rule_id"] == "shared.validator_rule_error"
)
self.assertNotIn("raw secret detail", finding["reason"])
self.assertIn("ValueError", finding["reason"])
self.assertIn("_boom", finding["reason"])
if __name__ == "__main__":
unittest.main()
+3 -14
View File
@@ -30,17 +30,11 @@ class TestMcpDaemonGuard(unittest.TestCase):
# Running under pytest already sets PYTEST_CURRENT_TEST.
mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest")
def test_test_native_runtime_install_under_pytest(self):
def test_mark_sanctioned_bootstrap_allows(self):
mcp_daemon_guard.clear_native_runtime_for_tests()
mcp_daemon_guard.install_test_native_runtime()
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport())
# Hermetic unit path still passes under pytest.
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon")
# Production mutation gate rejects test-mode records.
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
mcp_daemon_guard.assert_production_mutation_runtime("gitea_mutation")
self.assertIn("Test-mode", str(ctx.exception))
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
def test_env_alone_insufficient_when_force_unsanctioned(self):
mcp_daemon_guard.clear_native_runtime_for_tests()
@@ -93,11 +87,6 @@ class TestMcpDaemonGuard(unittest.TestCase):
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
gitea_auth.get_auth_header("gitea.prgs.cc")
def test_no_allow_test_bootstrap_public_parameter(self):
"""Production mark must not accept allow_test_bootstrap (#695)."""
sig = __import__("inspect").signature(mcp_daemon_guard.mark_sanctioned_daemon)
self.assertNotIn("allow_test_bootstrap", sig.parameters)
if __name__ == "__main__":
unittest.main()
+8 -10
View File
@@ -957,20 +957,17 @@ class TestControllerHandoff(unittest.TestCase):
if not line.startswith("- Workspace mutations:"))
result = assess_controller_handoff(review_base, role="review")
self.assertEqual(result["verdict"], "incomplete")
# #698: the canonical schema forbids the legacy fields, so the
# validator must demand the canonical names instead.
self.assertIn("Reviewed head SHA", result["missing_fields"])
self.assertIn("Review worktree path", result["missing_fields"])
self.assertIn("Pinned reviewed head", result["missing_fields"])
self.assertIn("Worktree path", result["missing_fields"])
self.assertIn("Merge result", result["missing_fields"])
for legacy in ("Pinned reviewed head", "Scratch worktree used"):
self.assertNotIn(legacy, result["missing_fields"])
complete = review_base + "\n" + "\n".join([
"- Selected PR: #999",
"- Reviewer eligibility: passed",
"- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Review worktree path: /repo/branches/review-pr-999",
"- Review worktree dirty before validation: no",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Worktree path: /repo/branches/review-pr-999",
"- Worktree dirty: no",
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
"- Unrelated local mutations: none",
"- Review decision: approve",
"- Merge result: merged",
@@ -1128,9 +1125,10 @@ class TestReviewHandoffPreciseMutationCategories(unittest.TestCase):
"- Safety: no self-review; no self-merge; no secrets",
"- Selected PR: #999",
"- Reviewer eligibility: passed",
"- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Worktree path: /repo/branches/review-pr-999",
"- Worktree dirty: no",
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
"- Unrelated local mutations: none",
"- Review decision: approve",
"- Merge result: none",