a2da4085e0
Adds --mode break-glass-drill with its own signed admission (drill: "break-glass", attempt id RPF-WP-0064-T10-*, snapshot_label). After the phase A restore and unseal it checks the held root token with lookup-self only, runs a fresh generate-root ceremony with re-entered shares, does one read-only sys/auth action, revokes the generated token (204 then 403) and drills a cancelled attempt. Phase A is unchanged. The verifier types the receipt extension and rejects token, accessor, OTP, nonce, encoded-token and share shapes. Offline 34 and native 16 pass. Founder runbooks for the drill and the read-only production SSO-down walk are in the T10 section. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 4515@bnt-lap001 Assistant-Session: 0bf96127-6e04-41be-bfda-d11717c3f354
384 lines
22 KiB
Python
384 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier for v2 isolated OpenBao restore receipts (RPF-WP-0064-T03).
|
|
|
|
Does not import the adapter. Checks the exact schema, typed digests, result
|
|
consistency, deadline, required teardown evidence, claims not made, refuses
|
|
placeholders and anything shaped like a credential, share, key or KV value,
|
|
and compares ``source_sha256`` against a clean checkout at ``adapter_commit``.
|
|
|
|
RPF-WP-0064-T10 break-glass drill receipts (``drill: "break-glass"``) carry a
|
|
typed extension: held-token facts, generate-root ceremony facts, revocation
|
|
status codes, the cancel drill and the snapshot the held-token result is
|
|
limited to. Every extension value is an enumerated, typed fact; anything else
|
|
(token, accessor, OTP, nonce, encoded token, share) is rejected.
|
|
|
|
Exit 0 = receipt accepted. Prints only PASS/FAIL and a reason code.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
SCHEMA = "railiance-openbao-restore-receipt/v2"
|
|
NAMESPACE = "railiance-openbao-restore-attempt"
|
|
MAX_DURATION = 2700
|
|
TOP = {"schema", "extends", "attempt_id", "material", "phase", "admission_sha256", "admission_namespace",
|
|
"signer_principal", "signer_fingerprint", "anchor_protected", "abort_operator", "adapter_commit",
|
|
"source_sha256", "started_at", "finished_at", "not_after", "deadline_seconds", "elapsed_seconds",
|
|
"image", "image_id", "snapshot", "restored", "network", "objects", "result", "reason", "teardown",
|
|
"kv_data_read", "no_secret_material_recorded", "claims_not_made"}
|
|
SNAPSHOT = {"ciphertext_sha256", "ciphertext_bytes", "plaintext_sha256", "source_cluster_id",
|
|
"source_applied_index", "source_receipt_sha256", "ciphertext_verified", "plaintext_verified"}
|
|
RESTORED = {"cluster_id", "applied_index", "sealed_after_restore", "unsealed_with_original_shares",
|
|
"cluster_id_match", "applied_index_not_behind", "throwaway_init_invalidated", "leader_wait"}
|
|
LEADER_WAIT = {"waited_seconds", "polls", "elected", "last_leader", "last_seal", "last_error"}
|
|
LEADER_FIELDS = {"ha_enabled": bool, "is_self": bool, "leader_address": str, "leader_cluster_address": str,
|
|
"raft_committed_index": int, "raft_applied_index": int, "active_time": str}
|
|
SEAL_FIELDS = {"sealed": bool, "initialized": bool, "storage_type": str, "version": str}
|
|
|
|
|
|
def typed_or_none(value: object, kind: type) -> bool:
|
|
if value is None:
|
|
return True
|
|
if kind is str:
|
|
return isinstance(value, str) and len(value) <= 200
|
|
return type(value) is kind
|
|
NETWORK = {"network_mode", "docker_networks", "interfaces", "published_ports", "mounts", "egress_refused"}
|
|
TEARDOWN = {"container_removed", "container_absent_verified", "plaintext_memfd_closed", "custody_memfd_closed",
|
|
"readback_credential_revoked", "docker_volumes_created", "performed_by"}
|
|
OBJECT = {"object", "path", "present", "source_match", "source_path", "observed_sha256"}
|
|
REQUIRED_OBJECTS = (
|
|
"auth_mount_keycape_services", "auth_mount_netkingdom", "kv_mount_platform", "keycape_services_config",
|
|
"secrets_engine_jwt_role", "netkingdom_oidc_config", "platform_admin_role",
|
|
"policy_secrets_engine_login_self", "kv_registration_metadata",
|
|
)
|
|
DRILL = "break-glass"
|
|
DRILL_TOP = {"drill", "held_root_token", "generate_root", "revocation", "cancel_drill", "claim_limits"}
|
|
HELD = {"valid", "policies", "ttl_zero", "expire_time_null", "orphan"}
|
|
GENERATE_ROOT = {"started", "shares_entered", "complete", "otp_decode_ok", "policies", "minimal_action_path",
|
|
"minimal_action_ok"}
|
|
REVOCATION = {"revoke_status", "post_revoke_lookup"}
|
|
DRILL_CLAIMS = {"production_root_token_validity", "production_generate_root_availability", "sso_down_access_path"}
|
|
DRILL_ABORT = {"held_token_refused", "generate_root_failed", "minimal_action_failed", "revocation_failed",
|
|
"cancel_drill_failed"}
|
|
POLICY_NAME = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,63}$")
|
|
SNAPSHOT_LABEL = re.compile(r"^(?=.*20\d{6})[a-z0-9][a-z0-9._-]{2,95}$")
|
|
CLAIMS = {"consumer_continuity", "live_jwt_login", "phase_b_network", "owner_acceptance",
|
|
"recurring_restore_assurance", "production_redirect"}
|
|
ABORT = {"admission_refused", "preflight_refused", "custody_identity_refused", "ciphertext_mismatch",
|
|
"decrypt_failed", "plaintext_mismatch", "target_start_failed", "network_containment_failed",
|
|
"throwaway_init_failed", "restore_failed", "unseal_refused", "leader_not_elected", "cluster_mismatch",
|
|
"applied_index_behind",
|
|
"readback_credential_failed", "readback_failed", "deadline_exceeded", "operator_abort", "parent_lost",
|
|
"internal_error"}
|
|
DIGEST = re.compile(r"^sha256:[a-f0-9]{64}$")
|
|
TIME = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ$")
|
|
UUID = re.compile(r"^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$")
|
|
# Secret-shaped values: OpenBao/Vault credentials, age identities, PEM blocks,
|
|
# base64 or hex blobs the size of unseal shares/keys, JWT-like triples.
|
|
SECRET_SHAPES = (
|
|
re.compile(r"\b(?:hv[sbr]|s|b|r)\.[A-Za-z0-9_-]{20,}"),
|
|
re.compile(r"AGE-SECRET-KEY-1", re.I),
|
|
re.compile(r"-----BEGIN"),
|
|
re.compile(r"(?<![a-f0-9])[a-f0-9]{32,}(?![a-f0-9])"),
|
|
re.compile(r"eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\."),
|
|
)
|
|
# Tokens without a prefix, accessors, OTPs, encoded tokens: a run of 20+ base62/64
|
|
# characters mixing upper case, lower case and digits. Paths, ids and labels do not.
|
|
MIXED_RUN = re.compile(r"[A-Za-z0-9+/=_]{20,}")
|
|
BASE64_BLOB = re.compile(r"(?<![:A-Za-z0-9+/])[A-Za-z0-9+/]{40,}={0,2}(?![A-Za-z0-9+/=])")
|
|
|
|
|
|
def base64_blob(text: str) -> bool:
|
|
"""Random base64 (shares, keys) mixes cases and digits; repo paths do not."""
|
|
return any(re.search(r"[A-Z]", m) and re.search(r"[a-z]", m) and re.search(r"[0-9+]", m)
|
|
for m in BASE64_BLOB.findall(text))
|
|
|
|
|
|
def mixed_run(text: str) -> bool:
|
|
return any(re.search(r"[A-Z]", m) and re.search(r"[a-z]", m) and re.search(r"[0-9]", m)
|
|
for m in MIXED_RUN.findall(text))
|
|
PLACEHOLDERS = re.compile(r"(?i)replace-with|placeholder|todo|changeme|<[^>]*>|example\.invalid|YYYY")
|
|
FORBIDDEN_KEYS = re.compile(r"(?i)^(?:data|value|values|secret|secrets|token|root_token|unseal_keys.*|keys?|"
|
|
r"otp|nonce|encoded_.*|password|plaintext|accessor|client_token|shares?|id)$")
|
|
|
|
|
|
class Reject(Exception):
|
|
pass
|
|
|
|
|
|
def need(condition: object, code: str) -> None:
|
|
if not condition:
|
|
raise Reject(code)
|
|
|
|
|
|
def strict(raw: bytes) -> object:
|
|
def unique(pairs):
|
|
out = {}
|
|
for key, value in pairs:
|
|
need(key not in out, "duplicate_key")
|
|
out[key] = value
|
|
return out
|
|
return json.loads(raw, object_pairs_hook=unique)
|
|
|
|
|
|
def walk(value: object, path: str = "$"):
|
|
if isinstance(value, dict):
|
|
for key, item in value.items():
|
|
yield path, key, None
|
|
yield from walk(item, f"{path}.{key}")
|
|
elif isinstance(value, list):
|
|
for index, item in enumerate(value):
|
|
yield from walk(item, f"{path}[{index}]")
|
|
else:
|
|
yield path, None, value
|
|
|
|
|
|
def digest_ok(value: object) -> bool:
|
|
return isinstance(value, str) and bool(DIGEST.fullmatch(value)) and len(set(value[7:])) > 2
|
|
|
|
|
|
def parse_time(value: object) -> datetime:
|
|
need(isinstance(value, str) and TIME.fullmatch(value), "bad_time")
|
|
return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=UTC)
|
|
|
|
|
|
def check_content(receipt: dict) -> None:
|
|
for path, key, value in walk(receipt):
|
|
if key is not None:
|
|
need(not FORBIDDEN_KEYS.match(key), "forbidden_field")
|
|
continue
|
|
if isinstance(value, str):
|
|
need(not PLACEHOLDERS.search(value), "placeholder")
|
|
if path == "$.adapter_commit":
|
|
continue
|
|
stripped = re.sub(r"sha256:[a-f0-9]{64}", "", value)
|
|
for shape in SECRET_SHAPES:
|
|
need(not shape.search(stripped), "secret_shaped_value")
|
|
need(not base64_blob(stripped), "secret_shaped_value")
|
|
if path != "$.signer_fingerprint": # SHA256:<base64> is regex-checked above
|
|
need(not mixed_run(stripped), "secret_shaped_value")
|
|
|
|
|
|
def check_sources(receipt: dict, root: Path) -> None:
|
|
sources = receipt["source_sha256"]
|
|
need(isinstance(sources, dict) and sources, "sources")
|
|
head = subprocess.run(["git", "-C", str(root), "rev-parse", "HEAD"], capture_output=True, timeout=10)
|
|
need(head.returncode == 0 and head.stdout.decode().strip() == receipt["adapter_commit"], "source_commit_mismatch")
|
|
dirty = subprocess.run(["git", "-C", str(root), "status", "--porcelain", "--", *sources],
|
|
capture_output=True, timeout=10)
|
|
need(dirty.returncode == 0 and not dirty.stdout.strip(), "source_checkout_dirty")
|
|
for name, expected in sources.items():
|
|
need(digest_ok(expected) and not name.startswith("/") and ".." not in Path(name).parts, "sources")
|
|
path = root / name
|
|
need(path.is_file() and not path.is_symlink(), "source_missing")
|
|
need("sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() == expected, "source_hash_mismatch")
|
|
|
|
|
|
def check_drill(receipt: dict, result: str) -> None:
|
|
"""Typed T10 extension; every value is an enumerated fact, never a credential."""
|
|
need(receipt["drill"] == DRILL, "drill")
|
|
held = receipt["held_root_token"]
|
|
need(isinstance(held, dict) and set(held) == HELD, "drill_fields")
|
|
need(all(held[k] in (True, False, None) for k in ("valid", "ttl_zero", "expire_time_null", "orphan")),
|
|
"drill_fields")
|
|
need(isinstance(held["policies"], list) and len(held["policies"]) <= 32
|
|
and all(isinstance(p, str) and POLICY_NAME.fullmatch(p) for p in held["policies"]), "drill_fields")
|
|
if held["valid"] is not True:
|
|
need(held["policies"] == [] and held["ttl_zero"] is None and held["expire_time_null"] is None
|
|
and held["orphan"] is None, "drill_inconsistent")
|
|
gen = receipt["generate_root"]
|
|
need(isinstance(gen, dict) and set(gen) == GENERATE_ROOT, "drill_fields")
|
|
need(all(type(gen[k]) is bool for k in ("started", "complete", "otp_decode_ok", "minimal_action_ok")),
|
|
"drill_fields")
|
|
need(type(gen["shares_entered"]) is int and 0 <= gen["shares_entered"] <= 2, "drill_fields")
|
|
need(isinstance(gen["policies"], list) and gen["policies"] in ([], ["root"]), "drill_fields")
|
|
need(gen["minimal_action_path"] == "sys/auth", "drill_fields")
|
|
rev = receipt["revocation"]
|
|
need(isinstance(rev, dict) and set(rev) == REVOCATION, "drill_fields")
|
|
need(all(rev[k] is None or (type(rev[k]) is int and 100 <= rev[k] <= 599) for k in REVOCATION), "drill_fields")
|
|
cancel = receipt["cancel_drill"]
|
|
need(isinstance(cancel, dict) and set(cancel) == {"started_then_cancelled"}
|
|
and type(cancel["started_then_cancelled"]) is bool, "drill_fields")
|
|
limits = receipt["claim_limits"]
|
|
need(isinstance(limits, dict) and set(limits) == {"production_root_token_validity"}, "claim_limits")
|
|
basis = limits["production_root_token_validity"]
|
|
need(isinstance(basis, dict) and set(basis) == {"as_of_snapshot", "snapshot_plaintext_sha256"}
|
|
and isinstance(basis["as_of_snapshot"], str) and SNAPSHOT_LABEL.fullmatch(basis["as_of_snapshot"])
|
|
and basis["snapshot_plaintext_sha256"] == receipt["snapshot"]["plaintext_sha256"], "claim_limits")
|
|
need(receipt["objects"] == [], "objects")
|
|
if result == "drilled":
|
|
need(held["valid"] in (True, False), "drill_inconsistent")
|
|
need(gen["started"] and gen["shares_entered"] == 2 and gen["complete"] and gen["otp_decode_ok"]
|
|
and gen["policies"] == ["root"] and gen["minimal_action_ok"], "drill_inconsistent")
|
|
need(rev["revoke_status"] == 204 and rev["post_revoke_lookup"] == 403, "drill_inconsistent")
|
|
need(cancel["started_then_cancelled"] is True, "drill_inconsistent")
|
|
need(receipt["teardown"]["readback_credential_revoked"] is True, "drill_inconsistent")
|
|
elif gen["policies"]:
|
|
need(gen["otp_decode_ok"], "drill_inconsistent")
|
|
|
|
|
|
def verify(receipt: object, *, source_root: Path, not_after: datetime | None = None,
|
|
now: datetime | None = None) -> str:
|
|
drill = isinstance(receipt, dict) and "drill" in receipt
|
|
need(isinstance(receipt, dict) and set(receipt) == (TOP | DRILL_TOP if drill else TOP), "schema_fields")
|
|
need(receipt["schema"] == SCHEMA and receipt["admission_namespace"] == NAMESPACE, "schema")
|
|
need(receipt["extends"] == "docs/openbao-restore-drill-evidence.example.json", "schema")
|
|
need(receipt["phase"] == "A" and receipt["material"] in ("original", "synthetic"), "phase")
|
|
attempt = r"RPF-WP-0064-T10-[A-Za-z0-9-]{3,64}" if drill else r"RPF-WP-0064-T0[47]-[A-Za-z0-9-]{3,64}"
|
|
need(isinstance(receipt["attempt_id"], str) and re.fullmatch(attempt, receipt["attempt_id"]), "attempt_id")
|
|
need(digest_ok(receipt["admission_sha256"]), "admission_digest")
|
|
need(isinstance(receipt["signer_principal"], str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._@-]{0,127}",
|
|
receipt["signer_principal"]), "signer")
|
|
need(isinstance(receipt["signer_fingerprint"], str)
|
|
and re.fullmatch(r"SHA256:[A-Za-z0-9+/]{43}", receipt["signer_fingerprint"]), "signer")
|
|
need(receipt["anchor_protected"] is True or receipt["material"] == "synthetic", "anchor_unprotected")
|
|
need(isinstance(receipt["adapter_commit"], str) and re.fullmatch(r"[a-f0-9]{40}", receipt["adapter_commit"]),
|
|
"adapter_commit")
|
|
image = receipt["image"]
|
|
need(isinstance(image, str) and re.fullmatch(r"quay\.io/openbao/openbao@sha256:[a-f0-9]{64}", image)
|
|
and digest_ok(image.split("@", 1)[1]), "image")
|
|
need(receipt["image_id"] is None or digest_ok(receipt["image_id"]), "image")
|
|
need(receipt["no_secret_material_recorded"] is True and receipt["kv_data_read"] is False, "secret_policy")
|
|
claims = CLAIMS | DRILL_CLAIMS if drill else CLAIMS
|
|
need(isinstance(receipt["claims_not_made"], list) and set(receipt["claims_not_made"]) == claims
|
|
and len(receipt["claims_not_made"]) == len(claims), "claims_not_made")
|
|
|
|
started, finished = parse_time(receipt["started_at"]), parse_time(receipt["finished_at"])
|
|
deadline = receipt["deadline_seconds"]
|
|
need(type(deadline) is int and 60 <= deadline <= MAX_DURATION, "deadline")
|
|
need(receipt["material"] == "synthetic" or deadline == MAX_DURATION, "deadline")
|
|
elapsed = receipt["elapsed_seconds"]
|
|
need(type(elapsed) is int and 0 <= elapsed == int((finished - started).total_seconds()), "elapsed")
|
|
need(elapsed <= deadline + 120, "deadline_overrun")
|
|
need(finished <= parse_time(receipt["not_after"]) or receipt["reason"] == "deadline_exceeded", "window")
|
|
if not_after is not None:
|
|
need((now or datetime.now(UTC)) <= not_after, "verification_window_closed")
|
|
|
|
snap = receipt["snapshot"]
|
|
need(isinstance(snap, dict) and set(snap) == SNAPSHOT, "snapshot_fields")
|
|
for field in ("ciphertext_sha256", "plaintext_sha256", "source_receipt_sha256"):
|
|
need(digest_ok(snap[field]), "snapshot_digest")
|
|
need(type(snap["ciphertext_bytes"]) is int and snap["ciphertext_bytes"] > 0, "snapshot")
|
|
need(isinstance(snap["source_cluster_id"], str) and UUID.fullmatch(snap["source_cluster_id"]), "snapshot")
|
|
need(type(snap["source_applied_index"]) is int and snap["source_applied_index"] > 0, "snapshot")
|
|
need(all(type(snap[f]) is bool for f in ("ciphertext_verified", "plaintext_verified")), "snapshot")
|
|
|
|
restored = receipt["restored"]
|
|
need(isinstance(restored, dict) and set(restored) == RESTORED, "restored_fields")
|
|
need(restored["cluster_id"] is None or (isinstance(restored["cluster_id"], str)
|
|
and UUID.fullmatch(restored["cluster_id"])), "restored")
|
|
need(restored["applied_index"] is None or type(restored["applied_index"]) is int, "restored")
|
|
wait = restored["leader_wait"]
|
|
if wait is not None:
|
|
need(isinstance(wait, dict) and set(wait) == LEADER_WAIT, "leader_wait")
|
|
need(type(wait["waited_seconds"]) is int and type(wait["polls"]) is int
|
|
and type(wait["elected"]) is bool, "leader_wait")
|
|
for name, spec in (("last_leader", LEADER_FIELDS), ("last_seal", SEAL_FIELDS)):
|
|
part = wait[name]
|
|
need(isinstance(part, dict) and set(part) == set(spec)
|
|
and all(typed_or_none(part[k], t) for k, t in spec.items()), "leader_wait")
|
|
need(wait["last_error"] is None or (isinstance(wait["last_error"], str) and len(wait["last_error"]) <= 200),
|
|
"leader_wait")
|
|
network = receipt["network"]
|
|
need(isinstance(network, dict) and set(network) == NETWORK, "network_fields")
|
|
teardown = receipt["teardown"]
|
|
need(isinstance(teardown, dict) and set(teardown) == TEARDOWN, "teardown_fields")
|
|
need(teardown["container_removed"] is True and teardown["container_absent_verified"] is True
|
|
and teardown["plaintext_memfd_closed"] is True and teardown["custody_memfd_closed"] is True
|
|
and teardown["docker_volumes_created"] == 0 and teardown["performed_by"] in ("session", "reaper"),
|
|
"teardown_missing")
|
|
need(teardown["readback_credential_revoked"] in (True, False, None), "teardown_fields")
|
|
|
|
objects = receipt["objects"]
|
|
need(isinstance(objects, list), "objects")
|
|
for row in objects:
|
|
need(isinstance(row, dict) and set(row) == OBJECT, "object_fields")
|
|
need(type(row["present"]) is bool and type(row["source_match"]) is bool, "object_fields")
|
|
need(row["observed_sha256"] is None or digest_ok(row["observed_sha256"]), "object_digest")
|
|
need(not row["present"] or row["observed_sha256"] is not None, "object_digest")
|
|
need(not row["source_match"] or row["present"], "object_fields")
|
|
need("/data/" not in row["path"], "kv_data_path")
|
|
need(row["source_path"] is None or row["source_path"] in receipt["source_sha256"], "object_source")
|
|
names = [row["object"] for row in objects]
|
|
|
|
result, reason = receipt["result"], receipt["reason"]
|
|
if drill:
|
|
check_drill(receipt, result)
|
|
if drill and result == "drilled":
|
|
need(reason == "break_glass_drill_complete", "result_inconsistent")
|
|
need(snap["ciphertext_verified"] and snap["plaintext_verified"], "restore_inconsistent")
|
|
need(restored["sealed_after_restore"] is True and restored["unsealed_with_original_shares"] is True
|
|
and restored["cluster_id_match"] is True and restored["applied_index_not_behind"] is True
|
|
and restored["throwaway_init_invalidated"] is True
|
|
and restored["cluster_id"] == snap["source_cluster_id"]
|
|
and restored["applied_index"] >= snap["source_applied_index"], "restore_inconsistent")
|
|
need(network["network_mode"] == "none" and network["docker_networks"] == ["none"]
|
|
and network["interfaces"] == ["lo"] and network["published_ports"] == [] and network["mounts"] == []
|
|
and network["egress_refused"] is True, "network_not_contained")
|
|
need(teardown["performed_by"] == "session", "teardown_missing")
|
|
need(digest_ok(receipt["image_id"]), "image")
|
|
elif result in ("restored", "refused") and not drill:
|
|
need(tuple(names) == REQUIRED_OBJECTS, "objects_incomplete")
|
|
need(snap["ciphertext_verified"] and snap["plaintext_verified"], "restore_inconsistent")
|
|
need(restored["sealed_after_restore"] is True and restored["unsealed_with_original_shares"] is True
|
|
and restored["cluster_id_match"] is True and restored["applied_index_not_behind"] is True
|
|
and restored["throwaway_init_invalidated"] is True
|
|
and restored["cluster_id"] == snap["source_cluster_id"]
|
|
and restored["applied_index"] >= snap["source_applied_index"], "restore_inconsistent")
|
|
need(network["network_mode"] == "none" and network["docker_networks"] == ["none"]
|
|
and network["interfaces"] == ["lo"] and network["published_ports"] == [] and network["mounts"] == []
|
|
and network["egress_refused"] is True, "network_not_contained")
|
|
need(teardown["readback_credential_revoked"] is True and teardown["performed_by"] == "session",
|
|
"teardown_missing")
|
|
need(digest_ok(receipt["image_id"]), "image")
|
|
all_ok = all(r["present"] and r["source_match"] for r in objects)
|
|
if result == "restored":
|
|
need(all_ok and reason == "all_required_objects_match", "result_inconsistent")
|
|
else:
|
|
missing = any(not r["present"] for r in objects)
|
|
need(not all_ok and reason == ("object_missing" if missing else "object_mismatch"), "result_inconsistent")
|
|
elif result == "aborted":
|
|
need(reason in (ABORT | DRILL_ABORT if drill else ABORT), "result_inconsistent")
|
|
need(not objects or len(objects) == len(REQUIRED_OBJECTS), "objects")
|
|
if network["docker_networks"]:
|
|
need(network["docker_networks"] == ["none"] and network["published_ports"] == [], "network_not_contained")
|
|
else:
|
|
raise Reject("result")
|
|
check_content(receipt)
|
|
check_sources(receipt, source_root)
|
|
return result
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
|
|
parser.add_argument("--receipt", required=True)
|
|
parser.add_argument("--source-root", required=True, help="clean checkout at the receipt's adapter_commit")
|
|
parser.add_argument("--not-after", help="refuse after this UTC time (YYYY-MM-DDTHH:MM:SSZ)")
|
|
parser.add_argument("--expect-result", choices=("restored", "refused", "aborted", "drilled"))
|
|
opts = parser.parse_args(argv)
|
|
try:
|
|
path = Path(opts.receipt)
|
|
need(path.is_file() and not path.is_symlink() and path.stat().st_size <= 256 * 1024, "receipt_file")
|
|
raw = path.read_bytes()
|
|
result = verify(strict(raw), source_root=Path(opts.source_root),
|
|
not_after=parse_time(opts.not_after) if opts.not_after else None)
|
|
need(opts.expect_result in (None, result), "unexpected_result")
|
|
except Reject as exc:
|
|
print(f"FAIL: {exc}")
|
|
return 1
|
|
except (ValueError, OSError, subprocess.SubprocessError):
|
|
print("FAIL: unreadable_receipt")
|
|
return 1
|
|
print(f"PASS: {result} receipt sha256:{hashlib.sha256(raw).hexdigest()}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|