fix(workflow): durable HMAC, dedicated mint capability, head-exact approve match (#709)
Address formal review 435 REQUEST_CHANGES on PR #710: - F4: require durable GITEA_IRRECOVERABLE_AUTH_HMAC_KEY (fail closed; no ephemeral per-process secret); bind key_version into HMAC; cross-process verify works - F5: dedicated gitea.decision_lock.irrecoverable_recovery only; reject reconciler equivalence; authoritative incident body + author + content_digest; reject self-authored incident evidence - F3 residual: lock_targets_merged_pr_approval requires recorded-head match when expected_head_sha is provided (legacy no-head approve no longer primary-clears) Co-Authored-By: Grok 4.5 (xAI) <[email protected]>
This commit is contained in:
Grok 4.5
@@ -55,3 +55,12 @@ GITEA_MCP_PROFILE=prgs
|
||||
# GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456
|
||||
# GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456
|
||||
# GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override
|
||||
|
||||
# Durable HMAC key for irrecoverable decision-lock provenance artifacts (#709 F4).
|
||||
# REQUIRED in production native MCP processes that mint or verify recovery
|
||||
# authorization (reconciler + merger must share the same key). Hex (preferred),
|
||||
# base64, or utf-8 literal (>=16 bytes). Never generate an ephemeral per-process
|
||||
# key — cross-process / post-restart verification would fail closed.
|
||||
# GITEA_IRRECOVERABLE_AUTH_HMAC_KEY=0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
|
||||
# Optional key version string bound into the signature (for rotation).
|
||||
# GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION=v1
|
||||
|
||||
+36
-20
@@ -5140,12 +5140,14 @@ def gitea_issue_irrecoverable_provenance_authorization(
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
) -> dict:
|
||||
"""Mint a server-side authorization artifact for irrecoverable recovery (#709 F1).
|
||||
"""Mint a server-side authorization artifact for irrecoverable recovery (#709 F1/F5).
|
||||
|
||||
Non-forgeable: requires production native MCP transport (or pytest), a
|
||||
dedicated/reconciler mutation capability, live head equality, and validated
|
||||
incident evidence. Confirmation is human intent only — never authorization.
|
||||
Caller Booleans are not accepted.
|
||||
Non-forgeable: requires production native MCP transport (or pytest), the
|
||||
dedicated ``gitea.decision_lock.irrecoverable_recovery`` capability (no
|
||||
reconciler equivalence), live head equality, durable HMAC key, and
|
||||
authoritative incident evidence (author + canonical content_digest).
|
||||
Confirmation is human intent only — never authorization. Caller Booleans
|
||||
are not accepted.
|
||||
"""
|
||||
import irrecoverable_provenance as irp
|
||||
|
||||
@@ -5224,7 +5226,7 @@ def gitea_issue_irrecoverable_provenance_authorization(
|
||||
report["reasons"].extend(head_gate.get("reasons") or [])
|
||||
return report
|
||||
|
||||
# Incident evidence live validation.
|
||||
# Incident evidence live validation (author + canonical digest, #709 F5).
|
||||
comment_payload = None
|
||||
comment_err = None
|
||||
try:
|
||||
@@ -5243,6 +5245,10 @@ def gitea_issue_irrecoverable_provenance_authorization(
|
||||
expected_remote=remote,
|
||||
expected_org=o,
|
||||
expected_repo=r,
|
||||
expected_pr_number=pr_number,
|
||||
expected_head_sha=str(expected_head_sha),
|
||||
mint_actor_username=actor,
|
||||
reject_self_authored=True,
|
||||
)
|
||||
if not incident_gate.get("valid"):
|
||||
report["reasons"].extend(incident_gate.get("reasons") or [])
|
||||
@@ -5285,19 +5291,23 @@ def gitea_issue_irrecoverable_provenance_authorization(
|
||||
)
|
||||
return report
|
||||
|
||||
artifact = irp.build_authorization_artifact(
|
||||
remote=remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
pr_number=pr_number,
|
||||
expected_head_sha=str(expected_head_sha),
|
||||
incident_issue=int(incident_issue),
|
||||
incident_comment_id=int(incident_comment_id),
|
||||
destroyed_subject=destroyed_subject,
|
||||
issuer_username=actor,
|
||||
issuer_profile=profile_name or "unknown",
|
||||
native_provenance=mcp_daemon_guard.mutation_provenance_fields(),
|
||||
)
|
||||
try:
|
||||
artifact = irp.build_authorization_artifact(
|
||||
remote=remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
pr_number=pr_number,
|
||||
expected_head_sha=str(expected_head_sha),
|
||||
incident_issue=int(incident_issue),
|
||||
incident_comment_id=int(incident_comment_id),
|
||||
destroyed_subject=destroyed_subject,
|
||||
issuer_username=actor,
|
||||
issuer_profile=profile_name or "unknown",
|
||||
native_provenance=mcp_daemon_guard.mutation_provenance_fields(),
|
||||
)
|
||||
except irp.AuthSecretError as exc:
|
||||
report["reasons"].append(str(exc))
|
||||
return report
|
||||
artifact["kind"] = mcp_session_state.KIND_IRRECOVERABLE_PROVENANCE_AUTH
|
||||
saved = mcp_session_state.save_state(
|
||||
kind=mcp_session_state.KIND_IRRECOVERABLE_PROVENANCE_AUTH,
|
||||
@@ -5313,7 +5323,8 @@ def gitea_issue_irrecoverable_provenance_authorization(
|
||||
report["success"] = True
|
||||
report["reasons"].append(
|
||||
"issued server-side irrecoverable provenance authorization "
|
||||
"(non-forgeable; bound to remote/org/repo/PR/head/incident)"
|
||||
"(non-forgeable; bound to remote/org/repo/PR/head/incident; "
|
||||
f"key_version={artifact.get('key_version')})"
|
||||
)
|
||||
return report
|
||||
|
||||
@@ -5471,8 +5482,13 @@ def gitea_record_irrecoverable_decision_lock_provenance(
|
||||
incident_comment_id=incident_comment_id,
|
||||
comment_payload=comment_payload if isinstance(comment_payload, dict) else None,
|
||||
comment_lookup_error=comment_err,
|
||||
expected_remote=remote,
|
||||
expected_org=o,
|
||||
expected_repo=r,
|
||||
expected_pr_number=pr_number,
|
||||
expected_head_sha=str(expected_head_sha),
|
||||
mint_actor_username=actor,
|
||||
reject_self_authored=True,
|
||||
)
|
||||
if not incident_gate.get("valid"):
|
||||
report["reasons"].extend(incident_gate.get("reasons") or [])
|
||||
|
||||
+375
-68
@@ -14,7 +14,6 @@ import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import uuid
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any
|
||||
@@ -30,14 +29,28 @@ 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"
|
||||
AUTH_TTL_HOURS = 24.0
|
||||
RECORD_TYPE = "irrecoverable_decision_provenance"
|
||||
AUTH_TYPE = "irrecoverable_provenance_authorization"
|
||||
|
||||
# Internal HMAC material is process-local and never caller-supplied. Pytest
|
||||
# gets a deterministic salt so hermetic tests are stable; production uses
|
||||
# transport fingerprint + random secret minted at process start.
|
||||
# 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"
|
||||
|
||||
_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:
|
||||
@@ -48,14 +61,91 @@ 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 auth_key_version() -> str:
|
||||
"""Return the active signing-key version string (never secret material)."""
|
||||
_process_secret() # ensure version is resolved with the secret
|
||||
return _PROCESS_AUTH_KEY_VERSION or DEFAULT_KEY_VERSION
|
||||
|
||||
|
||||
def _process_secret() -> bytes:
|
||||
global _PROCESS_AUTH_SECRET
|
||||
if _PROCESS_AUTH_SECRET is None:
|
||||
if mcp_daemon_guard.is_pytest_runtime():
|
||||
_PROCESS_AUTH_SECRET = b"pytest-irrecoverable-auth-v1"
|
||||
else:
|
||||
_PROCESS_AUTH_SECRET = secrets.token_bytes(32)
|
||||
return _PROCESS_AUTH_SECRET
|
||||
"""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:
|
||||
_PROCESS_AUTH_SECRET = _decode_key_material(env_raw)
|
||||
_PROCESS_AUTH_KEY_VERSION = env_ver or (
|
||||
_PYTEST_KEY_VERSION if pytest else DEFAULT_KEY_VERSION
|
||||
)
|
||||
_PROCESS_AUTH_SECRET_SOURCE = "env"
|
||||
return _PROCESS_AUTH_SECRET
|
||||
|
||||
if pytest:
|
||||
_PROCESS_AUTH_SECRET = _PYTEST_AUTH_SECRET
|
||||
_PROCESS_AUTH_KEY_VERSION = env_ver or _PYTEST_KEY_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:
|
||||
@@ -137,9 +227,19 @@ def _scope_payload(
|
||||
}
|
||||
|
||||
|
||||
def _sign_scope(scope: dict[str, Any], native_provenance: dict[str, Any]) -> str:
|
||||
"""HMAC over canonical scope + native transport fingerprint (non-caller)."""
|
||||
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(
|
||||
@@ -166,15 +266,14 @@ def assess_capability_for_irrecoverable_recovery(
|
||||
role_kind: str | None = None,
|
||||
profile_name: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Whether the active profile may mint/use irrecoverable recovery (#709 F1).
|
||||
"""Whether the active profile may mint/use irrecoverable recovery (#709 F5).
|
||||
|
||||
Accepts the dedicated capability, or a reconciler-shaped profile that
|
||||
already holds issue-comment mutation rights (interim equivalence until
|
||||
operators grant the dedicated op). Never treats bare ``gitea.read`` as
|
||||
sufficient.
|
||||
**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
|
||||
import reconciler_profile
|
||||
|
||||
allowed = list(allowed_operations or [])
|
||||
forbidden = list(forbidden_operations or [])
|
||||
@@ -191,36 +290,15 @@ def assess_capability_for_irrecoverable_recovery(
|
||||
"reasons": [],
|
||||
}
|
||||
|
||||
# Interim: reconciler profile with issue.comment (mutation, not read-only).
|
||||
is_reconciler = reconciler_profile.is_reconciler_profile(allowed, forbidden)
|
||||
comment_ok, _ = gitea_config.check_operation(
|
||||
"gitea.issue.comment", allowed, forbidden
|
||||
)
|
||||
role = (role_kind or "").strip().lower()
|
||||
name = (profile_name or "").strip().lower()
|
||||
role_looks_reconciler = role == "reconciler" or "reconciler" in name
|
||||
if is_reconciler and comment_ok:
|
||||
return {
|
||||
"allowed": True,
|
||||
"capability": CAPABILITY_IRRECOVERABLE_RECOVERY,
|
||||
"via": "reconciler_profile_equivalence",
|
||||
"reasons": [],
|
||||
}
|
||||
if role_looks_reconciler and comment_ok and not dedicated_ok:
|
||||
# Role metadata says reconciler but ops incomplete — still fail if
|
||||
# is_reconciler_profile is false (missing pr.close).
|
||||
reasons.append(
|
||||
"reconciler role metadata without reconciler-required operations "
|
||||
f"(need {CAPABILITY_IRRECOVERABLE_RECOVERY} or reconciler profile "
|
||||
"with gitea.pr.close + gitea.issue.comment; gitea.read alone is "
|
||||
"insufficient, #709 F1)"
|
||||
)
|
||||
else:
|
||||
reasons.append(
|
||||
f"missing dedicated capability {CAPABILITY_IRRECOVERABLE_RECOVERY} "
|
||||
"(gitea.read is insufficient; require reconciler-capable mutation "
|
||||
"profile or explicit grant, #709 F1)"
|
||||
)
|
||||
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,
|
||||
@@ -250,6 +328,108 @@ def assess_transport_for_auth_mint() -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def _incident_author(comment_payload: dict[str, Any]) -> str | None:
|
||||
user = comment_payload.get("user")
|
||||
if isinstance(user, dict):
|
||||
login = (user.get("login") or user.get("username") or "").strip()
|
||||
return login or None
|
||||
if isinstance(user, str) and user.strip():
|
||||
return user.strip()
|
||||
# Some payloads flatten author.
|
||||
for key in ("login", "author", "username"):
|
||||
val = comment_payload.get(key)
|
||||
if isinstance(val, str) and val.strip():
|
||||
return val.strip()
|
||||
return None
|
||||
|
||||
|
||||
def canonical_incident_fields(
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
pr_number: int,
|
||||
expected_head_sha: str,
|
||||
incident_issue: int,
|
||||
) -> dict[str, str]:
|
||||
"""Ordered fields that form the authoritative incident content digest."""
|
||||
head = normalize_head_sha(expected_head_sha) or ""
|
||||
return {
|
||||
"marker": INCIDENT_MARKER,
|
||||
"remote": str(remote or "").strip(),
|
||||
"org": str(org or "").strip(),
|
||||
"repo": str(repo or "").strip(),
|
||||
"pr_number": str(int(pr_number)),
|
||||
"expected_head_sha": head,
|
||||
"incident_issue": str(int(incident_issue)),
|
||||
}
|
||||
|
||||
|
||||
def incident_content_digest(fields: dict[str, str]) -> str:
|
||||
"""SHA-256 hex digest over canonical field lines (stable, sort-free order)."""
|
||||
# Fixed key order — do not sort; operator body must match this order.
|
||||
order = (
|
||||
"marker",
|
||||
"remote",
|
||||
"org",
|
||||
"repo",
|
||||
"pr_number",
|
||||
"expected_head_sha",
|
||||
"incident_issue",
|
||||
)
|
||||
lines = [f"{k}={fields[k]}" for k in order]
|
||||
blob = "\n".join(lines).encode("utf-8")
|
||||
return hashlib.sha256(blob).hexdigest()
|
||||
|
||||
|
||||
def build_canonical_incident_body(
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
pr_number: int,
|
||||
expected_head_sha: str,
|
||||
incident_issue: int,
|
||||
narrative: str | None = None,
|
||||
) -> str:
|
||||
"""Build an operator-postable canonical incident comment body (#709 F5)."""
|
||||
fields = canonical_incident_fields(
|
||||
remote=remote,
|
||||
org=org,
|
||||
repo=repo,
|
||||
pr_number=pr_number,
|
||||
expected_head_sha=expected_head_sha,
|
||||
incident_issue=incident_issue,
|
||||
)
|
||||
digest = incident_content_digest(fields)
|
||||
lines = [
|
||||
fields["marker"],
|
||||
f"remote: {fields['remote']}",
|
||||
f"org: {fields['org']}",
|
||||
f"repo: {fields['repo']}",
|
||||
f"pr_number: {fields['pr_number']}",
|
||||
f"expected_head_sha: {fields['expected_head_sha']}",
|
||||
f"incident_issue: {fields['incident_issue']}",
|
||||
f"content_digest: {digest}",
|
||||
]
|
||||
if narrative and narrative.strip():
|
||||
lines.extend(["", narrative.strip()])
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _parse_incident_field(body: str, name: str) -> str | None:
|
||||
"""Extract ``name: value`` or ``name=value`` from incident body lines."""
|
||||
for line in (body or "").splitlines():
|
||||
text = line.strip()
|
||||
if not text or text.startswith("#"):
|
||||
continue
|
||||
for sep in (":", "="):
|
||||
prefix = f"{name}{sep}"
|
||||
if text.lower().startswith(prefix.lower()):
|
||||
return text[len(prefix) :].strip()
|
||||
return None
|
||||
|
||||
|
||||
def assess_incident_evidence(
|
||||
*,
|
||||
incident_issue: int | None,
|
||||
@@ -259,8 +439,19 @@ def assess_incident_evidence(
|
||||
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,
|
||||
mint_actor_username: str | None = None,
|
||||
reject_self_authored: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate canonical incident evidence is present and live-fetched."""
|
||||
"""Validate authoritative incident evidence (live-fetched, #709 F5).
|
||||
|
||||
Requires:
|
||||
- live comment id match
|
||||
- non-empty author identity
|
||||
- canonical marker + scope fields + content_digest integrity
|
||||
- optional mint-actor independence (mint actor ≠ comment author)
|
||||
"""
|
||||
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")
|
||||
@@ -293,34 +484,139 @@ def assess_incident_evidence(
|
||||
if not body:
|
||||
reasons.append("incident comment body is empty (fail closed)")
|
||||
|
||||
# Soft scope hints when URLs are present (never hard-code issue numbers).
|
||||
author = _incident_author(comment_payload)
|
||||
if not author:
|
||||
reasons.append(
|
||||
"incident comment missing author identity (fail closed, #709 F5)"
|
||||
)
|
||||
|
||||
# Self-mint prevention: minting actor must not be the sole incident author.
|
||||
if (
|
||||
reject_self_authored
|
||||
and author
|
||||
and mint_actor_username
|
||||
and author.strip().lower() == str(mint_actor_username).strip().lower()
|
||||
):
|
||||
reasons.append(
|
||||
"incident comment author matches mint actor; self-authored incident "
|
||||
"evidence is not accepted (fail closed, #709 F5 review 435)"
|
||||
)
|
||||
|
||||
# Canonical content + digest (not any non-empty body).
|
||||
if body and INCIDENT_MARKER not in body:
|
||||
reasons.append(
|
||||
f"incident comment missing canonical marker {INCIDENT_MARKER!r} "
|
||||
"(fail closed, #709 F5)"
|
||||
)
|
||||
elif body:
|
||||
remote_f = _parse_incident_field(body, "remote")
|
||||
org_f = _parse_incident_field(body, "org")
|
||||
repo_f = _parse_incident_field(body, "repo")
|
||||
pr_f = _parse_incident_field(body, "pr_number")
|
||||
head_f = _parse_incident_field(body, "expected_head_sha")
|
||||
issue_f = _parse_incident_field(body, "incident_issue")
|
||||
digest_f = _parse_incident_field(body, "content_digest")
|
||||
|
||||
if expected_remote and remote_f and remote_f != str(expected_remote).strip():
|
||||
reasons.append(
|
||||
"incident body remote does not match expected remote (fail closed)"
|
||||
)
|
||||
if expected_org and org_f and org_f != str(expected_org).strip():
|
||||
reasons.append(
|
||||
"incident body org does not match expected org (fail closed)"
|
||||
)
|
||||
if expected_repo and repo_f and repo_f != str(expected_repo).strip():
|
||||
reasons.append(
|
||||
"incident body repo does not match expected repo (fail closed)"
|
||||
)
|
||||
if expected_pr_number is not None:
|
||||
try:
|
||||
if pr_f is None or int(pr_f) != int(expected_pr_number):
|
||||
reasons.append(
|
||||
"incident body pr_number does not match mint PR "
|
||||
"(fail closed, #709 F5)"
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
reasons.append(
|
||||
"incident body pr_number missing or invalid (fail closed)"
|
||||
)
|
||||
if expected_head_sha:
|
||||
if not head_f or not heads_equal(head_f, expected_head_sha):
|
||||
reasons.append(
|
||||
"incident body expected_head_sha does not match live mint "
|
||||
"head (fail closed, #709 F5)"
|
||||
)
|
||||
try:
|
||||
if issue_f is None or int(issue_f) != int(incident_issue): # type: ignore[arg-type]
|
||||
reasons.append(
|
||||
"incident body incident_issue does not match provided "
|
||||
"incident_issue (fail closed, #709 F5)"
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
reasons.append(
|
||||
"incident body incident_issue missing or invalid (fail closed)"
|
||||
)
|
||||
|
||||
# Content digest integrity when we have enough scope to recompute.
|
||||
if (
|
||||
remote_f
|
||||
and org_f
|
||||
and repo_f
|
||||
and pr_f
|
||||
and head_f
|
||||
and issue_f
|
||||
and digest_f
|
||||
):
|
||||
try:
|
||||
fields = canonical_incident_fields(
|
||||
remote=remote_f,
|
||||
org=org_f,
|
||||
repo=repo_f,
|
||||
pr_number=int(pr_f),
|
||||
expected_head_sha=head_f,
|
||||
incident_issue=int(issue_f),
|
||||
)
|
||||
expected_digest = incident_content_digest(fields)
|
||||
if not hmac.compare_digest(
|
||||
expected_digest.lower(), str(digest_f).strip().lower()
|
||||
):
|
||||
reasons.append(
|
||||
"incident content_digest mismatch (forged or incomplete "
|
||||
"canonical body; fail closed, #709 F5)"
|
||||
)
|
||||
except (TypeError, ValueError) as exc:
|
||||
reasons.append(
|
||||
f"incident content_digest recompute failed: {exc} (fail closed)"
|
||||
)
|
||||
else:
|
||||
reasons.append(
|
||||
"incident body missing required canonical fields "
|
||||
"(remote/org/repo/pr_number/expected_head_sha/incident_issue/"
|
||||
"content_digest; fail closed, #709 F5)"
|
||||
)
|
||||
|
||||
# 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 expected_org not in issue_url and issue_url:
|
||||
# Only fail when URL is present and clearly wrong-org; missing URL ok.
|
||||
if f"/{expected_org}/" not in issue_url:
|
||||
# html_url may be /user/repo/issues/n — check repo if provided
|
||||
if expected_repo and f"/{expected_repo}/" not in issue_url:
|
||||
reasons.append(
|
||||
"incident evidence URL does not match expected repository "
|
||||
"(fail closed)"
|
||||
)
|
||||
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,
|
||||
"comment": {
|
||||
"id": comment_payload.get("id"),
|
||||
"author": (
|
||||
(comment_payload.get("user") or {}).get("login")
|
||||
if isinstance(comment_payload.get("user"), dict)
|
||||
else comment_payload.get("user")
|
||||
),
|
||||
"author": author,
|
||||
"created_at": comment_payload.get("created_at"),
|
||||
"body_len": len(body),
|
||||
"has_canonical_marker": INCIDENT_MARKER in body if body else False,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -396,7 +692,12 @@ def build_authorization_artifact(
|
||||
expires_at=expires.isoformat(),
|
||||
authorization_id=authorization_id,
|
||||
)
|
||||
signature = _sign_scope(scope, provenance)
|
||||
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,
|
||||
@@ -407,6 +708,8 @@ def build_authorization_artifact(
|
||||
"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(),
|
||||
@@ -487,7 +790,7 @@ def verify_authorization_artifact(
|
||||
if not (auth.get("server_signature") or "").strip():
|
||||
reasons.append("authorization missing server_signature (fail closed)")
|
||||
|
||||
# Recompute signature over stored scope fields.
|
||||
# Recompute signature over stored scope fields + key_version.
|
||||
try:
|
||||
scope = _scope_payload(
|
||||
remote=str(auth.get("remote") or ""),
|
||||
@@ -507,14 +810,18 @@ def verify_authorization_artifact(
|
||||
native = auth.get("native_provenance") or {}
|
||||
if not isinstance(native, dict):
|
||||
native = {}
|
||||
expected_sig = _sign_scope(scope, native)
|
||||
stored_kv = str(auth.get("key_version") or auth_key_version())
|
||||
expected_sig = _sign_scope(scope, native, key_version=stored_kv)
|
||||
if not hmac.compare_digest(
|
||||
expected_sig, str(auth.get("server_signature") or "")
|
||||
):
|
||||
reasons.append(
|
||||
"authorization server_signature invalid (forged or corrupt; "
|
||||
"fail closed, #709 F1)"
|
||||
"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)")
|
||||
|
||||
|
||||
@@ -504,7 +504,13 @@ def lock_targets_merged_pr_approval(
|
||||
pr_number: int,
|
||||
expected_head_sha: str | None = None,
|
||||
) -> bool:
|
||||
"""True when *lock*'s last terminal mutation is approve of *pr_number*."""
|
||||
"""True when *lock*'s last terminal mutation is approve of *pr_number*.
|
||||
|
||||
When *expected_head_sha* is provided, a **recorded** terminal head must
|
||||
match. Legacy same-repo approve locks with no recorded head do **not**
|
||||
match (fail closed, #709 F3 residual / review 435) — callers must use the
|
||||
strict secondary path that refuses no-head clears.
|
||||
"""
|
||||
last = last_terminal_mutation(lock)
|
||||
if last is None:
|
||||
return False
|
||||
@@ -513,8 +519,12 @@ def lock_targets_merged_pr_approval(
|
||||
if last.get("pr_number") != pr_number:
|
||||
return False
|
||||
if expected_head_sha:
|
||||
want = normalize_head_sha(expected_head_sha)
|
||||
if not want:
|
||||
return False
|
||||
locked = mutation_head_sha(last, lock)
|
||||
if locked and not heads_equal(locked, expected_head_sha):
|
||||
# Require recorded-head match; unrecorded head is not a match (#709 F3).
|
||||
if not locked or not heads_equal(locked, want):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
"""#709: cross-profile decision-lock cleanup, overwrite protection, recovery.
|
||||
|
||||
Covers AC1–AC8 plus review-434 F1/F2/F3 remediations without fabricating
|
||||
historical PR provenance or special-casing live PR numbers in production code.
|
||||
Covers AC1–AC8 plus review-434 F1/F2/F3 and review-435 F3-residual/F4/F5
|
||||
remediations without fabricating historical PR provenance or special-casing
|
||||
live PR numbers in production code.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -63,6 +64,58 @@ RECONCILER_OPS = [
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
]
|
||||
DEDICATED_RECOVERY_OPS = RECONCILER_OPS + [
|
||||
irp.CAPABILITY_IRRECOVERABLE_RECOVERY,
|
||||
]
|
||||
DURABLE_TEST_HMAC_KEY = "0" * 64 # 32-byte hex durable key for F4 tests
|
||||
|
||||
|
||||
def _canonical_incident_body(
|
||||
*,
|
||||
pr_number=42,
|
||||
head=HEAD_A,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
incident_issue=700,
|
||||
narrative="forensic diagnosis",
|
||||
):
|
||||
return irp.build_canonical_incident_body(
|
||||
remote=remote,
|
||||
org=org,
|
||||
repo=repo,
|
||||
pr_number=pr_number,
|
||||
expected_head_sha=head,
|
||||
incident_issue=incident_issue,
|
||||
narrative=narrative,
|
||||
)
|
||||
|
||||
|
||||
def _incident_comment_payload(
|
||||
*,
|
||||
comment_id=11489,
|
||||
author="controller-ops",
|
||||
pr_number=42,
|
||||
head=HEAD_A,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
incident_issue=700,
|
||||
):
|
||||
return {
|
||||
"id": comment_id,
|
||||
"body": _canonical_incident_body(
|
||||
pr_number=pr_number,
|
||||
head=head,
|
||||
remote=remote,
|
||||
org=org,
|
||||
repo=repo,
|
||||
incident_issue=incident_issue,
|
||||
),
|
||||
"user": {"login": author},
|
||||
"created_at": "2026-07-13T00:00:00Z",
|
||||
"html_url": f"https://gitea.example/{org}/{repo}/issues/{incident_issue}#issuecomment-{comment_id}",
|
||||
}
|
||||
|
||||
|
||||
def _mint_auth(
|
||||
@@ -142,6 +195,24 @@ class TestAC1TargetApproval(unittest.TestCase):
|
||||
)
|
||||
)
|
||||
|
||||
def test_f3_residual_rejects_legacy_no_head_when_expected_head_given(self):
|
||||
"""Primary approve-match requires recorded-head (#709 F3 residual / 435)."""
|
||||
# APPROVE without head fields — legacy ledger.
|
||||
legacy = _lock([APPROVE]) # no head=
|
||||
self.assertIsNone(srdl.mutation_head_sha(APPROVE, legacy))
|
||||
self.assertFalse(
|
||||
srdl.lock_targets_merged_pr_approval(
|
||||
legacy,
|
||||
pr_number=100,
|
||||
expected_head_sha=HEAD_A,
|
||||
)
|
||||
)
|
||||
# Without expected_head_sha, PR-number-only match still works for
|
||||
# non-destructive callers that do not pass a head pin.
|
||||
self.assertTrue(
|
||||
srdl.lock_targets_merged_pr_approval(legacy, pr_number=100)
|
||||
)
|
||||
|
||||
|
||||
class TestF1AuthorizationNotSelfAssertable(unittest.TestCase):
|
||||
def test_operator_authorized_true_cannot_authorize_via_build(self):
|
||||
@@ -336,7 +407,8 @@ class TestF1AuthorizationNotSelfAssertable(unittest.TestCase):
|
||||
)
|
||||
self.assertFalse(a["allowed"])
|
||||
|
||||
def test_reconciler_capability_allowed(self):
|
||||
def test_f5_reconciler_equivalence_rejected(self):
|
||||
"""Reconciler profile without dedicated capability cannot mint (#709 F5)."""
|
||||
a = irp.assess_capability_for_irrecoverable_recovery(
|
||||
allowed_operations=RECONCILER_OPS,
|
||||
forbidden_operations=[
|
||||
@@ -347,7 +419,241 @@ class TestF1AuthorizationNotSelfAssertable(unittest.TestCase):
|
||||
role_kind="reconciler",
|
||||
profile_name="prgs-reconciler",
|
||||
)
|
||||
self.assertFalse(a["allowed"], a)
|
||||
self.assertTrue(
|
||||
any("dedicated" in r for r in a["reasons"]),
|
||||
msg=a["reasons"],
|
||||
)
|
||||
|
||||
def test_f5_dedicated_capability_allowed(self):
|
||||
a = irp.assess_capability_for_irrecoverable_recovery(
|
||||
allowed_operations=DEDICATED_RECOVERY_OPS,
|
||||
forbidden_operations=[
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.review",
|
||||
],
|
||||
role_kind="reconciler",
|
||||
profile_name="prgs-reconciler",
|
||||
)
|
||||
self.assertTrue(a["allowed"], a)
|
||||
self.assertEqual(a["via"], "dedicated_capability")
|
||||
|
||||
|
||||
class TestF5AuthoritativeIncidentEvidence(unittest.TestCase):
|
||||
def test_any_nonempty_body_rejected(self):
|
||||
g = irp.assess_incident_evidence(
|
||||
incident_issue=700,
|
||||
incident_comment_id=11489,
|
||||
comment_payload={
|
||||
"id": 11489,
|
||||
"body": "random forensic note without canonical fields",
|
||||
"user": {"login": "controller-ops"},
|
||||
},
|
||||
expected_remote="prgs",
|
||||
expected_org="Scaled-Tech-Consulting",
|
||||
expected_repo="Gitea-Tools",
|
||||
expected_pr_number=42,
|
||||
expected_head_sha=HEAD_A,
|
||||
)
|
||||
self.assertFalse(g["valid"], g)
|
||||
|
||||
def test_missing_author_rejected(self):
|
||||
body = _canonical_incident_body()
|
||||
g = irp.assess_incident_evidence(
|
||||
incident_issue=700,
|
||||
incident_comment_id=11489,
|
||||
comment_payload={"id": 11489, "body": body, "user": {}},
|
||||
expected_remote="prgs",
|
||||
expected_org="Scaled-Tech-Consulting",
|
||||
expected_repo="Gitea-Tools",
|
||||
expected_pr_number=42,
|
||||
expected_head_sha=HEAD_A,
|
||||
)
|
||||
self.assertFalse(g["valid"], g)
|
||||
self.assertTrue(any("author" in r for r in g["reasons"]), g["reasons"])
|
||||
|
||||
def test_self_authored_rejected(self):
|
||||
g = irp.assess_incident_evidence(
|
||||
incident_issue=700,
|
||||
incident_comment_id=11489,
|
||||
comment_payload=_incident_comment_payload(author="sysadmin"),
|
||||
expected_remote="prgs",
|
||||
expected_org="Scaled-Tech-Consulting",
|
||||
expected_repo="Gitea-Tools",
|
||||
expected_pr_number=42,
|
||||
expected_head_sha=HEAD_A,
|
||||
mint_actor_username="sysadmin",
|
||||
reject_self_authored=True,
|
||||
)
|
||||
self.assertFalse(g["valid"], g)
|
||||
self.assertTrue(
|
||||
any("self-authored" in r for r in g["reasons"]), g["reasons"]
|
||||
)
|
||||
|
||||
def test_canonical_body_with_independent_author_accepted(self):
|
||||
g = irp.assess_incident_evidence(
|
||||
incident_issue=700,
|
||||
incident_comment_id=11489,
|
||||
comment_payload=_incident_comment_payload(author="controller-ops"),
|
||||
expected_remote="prgs",
|
||||
expected_org="Scaled-Tech-Consulting",
|
||||
expected_repo="Gitea-Tools",
|
||||
expected_pr_number=42,
|
||||
expected_head_sha=HEAD_A,
|
||||
mint_actor_username="sysadmin",
|
||||
reject_self_authored=True,
|
||||
)
|
||||
self.assertTrue(g["valid"], g)
|
||||
|
||||
def test_tampered_content_digest_rejected(self):
|
||||
payload = _incident_comment_payload()
|
||||
payload["body"] = payload["body"].replace(
|
||||
"content_digest: ", "content_digest: " + "f" * 64 + "x"
|
||||
)
|
||||
# Force bad digest line
|
||||
lines = []
|
||||
for line in payload["body"].splitlines():
|
||||
if line.startswith("content_digest:"):
|
||||
lines.append("content_digest: " + "0" * 64)
|
||||
else:
|
||||
lines.append(line)
|
||||
payload["body"] = "\n".join(lines)
|
||||
g = irp.assess_incident_evidence(
|
||||
incident_issue=700,
|
||||
incident_comment_id=11489,
|
||||
comment_payload=payload,
|
||||
expected_remote="prgs",
|
||||
expected_org="Scaled-Tech-Consulting",
|
||||
expected_repo="Gitea-Tools",
|
||||
expected_pr_number=42,
|
||||
expected_head_sha=HEAD_A,
|
||||
mint_actor_username="sysadmin",
|
||||
)
|
||||
self.assertFalse(g["valid"], g)
|
||||
self.assertTrue(
|
||||
any("content_digest" in r for r in g["reasons"]), g["reasons"]
|
||||
)
|
||||
|
||||
|
||||
class TestF4DurableHmacKey(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
os.environ.pop(irp.ENV_AUTH_HMAC_KEY, None)
|
||||
os.environ.pop(irp.ENV_AUTH_HMAC_KEY_VERSION, None)
|
||||
irp.reset_process_auth_secret_for_tests()
|
||||
|
||||
def test_key_version_bound_into_artifact(self):
|
||||
irp.reset_process_auth_secret_for_tests()
|
||||
auth = _mint_auth()
|
||||
self.assertIn("key_version", auth)
|
||||
self.assertTrue(auth["key_version"])
|
||||
self.assertNotIn("server_secret", auth)
|
||||
self.assertNotIn("hmac_key", auth)
|
||||
|
||||
def test_production_fails_closed_without_durable_key(self):
|
||||
"""Non-pytest process without env key must not generate ephemeral secret."""
|
||||
script = (
|
||||
"import os, sys\n"
|
||||
"os.environ.pop('PYTEST_CURRENT_TEST', None)\n"
|
||||
"os.environ.pop('GITEA_IRRECOVERABLE_AUTH_HMAC_KEY', None)\n"
|
||||
# Force non-pytest path by patching guard after import.
|
||||
"import mcp_daemon_guard as g\n"
|
||||
"g.is_pytest_runtime = lambda: False\n"
|
||||
"import irrecoverable_provenance as irp\n"
|
||||
"irp.reset_process_auth_secret_for_tests()\n"
|
||||
"try:\n"
|
||||
" irp._process_secret()\n"
|
||||
" print('UNEXPECTED_OK')\n"
|
||||
"except irp.AuthSecretError as e:\n"
|
||||
" print('FAIL_CLOSED', 'ephemeral' in str(e).lower() or 'required' in str(e).lower())\n"
|
||||
)
|
||||
env = {k: v for k, v in os.environ.items() if not k.startswith("PYTEST")}
|
||||
env.pop("PYTEST_CURRENT_TEST", None)
|
||||
env.pop(irp.ENV_AUTH_HMAC_KEY, None)
|
||||
proc = subprocess.run(
|
||||
[sys.executable, "-c", script],
|
||||
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=30,
|
||||
)
|
||||
self.assertEqual(proc.returncode, 0, proc.stderr)
|
||||
out = (proc.stdout or "").strip()
|
||||
self.assertTrue(out.startswith("FAIL_CLOSED"), msg=out)
|
||||
|
||||
def test_cross_process_verify_with_durable_key(self):
|
||||
"""Mint in process A, verify in process B with same durable key (#709 F4)."""
|
||||
import json as _json
|
||||
|
||||
mint_script = (
|
||||
"import json, os, irrecoverable_provenance as irp\n"
|
||||
"irp.reset_process_auth_secret_for_tests()\n"
|
||||
"auth = irp.build_authorization_artifact(\n"
|
||||
" remote='prgs', org='o', repo='r', pr_number=10,\n"
|
||||
f" expected_head_sha={HEAD_A!r}, incident_issue=1, incident_comment_id=2,\n"
|
||||
" destroyed_subject=None, issuer_username='sysadmin',\n"
|
||||
" issuer_profile='prgs-reconciler',\n"
|
||||
" native_provenance={'native_mcp_transport': True, 'pytest': True,\n"
|
||||
" 'token_fingerprint': 'fp', 'entrypoint': 'pytest', 'pid': 1})\n"
|
||||
"print(json.dumps(auth))\n"
|
||||
)
|
||||
env = dict(os.environ)
|
||||
env[irp.ENV_AUTH_HMAC_KEY] = DURABLE_TEST_HMAC_KEY
|
||||
env[irp.ENV_AUTH_HMAC_KEY_VERSION] = "test-v1"
|
||||
mint = subprocess.run(
|
||||
[sys.executable, "-c", mint_script],
|
||||
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=30,
|
||||
)
|
||||
self.assertEqual(mint.returncode, 0, mint.stderr)
|
||||
auth = _json.loads(mint.stdout.strip())
|
||||
self.assertEqual(auth.get("key_version"), "test-v1")
|
||||
|
||||
verify_script = (
|
||||
"import json, sys, irrecoverable_provenance as irp\n"
|
||||
"irp.reset_process_auth_secret_for_tests()\n"
|
||||
"auth = json.loads(sys.stdin.read())\n"
|
||||
"v = irp.verify_authorization_artifact(\n"
|
||||
" auth, remote='prgs', org='o', repo='r', pr_number=10,\n"
|
||||
f" expected_head_sha={HEAD_A!r}, incident_issue=1, incident_comment_id=2)\n"
|
||||
"print(v.get('valid'), v.get('reasons'))\n"
|
||||
)
|
||||
verify = subprocess.run(
|
||||
[sys.executable, "-c", verify_script],
|
||||
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
input=_json.dumps(auth),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=30,
|
||||
)
|
||||
self.assertEqual(verify.returncode, 0, verify.stderr)
|
||||
self.assertTrue(
|
||||
(verify.stdout or "").strip().startswith("True"),
|
||||
msg=verify.stdout + verify.stderr,
|
||||
)
|
||||
|
||||
# Different durable key must fail verification (cross-process mismatch).
|
||||
env_bad = dict(env)
|
||||
env_bad[irp.ENV_AUTH_HMAC_KEY] = "1" * 64
|
||||
verify_bad = subprocess.run(
|
||||
[sys.executable, "-c", verify_script],
|
||||
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
input=_json.dumps(auth),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env_bad,
|
||||
timeout=30,
|
||||
)
|
||||
self.assertEqual(verify_bad.returncode, 0, verify_bad.stderr)
|
||||
self.assertTrue(
|
||||
(verify_bad.stdout or "").strip().startswith("False"),
|
||||
msg=verify_bad.stdout,
|
||||
)
|
||||
|
||||
|
||||
class TestF2MergerConsumer(unittest.TestCase):
|
||||
@@ -807,7 +1113,7 @@ class TestIrrecoverableToolF1(unittest.TestCase):
|
||||
"GITEA_MCP_SESSION_STATE_DIR": self._tmp.name,
|
||||
"GITEA_SESSION_PROFILE_LOCK": "prgs-reconciler",
|
||||
"GITEA_PROFILE_NAME": "prgs-reconciler",
|
||||
"GITEA_ALLOWED_OPERATIONS": ",".join(RECONCILER_OPS),
|
||||
"GITEA_ALLOWED_OPERATIONS": ",".join(DEDICATED_RECOVERY_OPS),
|
||||
},
|
||||
clear=False,
|
||||
)
|
||||
@@ -824,7 +1130,7 @@ class TestIrrecoverableToolF1(unittest.TestCase):
|
||||
return {
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": RECONCILER_OPS,
|
||||
"allowed_operations": DEDICATED_RECOVERY_OPS,
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
@@ -924,17 +1230,18 @@ class TestIrrecoverableToolF1(unittest.TestCase):
|
||||
if "/pulls/" in str(url):
|
||||
return {"head": {"sha": HEAD_A}, "state": "open"}
|
||||
if "/issues/comments/" in str(url):
|
||||
return {
|
||||
"id": 11489,
|
||||
"body": "forensic diagnosis",
|
||||
"user": {"login": "sysadmin"},
|
||||
"created_at": "2026-07-13T00:00:00Z",
|
||||
}
|
||||
return _incident_comment_payload(
|
||||
comment_id=11489,
|
||||
author="controller-ops",
|
||||
pr_number=50,
|
||||
head=HEAD_A,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
incident_issue=700,
|
||||
)
|
||||
raise AssertionError(f"unexpected API {method} {url}")
|
||||
|
||||
common = dict(
|
||||
get_profile=self._profile(),
|
||||
)
|
||||
with patch.object(self.mcp, "get_profile", return_value=self._profile()), patch.object(
|
||||
self.mcp, "_authenticated_username", return_value="sysadmin"
|
||||
), patch.object(
|
||||
|
||||
Reference in New Issue
Block a user