"""Server-side irrecoverable decision-lock provenance authorization (#709 AC5). Authorization is **not** a caller-supplied Boolean. A durable, non-forgeable authorization artifact must be minted under production native MCP transport (or pytest) with a dedicated mutation capability, live head binding, and validated incident evidence. Merger consumption is fail-closed and resolves only the historical-provenance blocker — never normal approval, lease, mergeability, anti-stomp, or workspace gates. """ from __future__ import annotations import hashlib import hmac import json import os import re import uuid from datetime import datetime, timedelta, timezone from typing import Any import mcp_daemon_guard import mcp_session_state from stale_review_decision_lock import heads_equal, normalize_head_sha # Dedicated mutation capability (#709 review 434 F1). CAPABILITY_IRRECOVERABLE_RECOVERY = "gitea.decision_lock.irrecoverable_recovery" KIND_AUTH = "irrecoverable_provenance_authorization" KIND_RECOVERY = mcp_session_state.KIND_IRRECOVERABLE_DECISION_PROVENANCE CONFIRMATION_PREFIX = "IRRECOVERABLE DECISION PROVENANCE PR" # Canonical incident-comment marker (#709 review 435 F5). INCIDENT_MARKER = "IRRECOVERABLE DECISION PROVENANCE INCIDENT" # #709 F7 (review 438): strictly canonical incident evidence. The digest binds # every security-relevant field, so evidence cannot be replayed or substituted # across repositories, PRs, decision locks, heads, actors, or recovery actions. INCIDENT_SCHEMA_VERSION = "2" RECOVERY_ACTION_IRRECOVERABLE_PROVENANCE = "irrecoverable_decision_lock_provenance" SUPPORTED_RECOVERY_ACTIONS = frozenset({RECOVERY_ACTION_IRRECOVERABLE_PROVENANCE}) INCIDENT_FIELD_ORDER = ( "schema_version", "remote", "org", "repo", "pr_number", "decision_lock_id", "destroyed_subject", "recovery_action", "recorded_head_sha", "expected_head_sha", "incident_issue", "evidence_author_id", "evidence_author_login", "mint_actor_id", "mint_actor_login", "key_version", "nonce", "issued_at", ) INCIDENT_DIGEST_FIELD = "content_digest" AUTH_TTL_HOURS = 24.0 RECORD_TYPE = "irrecoverable_decision_provenance" AUTH_TYPE = "irrecoverable_provenance_authorization" # Durable HMAC key origin (#709 review 435 F4). Never generate an ephemeral # production key — mint/verify across reconciler/merger processes and restarts # requires a shared secret from env (or pytest constant / explicit env override). ENV_AUTH_HMAC_KEY = "GITEA_IRRECOVERABLE_AUTH_HMAC_KEY" ENV_AUTH_HMAC_KEY_VERSION = "GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION" DEFAULT_KEY_VERSION = "v1" _PYTEST_AUTH_SECRET = b"pytest-irrecoverable-auth-v1" _PYTEST_KEY_VERSION = "pytest-v1" # #709 F6 (review 438): key version is part of the authenticated data and must # be present exactly once, well-formed, and equal to the configured active # version. There is deliberately no legacy fallback for versionless artifacts. KEY_VERSION_RE = re.compile(r"^[A-Za-z0-9._-]{1,64}$") KEY_VERSION_FIELD = "key_version" # Any of these aliases anywhere in the artifact counts as a key-version field; # more than one occurrence is a duplicate and fails closed even when the values # are identical. _KEY_VERSION_ALIASES = ( "key_version", "keyVersion", "key-version", "auth_key_version", ) _PROCESS_AUTH_SECRET: bytes | None = None _PROCESS_AUTH_KEY_VERSION: str | None = None _PROCESS_AUTH_SECRET_SOURCE: str | None = None class AuthSecretError(RuntimeError): """Raised when the durable HMAC signing key cannot be resolved (fail closed).""" def _now() -> datetime: return datetime.now(timezone.utc) def _now_iso() -> str: return _now().isoformat() def _decode_key_material(raw: str) -> bytes: """Decode operator-supplied key material (hex, base64, or utf-8 literal).""" text = (raw or "").strip() if not text: raise AuthSecretError("empty HMAC key material (fail closed, #709 F4)") # Prefer hex (64 chars = 32 bytes). if len(text) >= 32 and all(c in "0123456789abcdefABCDEF" for c in text): try: if len(text) % 2 == 0: decoded = bytes.fromhex(text) if len(decoded) >= 16: return decoded except ValueError: pass # base64 try: import base64 decoded = base64.b64decode(text, validate=True) if len(decoded) >= 16: return decoded except Exception: pass # utf-8 literal (min 16 chars after strip) encoded = text.encode("utf-8") if len(encoded) < 16: raise AuthSecretError( "HMAC key material too short (need >=16 bytes; fail closed, #709 F4)" ) return encoded def reset_process_auth_secret_for_tests() -> None: """Clear cached HMAC key (tests only).""" global _PROCESS_AUTH_SECRET, _PROCESS_AUTH_KEY_VERSION, _PROCESS_AUTH_SECRET_SOURCE _PROCESS_AUTH_SECRET = None _PROCESS_AUTH_KEY_VERSION = None _PROCESS_AUTH_SECRET_SOURCE = None def _validate_key_version_text(value: Any, *, source: str) -> tuple[str, list[str]]: """Validate a key-version string. Never echoes key material — versions only.""" if isinstance(value, bool) or not isinstance(value, str): return "", [ f"{source} key version is malformed (must be a string; fail closed, #709 F6)" ] text = value.strip() if not text: return "", [f"{source} key version is empty (fail closed, #709 F6)"] if value != text: return "", [ f"{source} key version has surrounding whitespace (fail closed, #709 F6)" ] if not KEY_VERSION_RE.match(text): return "", [ f"{source} key version is malformed (allowed charset A-Za-z0-9._-, " "1-64 chars; fail closed, #709 F6)" ] return text, [] def auth_key_version() -> str: """Return the active configured signing-key version (never secret material).""" _process_secret() # ensure version is resolved with the secret version = _PROCESS_AUTH_KEY_VERSION or "" if not version: raise AuthSecretError( "active HMAC key version is unresolved (fail closed, #709 F6 review 438)" ) return version def supported_key_versions() -> tuple[str, ...]: """Versions accepted for verification. Exactly one version — the configured active one — is honored. Rotation is performed by changing ``GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION`` (and the key); artifacts minted under a superseded version stop verifying, which is the intended fail-closed rotation behavior (#709 F6 review 438). """ return (auth_key_version(),) def extract_artifact_key_version(auth: dict[str, Any] | None) -> dict[str, Any]: """Require exactly one nonempty, well-formed key-version field (#709 F6). Missing, empty, malformed, or duplicated (even identical-valued) key-version fields fail closed. Versionless artifacts are never accepted — there is no legacy fallback. """ if not isinstance(auth, dict): return { "valid": False, "key_version": None, "reasons": ["authorization artifact missing (fail closed)"], } containers: list[tuple[str, dict[str, Any]]] = [("artifact", auth)] for nested in ("native_provenance", "scope"): value = auth.get(nested) if isinstance(value, dict): containers.append((nested, value)) seen: list[tuple[str, Any]] = [] for container_name, container in containers: for alias in _KEY_VERSION_ALIASES: if alias in container: seen.append((f"{container_name}.{alias}", container[alias])) if not seen: return { "valid": False, "key_version": None, "reasons": [ "authorization is missing key_version; versionless artifacts are " "never accepted (no legacy fallback; fail closed, #709 F6 review 438)" ], } if len(seen) > 1: names = ", ".join(sorted(name for name, _ in seen)) return { "valid": False, "key_version": None, "reasons": [ f"authorization carries duplicate key-version fields ({names}); " "exactly one is required, even when the values are identical " "(fail closed, #709 F6 review 438)" ], } name, raw = seen[0] text, reasons = _validate_key_version_text(raw, source=f"authorization {name}") if reasons: return {"valid": False, "key_version": None, "reasons": reasons} return {"valid": True, "key_version": text, "reasons": []} def assess_artifact_key_version( auth: dict[str, Any] | None, *, expected_version: str | None = None, ) -> dict[str, Any]: """Artifact key version must exactly equal the configured active version.""" extracted = extract_artifact_key_version(auth) if not extracted.get("valid"): return extracted have = str(extracted.get("key_version") or "") try: want = (expected_version or "").strip() or auth_key_version() except AuthSecretError as exc: return { "valid": False, "key_version": None, "reasons": [ f"configured HMAC key version unavailable: {exc} " "(fail closed, #709 F6)" ], } if have != want: return { "valid": False, "key_version": None, "reasons": [ f"authorization key_version {have!r} is unknown or does not match " f"the configured expected version {want!r} (fail closed, #709 F6 " "review 438)" ], } return {"valid": True, "key_version": have, "reasons": []} def _process_secret() -> bytes: """Resolve durable HMAC key for irrecoverable auth artifacts (#709 F4). Production (non-pytest): **requires** ``GITEA_IRRECOVERABLE_AUTH_HMAC_KEY``. Silently generating an ephemeral per-process key is forbidden — that made mint+verify fail across merger/reconciler processes and restarts. Pytest: uses a fixed constant unless the env key is set (so cross-process regression tests can inject a shared durable key). """ global _PROCESS_AUTH_SECRET, _PROCESS_AUTH_KEY_VERSION, _PROCESS_AUTH_SECRET_SOURCE if _PROCESS_AUTH_SECRET is not None: return _PROCESS_AUTH_SECRET env_raw = (os.environ.get(ENV_AUTH_HMAC_KEY) or "").strip() env_ver = (os.environ.get(ENV_AUTH_HMAC_KEY_VERSION) or "").strip() pytest = mcp_daemon_guard.is_pytest_runtime() if env_raw: secret = _decode_key_material(env_raw) if not env_ver and not pytest: # #709 F6 (review 438): production must configure the key version # explicitly; silently defaulting it makes rotation ambiguous. raise AuthSecretError( f"{ENV_AUTH_HMAC_KEY_VERSION} is required for production " "irrecoverable auth HMAC; an implicit default key version is " "forbidden (fail closed, #709 F6 review 438)" ) version, version_reasons = _validate_key_version_text( env_ver or (_PYTEST_KEY_VERSION if pytest else ""), source="configured", ) if version_reasons: raise AuthSecretError("; ".join(version_reasons)) _PROCESS_AUTH_SECRET = secret _PROCESS_AUTH_KEY_VERSION = version _PROCESS_AUTH_SECRET_SOURCE = "env" return _PROCESS_AUTH_SECRET if pytest: version, version_reasons = _validate_key_version_text( env_ver or _PYTEST_KEY_VERSION, source="configured" ) if version_reasons: raise AuthSecretError("; ".join(version_reasons)) _PROCESS_AUTH_SECRET = _PYTEST_AUTH_SECRET _PROCESS_AUTH_KEY_VERSION = version _PROCESS_AUTH_SECRET_SOURCE = "pytest_constant" return _PROCESS_AUTH_SECRET # Production fail-closed: do NOT call secrets.token_bytes. raise AuthSecretError( f"{ENV_AUTH_HMAC_KEY} is required for production irrecoverable auth HMAC; " "ephemeral per-process key generation is forbidden (fail closed, #709 F4 " "review 435). Configure a durable shared secret for mint+verify across " "processes and restarts." ) def expected_confirmation(pr_number: int) -> str: """Human intent confirmation text (not an authorization credential).""" return f"{CONFIRMATION_PREFIX} {int(pr_number)}" def auth_state_profile_identity( *, remote: str, org: str, repo: str, pr_number: int, expected_head_sha: str, ) -> str: """Stable durable key segment for one exact-scope authorization.""" head = normalize_head_sha(expected_head_sha) or "nohead" segs = [ KIND_AUTH, mcp_session_state._sanitize_segment(remote), mcp_session_state._sanitize_segment(org), mcp_session_state._sanitize_segment(repo), f"pr{int(pr_number)}", head[:16], ] return "-".join(segs) def recovery_state_profile_identity( *, remote: str, org: str, repo: str, pr_number: int, expected_head_sha: str, ) -> str: head = normalize_head_sha(expected_head_sha) or "nohead" segs = [ KIND_RECOVERY, mcp_session_state._sanitize_segment(remote), mcp_session_state._sanitize_segment(org), mcp_session_state._sanitize_segment(repo), f"pr{int(pr_number)}", head[:16], ] return "-".join(segs) def _scope_payload( *, remote: str, org: str, repo: str, pr_number: int, expected_head_sha: str, incident_issue: int, incident_comment_id: int, destroyed_subject: str | None, issuer_username: str, issuer_profile: str, created_at: str, expires_at: str, authorization_id: str, ) -> dict[str, Any]: return { "authorization_id": authorization_id, "remote": remote, "org": org, "repo": repo, "blocked_pr_number": int(pr_number), "expected_head_sha": normalize_head_sha(expected_head_sha), "incident_issue": int(incident_issue), "incident_comment_id": int(incident_comment_id), "destroyed_subject": (destroyed_subject or "").strip() or None, "issuer_username": issuer_username, "issuer_profile": issuer_profile, "created_at": created_at, "expires_at": expires_at, } def _sign_scope( scope: dict[str, Any], native_provenance: dict[str, Any], *, key_version: str | None = None, ) -> str: """HMAC over key_version + canonical scope + native transport fingerprint. The signing key itself is never serialized. Key *version* is bound into the MAC so operators can rotate durable keys without ambiguous verification. """ material = { "key_version": (key_version or auth_key_version()), "scope": scope, "native": { "native_mcp_transport": bool( native_provenance.get("native_mcp_transport") ), "production_native_mcp_transport": bool( native_provenance.get("production_native_mcp_transport") ), "token_fingerprint": native_provenance.get("token_fingerprint"), "entrypoint": native_provenance.get("entrypoint"), "pid": native_provenance.get("pid"), }, } blob = json.dumps(material, sort_keys=True, separators=(",", ":")).encode( "utf-8" ) return hmac.new(_process_secret(), blob, hashlib.sha256).hexdigest() def assess_capability_for_irrecoverable_recovery( *, allowed_operations: list[str] | None, forbidden_operations: list[str] | None = None, role_kind: str | None = None, profile_name: str | None = None, ) -> dict[str, Any]: """Whether the active profile may mint/use irrecoverable recovery (#709 F5). **Dedicated capability only.** Reconciler role / ``gitea.issue.comment`` equivalence is intentionally rejected so a reconciler cannot self-mint recovery authority by authoring its own incident comment (#709 review 435 F5). Bare ``gitea.read`` is never sufficient. """ import gitea_config allowed = list(allowed_operations or []) forbidden = list(forbidden_operations or []) reasons: list[str] = [] dedicated_ok, _ = gitea_config.check_operation( CAPABILITY_IRRECOVERABLE_RECOVERY, allowed, forbidden ) if dedicated_ok: return { "allowed": True, "capability": CAPABILITY_IRRECOVERABLE_RECOVERY, "via": "dedicated_capability", "reasons": [], } role = (role_kind or "").strip().lower() name = (profile_name or "").strip().lower() role_hint = role or name or "unknown" reasons.append( f"missing dedicated capability {CAPABILITY_IRRECOVERABLE_RECOVERY} " f"(profile={role_hint!r}; reconciler/issue.comment equivalence is " "not accepted — dedicated grant required, #709 F5 review 435; " "gitea.read alone is insufficient)" ) return { "allowed": False, "capability": CAPABILITY_IRRECOVERABLE_RECOVERY, "via": None, "reasons": reasons, } def assess_transport_for_auth_mint() -> dict[str, Any]: """Native transport required for minting non-forgeable auth artifacts.""" reasons: list[str] = [] native = mcp_daemon_guard.is_native_mcp_transport() pytest = mcp_daemon_guard.is_pytest_runtime() production = mcp_daemon_guard.is_production_native_mcp_transport() if not native and not pytest: reasons.append( "irrecoverable provenance authorization requires production native " "MCP transport; ordinary Python processes cannot mint acceptable " "recovery authorization (#709 F1)" ) return { "allowed": not reasons, "native_mcp_transport": native, "production_native_mcp_transport": production, "pytest": pytest, "reasons": reasons, } def incident_actor_identity(payload: dict[str, Any] | None) -> dict[str, Any]: """Resolve one stable actor identity from a comment payload (#709 F7). Uses the immutable numeric user id as the primary identity and requires the login to be internally consistent. Multiple or conflicting ids/logins (e.g. ``user.login`` disagreeing with ``user.username``) fail closed rather than silently preferring one string (review 438 F7). """ reasons: list[str] = [] if not isinstance(payload, dict): return { "valid": False, "user_id": None, "login": None, "reasons": ["actor payload missing (fail closed, #709 F7)"], } ids: set[int] = set() logins: set[str] = set() def _add_id(value: Any) -> None: if isinstance(value, bool) or value is None: return if isinstance(value, int): ids.add(int(value)) elif isinstance(value, str) and value.strip().lstrip("-").isdigit(): ids.add(int(value.strip())) def _add_login(value: Any) -> None: if isinstance(value, str) and value.strip(): logins.add(value.strip()) user = payload.get("user") if isinstance(user, dict): _add_id(user.get("id")) _add_id(user.get("user_id")) _add_login(user.get("login")) _add_login(user.get("username")) elif isinstance(user, str): _add_login(user) for key in ("login", "author", "username"): _add_login(payload.get(key)) _add_id(payload.get("user_id")) if len(logins) > 1: reasons.append( "incident comment author has conflicting logins " f"({sorted(logins)!r}); a single stable identity is required " "(fail closed, #709 F7 review 438)" ) if len(ids) > 1: reasons.append( "incident comment author has conflicting user ids " f"({sorted(ids)!r}); a single stable identity is required " "(fail closed, #709 F7 review 438)" ) if not ids: reasons.append( "incident comment author is missing a stable user id; " "display-name-only identity is not accepted " "(fail closed, #709 F7 review 438)" ) if not logins: reasons.append( "incident comment author login is missing (fail closed, #709 F7)" ) return { "valid": not reasons, "user_id": next(iter(ids)) if len(ids) == 1 else None, "login": next(iter(logins)) if len(logins) == 1 else None, "reasons": reasons, } def canonical_incident_fields( *, remote: str, org: str, repo: str, pr_number: int, decision_lock_id: str, destroyed_subject: str, recovery_action: str, recorded_head_sha: str, expected_head_sha: str, incident_issue: int, evidence_author_id: int, evidence_author_login: str, mint_actor_id: int, mint_actor_login: str, key_version: str, nonce: str, issued_at: str, ) -> dict[str, str]: """Ordered fields that form the authoritative incident content digest. Every security-relevant element of the recovery scope is bound here so the digest cannot be replayed across repositories, PRs, decision locks, heads, actors, or recovery actions (#709 F7 review 438). """ action = str(recovery_action or "").strip() if action not in SUPPORTED_RECOVERY_ACTIONS: raise ValueError( f"unsupported recovery_action {action!r}; supported=" f"{sorted(SUPPORTED_RECOVERY_ACTIONS)!r} (fail closed, #709 F7)" ) fields = { "schema_version": INCIDENT_SCHEMA_VERSION, "remote": str(remote or ""), "org": str(org or ""), "repo": str(repo or ""), "pr_number": str(int(pr_number)), "decision_lock_id": str(decision_lock_id or ""), "destroyed_subject": str(destroyed_subject or ""), "recovery_action": action, "recorded_head_sha": normalize_head_sha(recorded_head_sha) or "", "expected_head_sha": normalize_head_sha(expected_head_sha) or "", "incident_issue": str(int(incident_issue)), "evidence_author_id": str(int(evidence_author_id)), "evidence_author_login": str(evidence_author_login or ""), "mint_actor_id": str(int(mint_actor_id)), "mint_actor_login": str(mint_actor_login or ""), "key_version": str(key_version or ""), "nonce": str(nonce or ""), "issued_at": str(issued_at or ""), } for name in INCIDENT_FIELD_ORDER: value = fields[name] if not value.strip(): raise ValueError( f"canonical incident field {name!r} is empty (fail closed, #709 F7)" ) if value != value.strip(): raise ValueError( f"canonical incident field {name!r} has surrounding whitespace " "(fail closed, #709 F7)" ) if "\n" in value or "\r" in value: raise ValueError( f"canonical incident field {name!r} contains a newline; ambiguous " "evidence cannot be built (fail closed, #709 F7)" ) return fields def incident_content_digest(fields: dict[str, str]) -> str: """SHA-256 hex digest over the canonical field lines (fixed order, no sort).""" lines = [INCIDENT_MARKER] for name in INCIDENT_FIELD_ORDER: if name not in fields: raise ValueError( f"canonical incident field {name!r} missing for digest " "(fail closed, #709 F7)" ) lines.append(f"{name}={fields[name]}") blob = "\n".join(lines).encode("utf-8") return hashlib.sha256(blob).hexdigest() def render_canonical_incident_block(fields: dict[str, str]) -> str: """Render the one accepted canonical representation of *fields*.""" lines = [INCIDENT_MARKER] lines.extend(f"{name}: {fields[name]}" for name in INCIDENT_FIELD_ORDER) lines.append(f"{INCIDENT_DIGEST_FIELD}: {incident_content_digest(fields)}") return "\n".join(lines) def _narrative_is_ambiguous(narrative: str) -> list[str]: """Reject narrative text that could be mistaken for canonical evidence.""" reasons: list[str] = [] known = set(INCIDENT_FIELD_ORDER) | {INCIDENT_DIGEST_FIELD} for line in narrative.split("\n"): text = line.strip() if text == INCIDENT_MARKER: reasons.append( "narrative repeats the canonical marker (ambiguous evidence)" ) continue if ":" in text and text.split(":", 1)[0].strip() in known: reasons.append( f"narrative contains canonical field line {text.split(':', 1)[0]!r} " "(ambiguous evidence)" ) return reasons def build_canonical_incident_body( *, remote: str, org: str, repo: str, pr_number: int, decision_lock_id: str, destroyed_subject: str, recovery_action: str, recorded_head_sha: str, expected_head_sha: str, incident_issue: int, evidence_author_id: int, evidence_author_login: str, mint_actor_id: int, mint_actor_login: str, key_version: str, nonce: str, issued_at: str, narrative: str | None = None, ) -> str: """Build the one accepted canonical incident comment body (#709 F7). The builder is the single source of the accepted format and refuses to emit anything ambiguous: empty/multiline fields, unsupported recovery actions, and narrative text that mimics canonical evidence all raise ``ValueError``. Its own output is re-parsed before return, so a body this function produces always validates as canonical. """ fields = canonical_incident_fields( remote=remote, org=org, repo=repo, pr_number=pr_number, decision_lock_id=decision_lock_id, destroyed_subject=destroyed_subject, recovery_action=recovery_action, recorded_head_sha=recorded_head_sha, expected_head_sha=expected_head_sha, incident_issue=incident_issue, evidence_author_id=evidence_author_id, evidence_author_login=evidence_author_login, mint_actor_id=mint_actor_id, mint_actor_login=mint_actor_login, key_version=key_version, nonce=nonce, issued_at=issued_at, ) body = render_canonical_incident_block(fields) if narrative and narrative.strip(): text = narrative.strip() ambiguous = _narrative_is_ambiguous(text) if ambiguous: raise ValueError( "; ".join(ambiguous) + " (fail closed, #709 F7 review 438)" ) body = f"{body}\n\n{text}" check = parse_canonical_incident_body(body) if not check.get("valid"): raise ValueError( "builder produced a non-canonical body: " + "; ".join(check.get("reasons") or ["unknown"]) ) return body def parse_canonical_incident_body(body: str | None) -> dict[str, Any]: """Strictly parse the canonical incident block (#709 F7 review 438). Enforces the exact schema: marker on the first line, every field present exactly once, in fixed canonical order, with no unknown, duplicate, conflicting, or empty fields anywhere in the body. """ reasons: list[str] = [] text = (body or "").replace("\r\n", "\n").replace("\r", "\n").strip("\n") if not text.strip(): return { "valid": False, "reasons": ["incident comment body is empty (fail closed)"], "fields": {}, } lines = text.split("\n") if lines[0] != INCIDENT_MARKER: return { "valid": False, "reasons": [ f"incident body line 1 must be exactly {INCIDENT_MARKER!r} " "(canonical marker position; fail closed, #709 F7 review 438)" ], "fields": {}, } order = list(INCIDENT_FIELD_ORDER) + [INCIDENT_DIGEST_FIELD] block_len = 1 + len(order) if len(lines) < block_len: return { "valid": False, "reasons": [ "incident body canonical block is incomplete; every field is " "required exactly once in canonical order (fail closed, #709 F7)" ], "fields": {}, } fields: dict[str, str] = {} for index, name in enumerate(order, start=1): line = lines[index] prefix = f"{name}: " if not line.startswith(prefix): return { "valid": False, "reasons": [ f"incident body line {index + 1} must be field {name!r} in " "canonical order (reordered, duplicated, missing, or unknown " "field; fail closed, #709 F7 review 438)" ], "fields": {}, } value = line[len(prefix) :] if not value.strip(): reasons.append( f"incident field {name!r} is empty (fail closed, #709 F7)" ) elif value != value.strip(): reasons.append( f"incident field {name!r} has surrounding whitespace " "(fail closed, #709 F7)" ) fields[name] = value # Nothing outside the canonical block may restate the marker or any field. known = set(order) tail = lines[block_len:] if tail and tail[0].strip(): reasons.append( "narrative must be separated from the canonical block by a blank " "line (fail closed, #709 F7)" ) for line in tail: stripped = line.strip() if stripped == INCIDENT_MARKER: reasons.append( "incident body repeats the canonical marker outside the signed " "block (ambiguous evidence; fail closed, #709 F7 review 438)" ) continue if ":" in stripped and stripped.split(":", 1)[0].strip() in known: duplicated = stripped.split(":", 1)[0].strip() reasons.append( f"incident body restates canonical field {duplicated!r} outside " "the signed block (duplicate/conflicting; fail closed, #709 F7)" ) return { "valid": not reasons, "reasons": reasons, "fields": fields, "canonical_block": "\n".join(lines[:block_len]), } def assess_incident_evidence( *, incident_issue: int | None, incident_comment_id: int | None, comment_payload: dict[str, Any] | None, comment_lookup_error: str | None = None, expected_remote: str | None = None, expected_org: str | None = None, expected_repo: str | None = None, expected_pr_number: int | None = None, expected_head_sha: str | None = None, expected_recorded_head_sha: str | None = None, expected_decision_lock_id: str | None = None, expected_recovery_action: str | None = None, expected_key_version: str | None = None, mint_actor_id: int | None = None, mint_actor_username: str | None = None, reject_self_authored: bool = True, ) -> dict[str, Any]: """Validate strictly canonical, fully scoped incident evidence (#709 F7). The body must be exactly the one canonical representation produced by :func:`build_canonical_incident_body`: marker first, every field present once in fixed order, no duplicates/unknowns/conflicts, digest binding the complete recovery scope, and a single stable actor identity independent of the minting actor. Reordered, duplicated, conflicting, replayed, or substituted evidence fails closed (review 438 F7). """ reasons: list[str] = [] if incident_issue is None or int(incident_issue) <= 0: reasons.append("incident_issue is required and must be a positive integer") if incident_comment_id is None or int(incident_comment_id) <= 0: reasons.append( "incident_comment_id is required and must be a positive integer" ) if comment_lookup_error: reasons.append( f"incident evidence lookup failed: {comment_lookup_error} (fail closed)" ) if not isinstance(comment_payload, dict): if not comment_lookup_error: reasons.append( "incident evidence not found or not a comment object (fail closed)" ) return {"valid": False, "reasons": reasons, "comment": None} cid = comment_payload.get("id") try: if int(cid) != int(incident_comment_id): # type: ignore[arg-type] reasons.append( "incident comment id mismatch against live payload (fail closed)" ) except (TypeError, ValueError): reasons.append("incident comment payload missing valid id (fail closed)") # Edited evidence is not authoritative (#709 F5 review 435). created_at = str(comment_payload.get("created_at") or "").strip() updated_at = str(comment_payload.get("updated_at") or "").strip() if updated_at and created_at and updated_at != created_at: reasons.append( "incident comment has been edited after creation; edited evidence is " "not authoritative (fail closed, #709 F5 review 435)" ) actor = incident_actor_identity(comment_payload) if not actor.get("valid"): reasons.extend(actor.get("reasons") or []) parsed = parse_canonical_incident_body(comment_payload.get("body")) fields = parsed.get("fields") or {} if not parsed.get("valid"): reasons.extend(parsed.get("reasons") or []) else: # Every field is bound to the exact live recovery scope. def _match(name: str, expected: Any, *, label: str | None = None) -> None: if expected is None or str(expected).strip() == "": return have = fields.get(name, "") if have != str(expected).strip(): reasons.append( f"incident body {label or name} does not match the live " f"recovery scope (fail closed, #709 F7 review 438)" ) if fields.get("schema_version") != INCIDENT_SCHEMA_VERSION: reasons.append( f"incident schema_version must be {INCIDENT_SCHEMA_VERSION!r} " "(fail closed, #709 F7)" ) _match("remote", expected_remote) _match("org", expected_org) _match("repo", expected_repo) _match("decision_lock_id", expected_decision_lock_id) _match("recovery_action", expected_recovery_action) _match("key_version", expected_key_version) if fields.get("recovery_action") not in SUPPORTED_RECOVERY_ACTIONS: reasons.append( f"incident recovery_action {fields.get('recovery_action')!r} is " "not a supported recovery action (fail closed, #709 F7)" ) if expected_pr_number is not None: try: if int(fields.get("pr_number", "")) != int(expected_pr_number): reasons.append( "incident body pr_number does not match mint PR " "(fail closed, #709 F7)" ) except (TypeError, ValueError): reasons.append("incident body pr_number invalid (fail closed)") try: if int(fields.get("incident_issue", "")) != int(incident_issue): # type: ignore[arg-type] reasons.append( "incident body incident_issue does not match provided " "incident_issue (fail closed, #709 F7)" ) except (TypeError, ValueError): reasons.append("incident body incident_issue invalid (fail closed)") if expected_head_sha and not heads_equal( fields.get("expected_head_sha"), expected_head_sha ): reasons.append( "incident body expected_head_sha does not match the live mint " "head (fail closed, #709 F7)" ) if expected_recorded_head_sha and not heads_equal( fields.get("recorded_head_sha"), expected_recorded_head_sha ): reasons.append( "incident body recorded_head_sha does not match the recorded " "decision-lock head (fail closed, #709 F7 review 438)" ) # Evidence author must be the actual live comment author. if actor.get("valid"): try: if int(fields.get("evidence_author_id", "")) != int( actor.get("user_id") ): reasons.append( "incident body evidence_author_id does not match the live " "comment author (fail closed, #709 F7 review 438)" ) except (TypeError, ValueError): reasons.append( "incident body evidence_author_id invalid (fail closed)" ) if fields.get("evidence_author_login") != str(actor.get("login") or ""): reasons.append( "incident body evidence_author_login does not match the live " "comment author (fail closed, #709 F7 review 438)" ) # Minting actor must be the live caller, and must not be the author. if mint_actor_id is not None: try: if int(fields.get("mint_actor_id", "")) != int(mint_actor_id): reasons.append( "incident body mint_actor_id does not match the minting " "actor (fail closed, #709 F7 review 438)" ) except (TypeError, ValueError): reasons.append("incident body mint_actor_id invalid (fail closed)") if mint_actor_username and fields.get("mint_actor_login") != str( mint_actor_username ).strip(): reasons.append( "incident body mint_actor_login does not match the minting actor " "(fail closed, #709 F7 review 438)" ) # Digest binds the complete canonical content. try: expected_digest = incident_content_digest(fields) if not hmac.compare_digest( expected_digest.lower(), str(fields.get(INCIDENT_DIGEST_FIELD, "")).strip().lower(), ): reasons.append( "incident content_digest mismatch (forged, substituted, or " "incomplete canonical body; fail closed, #709 F7)" ) except (TypeError, ValueError) as exc: reasons.append( f"incident content_digest recompute failed: {exc} (fail closed)" ) # Reconstruct the exact canonical representation and require equality. try: rebuilt = render_canonical_incident_block(fields) if rebuilt != parsed.get("canonical_block"): reasons.append( "incident body is not the exact canonical representation " "(fail closed, #709 F7 review 438)" ) except (TypeError, ValueError) as exc: reasons.append( f"incident canonical reconstruction failed: {exc} (fail closed)" ) # Independent-author requirement (#709 F5 review 435), by stable identity. if reject_self_authored and actor.get("valid"): if mint_actor_id is not None and int(actor.get("user_id") or -1) == int( mint_actor_id ): reasons.append( "incident comment author is the minting actor; self-authored " "incident evidence is not accepted (fail closed, #709 F5 review 435)" ) elif ( mint_actor_username and str(actor.get("login") or "").strip().lower() == str(mint_actor_username).strip().lower() ): reasons.append( "incident comment author is the minting actor; self-authored " "incident evidence is not accepted (fail closed, #709 F5 review 435)" ) # Hard scope check when URLs are present. issue_url = str( comment_payload.get("issue_url") or comment_payload.get("html_url") or "" ) if expected_org and issue_url and f"/{expected_org}/" not in issue_url: if expected_repo and f"/{expected_repo}/" not in issue_url: reasons.append( "incident evidence URL does not match expected repository " "(fail closed)" ) return { "valid": not reasons, "reasons": reasons, "fields": fields, "comment": { "id": comment_payload.get("id"), "author": actor.get("login"), "author_id": actor.get("user_id"), "created_at": comment_payload.get("created_at"), "canonical": bool(parsed.get("valid")), }, } def assess_live_head_binding( *, expected_head_sha: str | None, live_head_sha: str | None, pr_lookup_error: str | None = None, pr_state: str | None = None, ) -> dict[str, Any]: """expected_head_sha mandatory and must equal live PR head.""" reasons: list[str] = [] want = normalize_head_sha(expected_head_sha) have = normalize_head_sha(live_head_sha) if not want: reasons.append( "expected_head_sha is mandatory and must be a non-empty SHA " "(fail closed, #709 F1)" ) if pr_lookup_error: reasons.append(f"live PR head lookup failed: {pr_lookup_error} (fail closed)") if want and not have: reasons.append("live PR head SHA unavailable (fail closed)") if want and have and not heads_equal(want, have): reasons.append( "expected_head_sha does not equal live PR head " f"(expected={want[:12]}… live={have[:12]}…; fail closed, #709 F1)" ) return { "valid": not reasons, "expected_head_sha": want, "live_head_sha": have, "pr_state": pr_state, "reasons": reasons, } def build_authorization_artifact( *, remote: str, org: str, repo: str, pr_number: int, expected_head_sha: str, incident_issue: int, incident_comment_id: int, destroyed_subject: str | None, issuer_username: str, issuer_profile: str, native_provenance: dict[str, Any] | None = None, ttl_hours: float = AUTH_TTL_HOURS, ) -> dict[str, Any]: """Build a server-side authorization artifact (caller cannot forge signature).""" provenance = dict( native_provenance or mcp_daemon_guard.mutation_provenance_fields() ) created = _now() expires = created + timedelta(hours=float(ttl_hours)) authorization_id = str(uuid.uuid4()) scope = _scope_payload( remote=remote, org=org, repo=repo, pr_number=pr_number, expected_head_sha=expected_head_sha, incident_issue=incident_issue, incident_comment_id=incident_comment_id, destroyed_subject=destroyed_subject, issuer_username=issuer_username, issuer_profile=issuer_profile, created_at=created.isoformat(), expires_at=expires.isoformat(), authorization_id=authorization_id, ) try: kv = auth_key_version() signature = _sign_scope(scope, provenance, key_version=kv) except AuthSecretError as exc: # Surface fail-closed mint: caller must not get a forgeable blank sig. raise AuthSecretError(str(exc)) from exc return { "kind": KIND_AUTH, "auth_type": AUTH_TYPE, "record_type": AUTH_TYPE, "status": "issued", "consumption_state": "issued", "consumed_at": None, "recovery_critical": True, "issue_ref": "#709", "server_signature": signature, "key_version": kv, # Never serialize the secret; only the version id. "native_provenance": provenance, **scope, "timestamp": created.isoformat(), "recorded_at": created.isoformat(), "updated_at": created.isoformat(), } def verify_authorization_artifact( auth: dict[str, Any] | None, *, remote: str, org: str, repo: str, pr_number: int, expected_head_sha: str, incident_issue: int | None = None, incident_comment_id: int | None = None, require_unconsumed: bool = True, now: datetime | None = None, ) -> dict[str, Any]: """Fail-closed verification of a server-side authorization artifact.""" reasons: list[str] = [] if not isinstance(auth, dict): return { "valid": False, "reasons": ["authorization artifact missing (fail closed)"], } if (auth.get("kind") or auth.get("auth_type") or "") not in ( KIND_AUTH, AUTH_TYPE, ) and auth.get("record_type") != AUTH_TYPE: if (auth.get("kind") or "") != KIND_AUTH: reasons.append( f"authorization kind mismatch (expected {KIND_AUTH!r}; fail closed)" ) for field, want in ( ("remote", remote), ("org", org), ("repo", repo), ): have = (str(auth.get(field) or "")).strip() if not have or have != (want or "").strip(): reasons.append( f"authorization {field} mismatch " f"(stored={have!r}, expected={want!r}; fail closed)" ) try: if int(auth.get("blocked_pr_number")) != int(pr_number): reasons.append("authorization PR number mismatch (fail closed)") except (TypeError, ValueError): reasons.append("authorization missing blocked_pr_number (fail closed)") if not heads_equal(auth.get("expected_head_sha"), expected_head_sha): reasons.append("authorization head SHA mismatch (fail closed)") if incident_issue is not None: try: if int(auth.get("incident_issue")) != int(incident_issue): reasons.append("authorization incident_issue mismatch (fail closed)") except (TypeError, ValueError): reasons.append("authorization missing incident_issue (fail closed)") if incident_comment_id is not None: try: if int(auth.get("incident_comment_id")) != int(incident_comment_id): reasons.append( "authorization incident_comment_id mismatch (fail closed)" ) except (TypeError, ValueError): reasons.append("authorization missing incident_comment_id (fail closed)") if not (auth.get("issuer_username") or "").strip(): reasons.append("authorization missing issuer_username (fail closed)") if not (auth.get("issuer_profile") or "").strip(): reasons.append("authorization missing issuer_profile (fail closed)") if not (auth.get("server_signature") or "").strip(): reasons.append("authorization missing server_signature (fail closed)") # #709 F6 (review 438): validate key version *before* any MAC work so an # attacker-chosen version can never select the signing key. Missing, empty, # malformed, duplicated, unknown, and mismatched versions all fail closed. key_version_gate = assess_artifact_key_version(auth) verified_key_version = ( key_version_gate.get("key_version") if key_version_gate.get("valid") else None ) if not key_version_gate.get("valid"): reasons.extend( key_version_gate.get("reasons") or ["authorization key_version invalid (fail closed, #709 F6)"] ) # Recompute signature over stored scope fields + verified key_version. if verified_key_version is None: reasons.append( "authorization server_signature not verified: key version failed " "validation (fail closed, #709 F6 review 438)" ) else: try: scope = _scope_payload( remote=str(auth.get("remote") or ""), org=str(auth.get("org") or ""), repo=str(auth.get("repo") or ""), pr_number=int(auth.get("blocked_pr_number")), expected_head_sha=str(auth.get("expected_head_sha") or ""), incident_issue=int(auth.get("incident_issue")), incident_comment_id=int(auth.get("incident_comment_id")), destroyed_subject=auth.get("destroyed_subject"), issuer_username=str(auth.get("issuer_username") or ""), issuer_profile=str(auth.get("issuer_profile") or ""), created_at=str(auth.get("created_at") or ""), expires_at=str(auth.get("expires_at") or ""), authorization_id=str(auth.get("authorization_id") or ""), ) native = auth.get("native_provenance") or {} if not isinstance(native, dict): native = {} expected_sig = _sign_scope( scope, native, key_version=verified_key_version ) if not hmac.compare_digest( expected_sig, str(auth.get("server_signature") or "") ): reasons.append( "authorization server_signature invalid (forged, corrupt, or " "HMAC key mismatch across process/restart; fail closed, " "#709 F4/F1)" ) except AuthSecretError as exc: reasons.append(f"authorization HMAC key unavailable: {exc} (fail closed)") except (TypeError, ValueError) as exc: reasons.append(f"authorization scope incomplete: {exc} (fail closed)") # Native provenance required on the artifact itself. native = auth.get("native_provenance") or {} if not isinstance(native, dict) or not ( native.get("native_mcp_transport") or native.get("pytest") ): # Pytest artifacts stamp pytest=True via mutation_provenance_fields. if not mcp_daemon_guard.is_pytest_runtime(): if not (isinstance(native, dict) and native.get("native_mcp_transport")): reasons.append( "authorization lacks native transport provenance (fail closed)" ) # Expiry / consumption. now_dt = now or _now() expires_raw = auth.get("expires_at") expires_dt = None if expires_raw: text = str(expires_raw).strip() if text.endswith("Z"): text = text[:-1] + "+00:00" try: expires_dt = datetime.fromisoformat(text) if expires_dt.tzinfo is None: expires_dt = expires_dt.replace(tzinfo=timezone.utc) except ValueError: reasons.append("authorization expires_at unparseable (fail closed)") else: reasons.append("authorization missing expires_at (fail closed)") if expires_dt is not None and now_dt > expires_dt: reasons.append("authorization expired (fail closed)") state = (auth.get("consumption_state") or auth.get("status") or "").strip() if require_unconsumed and state in ("consumed", "expired"): reasons.append( f"authorization already {state}; cannot be replayed (fail closed)" ) if require_unconsumed and auth.get("consumed_at"): reasons.append("authorization already consumed (fail closed)") return { "valid": not reasons, "reasons": reasons, "authorization_id": auth.get("authorization_id"), "consumption_state": state or None, "key_version": verified_key_version, } def build_irrecoverable_provenance_record( *, pr_number: int, head_sha: str, remote: str, org: str, repo: str, actor_username: str | None, profile_name: str | None, reason: str, incident_issue: int, incident_comment_id: int, authorization: dict[str, Any], destroyed_subject: str | None = None, historical_provenance_subject: str | None = None, ) -> dict[str, Any]: """Truthful absence-of-proof record. Never sets applied=True. ``merger_may_accept`` is True only when *authorization* verifies for the exact scope. Caller-supplied Booleans are never consulted. """ head = normalize_head_sha(head_sha) auth_check = verify_authorization_artifact( authorization, remote=remote, org=org, repo=repo, pr_number=pr_number, expected_head_sha=head or "", incident_issue=incident_issue, incident_comment_id=incident_comment_id, require_unconsumed=True, ) may_accept = bool(auth_check.get("valid")) return { "event": "irrecoverable_decision_lock_provenance", "status": "provenance_irrecoverable", "record_type": RECORD_TYPE, "kind": KIND_RECOVERY, "operator_recovery_required": True, "issue_ref": "#709", "recovery_critical": True, "applied": False, "historical_cleanup_proven": False, "timestamp": _now_iso(), "pr_number": int(pr_number), "blocked_pr_number": int(pr_number), "head_sha": head, "remote": remote, "org": org, "repo": repo, "actor_username": actor_username, "profile_name": profile_name, "reason": reason, "incident_issue": int(incident_issue), "incident_comment_id": int(incident_comment_id), # Legacy field for audit readability; not a caller Boolean gate. "incident_ref": f"issue:{int(incident_issue)}/comment:{int(incident_comment_id)}", "authorization_id": authorization.get("authorization_id"), "authorization_issuer": authorization.get("issuer_username"), "authorization_issuer_profile": authorization.get("issuer_profile"), "authorization_verified": may_accept, "authorization_verify_reasons": list(auth_check.get("reasons") or []), "destroyed_subject": (destroyed_subject or "").strip() or None, "historical_provenance_subject": ( (historical_provenance_subject or destroyed_subject or "").strip() or None ), "consumption_state": "issued", "consumed_at": None, "merger_may_accept": may_accept, "acceptance_rule": ( "Merger may accept this record only when a server-side authorization " "artifact verifies for remote/org/repo/PR/exact-head/incident, the " "record is durable and read back, the auth is unexpired and unconsumed, " "and normal merge gates still pass. Resolves only the historical " "prior-provenance blocker; never proves historical cleanup " "(applied=false, historical_cleanup_proven=false)." ), "native_provenance": mcp_daemon_guard.mutation_provenance_fields(), } def format_irrecoverable_audit_comment(record: dict[str, Any]) -> str: """Markdown body for irrecoverable provenance audit (no applied=true claim).""" lines = [ "## Irrecoverable decision-lock provenance (#709)", "", "Status: **PROVENANCE_IRRECOVERABLE** (not applied cleanup)", "", f"- actor: `{record.get('actor_username')}`", f"- profile: `{record.get('profile_name')}`", f"- timestamp: `{record.get('timestamp')}`", f"- PR: `#{record.get('pr_number')}`", f"- head_sha: `{record.get('head_sha')}`", f"- incident_issue: `{record.get('incident_issue')}`", f"- incident_comment_id: `{record.get('incident_comment_id')}`", f"- authorization_id: `{record.get('authorization_id')}`", f"- authorization_issuer: `{record.get('authorization_issuer')}`", f"- authorization_verified: `{record.get('authorization_verified')}`", f"- historical_cleanup_proven: `{record.get('historical_cleanup_proven')}`", f"- applied: `{record.get('applied')}` (must remain false)", f"- merger_may_accept: `{record.get('merger_may_accept')}`", f"- destroyed_subject: `{record.get('destroyed_subject')}`", "", f"Reason: {record.get('reason')}", "", "This record documents **absence of proof**, not successful cleanup.", "It must not be reused for a different PR or head (#709 AC6).", "Authorization is a server-side artifact — not a caller Boolean.", ] return "\n".join(lines) def assess_merger_consumption( recovery: dict[str, Any] | None, authorization: dict[str, Any] | None, *, remote: str, org: str, repo: str, pr_number: int, live_head_sha: str | None, # Normal merge gate outcomes (must still pass independently). approval_at_current_head: bool | None = None, has_blocking_change_requests: bool | None = None, mergeable: bool | None = None, lease_ok: bool | None = None, runtime_ok: bool | None = None, workspace_ok: bool | None = None, anti_stomp_ok: bool | None = None, now: datetime | None = None, ) -> dict[str, Any]: """Fail-closed merger assessment for consuming an irrecoverable recovery. Resolves **only** the historical prior-provenance blocker when all recovery checks pass. Never grants a pass when normal merge gates fail. """ reasons: list[str] = [] result: dict[str, Any] = { "allowed": False, "resolves_prior_provenance_blocker": False, "historical_cleanup_proven": False, "irrecoverable_recovery_authorized": False, "recovery_record_consumed": False, "reasons": reasons, "normal_gates": { "approval_at_current_head": approval_at_current_head, "has_blocking_change_requests": has_blocking_change_requests, "mergeable": mergeable, "lease_ok": lease_ok, "runtime_ok": runtime_ok, "workspace_ok": workspace_ok, "anti_stomp_ok": anti_stomp_ok, }, } if not isinstance(recovery, dict): reasons.append("recovery record missing (fail closed)") return result if (recovery.get("record_type") or recovery.get("kind")) not in ( RECORD_TYPE, KIND_RECOVERY, "irrecoverable_decision_provenance", ): if recovery.get("status") != "provenance_irrecoverable": reasons.append("recovery record type/status invalid (fail closed)") if recovery.get("applied") is True: reasons.append( "recovery record claims applied=true; refuse (fabrication, fail closed)" ) if recovery.get("historical_cleanup_proven") is True: reasons.append( "recovery record claims historical_cleanup_proven=true; refuse " "(fail closed)" ) for field, want in (("remote", remote), ("org", org), ("repo", repo)): have = (str(recovery.get(field) or "")).strip() if have != (want or "").strip(): reasons.append( f"recovery {field} mismatch (stored={have!r}, expected={want!r})" ) try: if int(recovery.get("pr_number") or recovery.get("blocked_pr_number")) != int( pr_number ): reasons.append("recovery PR number mismatch (fail closed)") except (TypeError, ValueError): reasons.append("recovery missing pr_number (fail closed)") if not heads_equal(recovery.get("head_sha"), live_head_sha): reasons.append( "recovery head SHA does not match live PR head (fail closed)" ) if recovery.get("consumption_state") == "consumed" or recovery.get("consumed_at"): # Idempotent: already consumed for this exact scope is OK if head matches. result["recovery_record_consumed"] = True reasons.append("recovery record already consumed (idempotent check)") auth_check = verify_authorization_artifact( authorization, remote=remote, org=org, repo=repo, pr_number=pr_number, expected_head_sha=str(live_head_sha or ""), incident_issue=recovery.get("incident_issue"), incident_comment_id=recovery.get("incident_comment_id"), # When recovery already consumed, allow already-consumed auth for # idempotent re-report; otherwise require unconsumed. require_unconsumed=not result["recovery_record_consumed"], now=now, ) if not auth_check.get("valid"): reasons.extend(auth_check.get("reasons") or ["authorization invalid"]) else: result["irrecoverable_recovery_authorized"] = True if not recovery.get("merger_may_accept") and not result["recovery_record_consumed"]: reasons.append( "recovery record merger_may_accept is false (fail closed)" ) # Auth id binding. if ( authorization and recovery.get("authorization_id") and authorization.get("authorization_id") and recovery.get("authorization_id") != authorization.get("authorization_id") ): reasons.append("recovery authorization_id does not match artifact (fail closed)") # Normal gates: if explicitly False, refuse consumption as merge-authorizing. normal_blockers: list[str] = [] if approval_at_current_head is False: normal_blockers.append("missing/stale approval at current head") if has_blocking_change_requests is True: normal_blockers.append("blocking change requests present") if mergeable is False: normal_blockers.append("PR not mergeable") if lease_ok is False: normal_blockers.append("lease gate failed") if runtime_ok is False: normal_blockers.append("runtime gate failed") if workspace_ok is False: normal_blockers.append("workspace gate failed") if anti_stomp_ok is False: normal_blockers.append("anti-stomp gate failed") if normal_blockers: reasons.append( "recovery cannot bypass normal merge gates: " + "; ".join(normal_blockers) + " (#709 F2)" ) result["resolves_prior_provenance_blocker"] = False result["allowed"] = False result["reasons"] = reasons return result # Filter pure informational "already consumed" when everything else matches # for idempotent success. hard = [ r for r in reasons if "already consumed" not in r ] if not hard and result["irrecoverable_recovery_authorized"]: result["allowed"] = True result["resolves_prior_provenance_blocker"] = True result["historical_cleanup_proven"] = False if result["recovery_record_consumed"]: reasons.append( "idempotent: prior-provenance blocker already resolved for this scope" ) else: reasons.append( "prior-provenance blocker may be resolved by consuming this record " "(historical cleanup remains unproven)" ) result["reasons"] = reasons return result def mark_consumed( record: dict[str, Any], *, consumer_username: str | None, consumer_profile: str | None, ) -> dict[str, Any]: """Return a copy of *record* marked consumed (crash-safe write is caller's job).""" out = dict(record) out["consumption_state"] = "consumed" out["status"] = out.get("status") or "issued" if out.get("kind") == KIND_AUTH or out.get("auth_type") == AUTH_TYPE: out["status"] = "consumed" out["consumed_at"] = _now_iso() out["consumed_by"] = consumer_username out["consumed_by_profile"] = consumer_profile out["updated_at"] = out["consumed_at"] return out def assess_profile_path_identity(profile_identity: str | None) -> dict[str, Any]: """Reject traversal / malformed profile identity segments (#709 F3).""" reasons: list[str] = [] raw = profile_identity if profile_identity is not None else "" text = str(raw) if not text.strip(): reasons.append("profile identity empty (fail closed)") return {"valid": False, "reasons": reasons, "sanitized": None} if text != text.strip(): reasons.append("profile identity has surrounding whitespace (fail closed)") if ".." in text or "/" in text or "\\" in text or "\x00" in text: reasons.append( "profile identity contains path traversal or separator characters " "(fail closed, #709 F3)" ) if text.startswith("-") or text.startswith("."): reasons.append("profile identity has unsafe leading character (fail closed)") # After sanitize, must not collapse to something that collides emptily. sanitized = mcp_session_state._sanitize_segment(text) if sanitized in ("_", ""): reasons.append("profile identity sanitizes to empty (fail closed)") if sanitized != text and any(c in text for c in ("..", "/", "\\")): # Already covered; keep fail closed. pass return { "valid": not reasons, "reasons": reasons, "sanitized": sanitized if not reasons else None, }