AID-E-004: Incident Exploit-Path Closure Verification
After applicable eviction is complete and the canonical Harden or Restore owner has produced signed fix evidence, independently verify that the exact exploited path is closed across the complete incident-defined affected population. Bind the population to immutable asset, workload, service, release, fix-artifact, and policy identities; read back effective versions, configurations, policies, routes, and enforcement state using a separately credentialed verifier; then replay the exact exploit plus policy-defined negative, bypass, stale-state, and partial-rollout cases from an independent vantage point. Accept closure only when every applicable affected target consumes the exact approved fix and every required replay shows that the exploited path can no longer reach the protected effect. An omitted target, missing or unverifiable fix receipt, unavailable readback, incomplete replay population, verifier error, deployment-tool success response, or owner assertion cannot establish closure.
Pillars: infra, app
Lifecycle: improvement
Scope boundary
Owns independent incident-scoped verification that the exact exploited path is closed across the complete affected population after the canonical Harden or Restore owner supplies signed fix evidence. It does not implement or inherit the patch, configuration, authorization, isolation, or recovered-artifact result.
AID-R-004: AID-E-004 verifies closure of the exact exploited path for the incident-defined population; AID-R-004 propagates the proven remediation across the wider affected fleet and future baselines. Incident closure evidence is an input to fleet propagation, but it does not prove organization-wide rollout or recurrence prevention.
Implementation guidance
- AID-E-004-G001: Verify the incident scope and canonical Harden or Restore fix receipts before closure testing.
- AID-E-004-G002: Resolve and sign the complete incident-defined affected target and exploit-path population.
- AID-E-004-G003: Independently read back effective fix state across every affected target and enforcement point.
- AID-E-004-G004: Reconstruct a safe, immutable replay bundle for the exact exploited path.
- AID-E-004-G005: Replay the exact exploit against every applicable affected path and observe the protected effect.
- AID-E-004-G006: Replay policy-defined bypass, negative, stale-state, and partial-rollout cases without expanding the fix scope.
- AID-E-004-G007: Run clean functional, authorization, and resilience controls against the fixed affected population.
- AID-E-004-G008: Compute and sign an independent exploit-path closure finding from raw readback and replay evidence.
AID-E-004-G001
Verify the incident scope and canonical Harden or Restore fix receipts before closure testing.
Before closure work begins
Start only after every incident-applicable eviction action is complete and each canonical Harden or Restore owner has produced a signed fix receipt. The closure tester receives a signed manifest that enumerates the exact applicable receipt IDs, owner controls, fix-artifact digests, and owner target-population digests. It verifies and normalizes this set but does not apply or modify a fix.
Define the receipt-to-target binding
Each owner receipt must bind its own receipt_id, incident, owner control, immutable fix-artifact digest, and target_population_sha256. The target-population digest is SHA-256 over a compact, key-sorted JSON array, sorted by (target_id, asset_version, exploit_path_id), whose objects contain exactly those three fields. G002 reconstructs that projection from the incident population and rejects any missing, extra, or differently bound receipt.
Verify and normalize the applicable fix set
Use pinned trust roots and a read-only identity. Reject an empty manifest, duplicate receipt ID, unknown owner key, wrong incident, receipt-ID mismatch, mismatched fix or population digest, or invalid signature. The verification timeout is a required positive value injected from the signed incident closure policy; it is not a universal framework constant.
set -euo pipefail
: "${INCIDENT_ID:?required}"
: "${E004_FIX_RECEIPT_VERIFY_TIMEOUT_SECONDS:?required}"
[[ "$E004_FIX_RECEIPT_VERIFY_TIMEOUT_SECONDS" =~ ^[0-9]+([.][0-9]+)?$ ]]
awk -v value="$E004_FIX_RECEIPT_VERIFY_TIMEOUT_SECONDS" 'BEGIN { exit !(value > 0) }'
snapshot_root="$(mktemp -d)"
chmod 700 "$snapshot_root"
trap 'rm -rf -- "$snapshot_root"' EXIT
manifest_source="evidence/applicable-fix-receipts.json"
manifest_bundle_source="${manifest_source}.sigstore.json"
manifest="$snapshot_root/applicable-fix-receipts.json"
manifest_bundle="$snapshot_root/applicable-fix-receipts.json.sigstore.json"
install -m 0400 -- "$manifest_source" "$manifest"
install -m 0400 -- "$manifest_bundle_source" "$manifest_bundle"
timeout "$E004_FIX_RECEIPT_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/closure-scope-authority.pub \
--bundle "$manifest_bundle" "$manifest"
jq -e --arg incident "$INCIDENT_ID" '
type == "object"
and keys == ["incident_id", "receipts", "schema_version"]
and .schema_version == "aidefend.applicable-fix-receipts/v1"
and .incident_id == $incident
and (.receipts | type == "array" and length > 0)
and ((.receipts | map(.receipt_id) | length)
== (.receipts | map(.receipt_id) | unique | length))
and all(.receipts[];
type == "object"
and keys == [
"expected_fix_sha256", "expected_population_sha256",
"owner_control", "receipt_id"
]
and (.receipt_id | test("^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"))
and (.owner_control | test("^AID-[A-Z]+-[0-9]{3}(\\.[0-9]{3})?$"))
and (.expected_fix_sha256 | test("^[0-9a-f]{64}$"))
and .expected_fix_sha256 != ("0" * 64)
and (.expected_population_sha256 | test("^[0-9a-f]{64}$"))
and .expected_population_sha256 != ("0" * 64)
)
' "$manifest" >/dev/null
tmp="$snapshot_root/canonical-fix-receipts.jsonl"
: >"$tmp"
verified_count=0
while IFS=$'\t' read -r receipt_id owner_control fix_sha population_sha; do
receipt_source="evidence/fix-receipts/${receipt_id}.json"
bundle_source="${receipt_source}.sigstore.json"
key="keys/fix-owners/${owner_control}.pub"
receipt="$snapshot_root/${receipt_id}.json"
bundle="$snapshot_root/${receipt_id}.json.sigstore.json"
test -s "$receipt_source"
test -s "$bundle_source"
test -s "$key"
install -m 0400 -- "$receipt_source" "$receipt"
install -m 0400 -- "$bundle_source" "$bundle"
timeout "$E004_FIX_RECEIPT_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key "$key" --bundle "$bundle" "$receipt"
jq -e \
--arg receipt_id "$receipt_id" \
--arg incident "$INCIDENT_ID" \
--arg owner "$owner_control" \
--arg fix "$fix_sha" \
--arg population "$population_sha" '
type == "object"
and .receipt_id == $receipt_id
and .incident_id == $incident
and .owner_control == $owner
and .fix_artifact_sha256 == $fix
and .target_population_sha256 == $population
' "$receipt" >/dev/null
jq -cS \
--arg receipt_id "$receipt_id" \
--arg incident "$INCIDENT_ID" \
--arg owner "$owner_control" \
--arg fix "$fix_sha" \
--arg population "$population_sha" '
{
receipt_id: $receipt_id,
incident_id: $incident,
owner_control: $owner,
fix_artifact_sha256: $fix,
target_population_sha256: $population
}
' "$receipt" >>"$tmp"
verified_count=$((verified_count + 1))
done < <(jq -r '.receipts[] |
[.receipt_id, .owner_control, .expected_fix_sha256,
.expected_population_sha256] | @tsv' "$manifest")
((verified_count > 0))
test ! -e evidence/canonical-fix-receipts.json
jq -n -S --arg incident "$INCIDENT_ID" --slurpfile receipts "$tmp" '
{
schema_version: "aidefend.canonical-fix-receipts/v2",
incident_id: $incident,
receipts: ($receipts | sort_by(.receipt_id))
}
' > evidence/canonical-fix-receipts.json
jq -e '.receipts | type == "array" and length > 0' \
evidence/canonical-fix-receipts.json >/dev/null
timeout "$E004_FIX_RECEIPT_VERIFY_TIMEOUT_SECONDS" \
cosign sign-blob --yes --key env://FIX_RECEIPT_SET_SIGNING_KEY \
--bundle evidence/canonical-fix-receipts.json.sigstore.json \
evidence/canonical-fix-receipts.json
timeout "$E004_FIX_RECEIPT_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/fix-receipt-set-verifier.pub \
--bundle evidence/canonical-fix-receipts.json.sigstore.json \
evidence/canonical-fix-receipts.jsonThe normalized receipt set is a control-local handoff for this incident closure run. It must be regenerated when the incident population or an owner fix changes; it is not a framework-wide evidence catalog.
Bind the replay to the concrete incident path
For a compromised AI dependency or plugin, bind exploit_path_id to the affected component digest and vulnerable invocation path, and require independent readback of the fixed component on every affected target before replay. For a connected-tool authority incident, bind it to the exact tool, action, policy and protected side effect; replay the original unauthorized request plus policy-selected bypass and clean-control cases through the effective tool path. Supply these cases to G002-G005 under the signed incident population. The closure verifier establishes that the known path is closed; the canonical Harden or Restore owner retains credit for applying the fix. Unrelated targets or attack families cannot inherit this relationship.
AID-E-004-G002
Resolve and sign the complete incident-defined affected target and exploit-path population.
Build the exact population from authoritative sources
Use this path when the incident identifies one or more affected deployed targets and exploitable paths that must be proven closed. Expand the signed incident scope through deployment inventory, service discovery, gateway and route registries, workload schedulers, model or artifact registries, and cloud or serverless inventories. Record the exact target, deployed asset version, exploit path, enforcement point, approved readback adapter, owner control, fix receipt, and expected effective-state digest for every row. Dynamic replicas and scaled-to-zero targets require authoritative revision membership rather than a one-time list of running hosts.
Reconcile every owner receipt to its target subset
The builder verifies the signed incident-target source and the canonical receipt set from G001. For each receipt it recomputes the documented target projection over exactly target_id, asset_version, and exploit_path_id. Missing, unused, extra, or owner-mismatched receipts; duplicate target identities; wildcards; unresolved aliases; or an omitted region, route, revision, replica, or enforcement point block population issuance.
set -euo pipefail
: "${E004_CLOSURE_VERIFY_TIMEOUT_SECONDS:?required}"
[[ "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" =~ ^[0-9]+([.][0-9]+)?$ ]]
awk -v value="$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" 'BEGIN { exit !(value > 0) }'
snapshot_root="$(mktemp -d)"
chmod 700 "$snapshot_root"
trap 'rm -rf -- "$snapshot_root"' EXIT
targets="$snapshot_root/incident-targets.jsonl"
targets_bundle="$snapshot_root/incident-targets.jsonl.sigstore.json"
receipts="$snapshot_root/canonical-fix-receipts.json"
receipts_bundle="$snapshot_root/canonical-fix-receipts.json.sigstore.json"
install -m 0400 -- evidence/incident-targets.jsonl "$targets"
install -m 0400 -- evidence/incident-targets.jsonl.sigstore.json "$targets_bundle"
install -m 0400 -- evidence/canonical-fix-receipts.json "$receipts"
install -m 0400 -- evidence/canonical-fix-receipts.json.sigstore.json "$receipts_bundle"
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/incident-scope-authority.pub \
--bundle "$targets_bundle" "$targets"
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/fix-receipt-set-verifier.pub \
--bundle "$receipts_bundle" "$receipts"
E004_INCIDENT_TARGETS_SNAPSHOT="$targets" \
E004_FIX_RECEIPTS_SNAPSHOT="$receipts" \
python closure/build_population.py
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign sign-blob --yes --key env://CLOSURE_SCOPE_SIGNING_KEY \
--bundle evidence/closure-population.json.sigstore.json \
evidence/closure-population.json
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/closure-scope-authority.pub \
--bundle evidence/closure-population.json.sigstore.json \
evidence/closure-population.jsonExecutable affected-population builder
# File: closure/build_population.py
from __future__ import annotations
import hashlib
import json
import os
from collections import defaultdict
from pathlib import Path
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate signed JSON key: {key}")
value[key] = item
return value
def reject_nonfinite(value):
raise ValueError(f"non-finite signed JSON value: {value}")
def strict_json(raw: bytes, label: str):
try:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def require_sha256(value: object, label: str) -> str:
if (
not isinstance(value, str)
or len(value) != 64
or set(value) - set("0123456789abcdef")
or value == "0" * 64
):
raise ValueError(f"{label} must be a nonzero lowercase SHA-256 digest")
return value
expected_incident = os.environ["INCIDENT_ID"]
receipts_raw = Path(os.environ["E004_FIX_RECEIPTS_SNAPSHOT"]).read_bytes()
receipts_document = strict_json(receipts_raw, "canonical fix-receipt set")
receipt_fields = {
"receipt_id", "incident_id", "owner_control",
"fix_artifact_sha256", "target_population_sha256",
}
if (
not isinstance(receipts_document, dict)
or set(receipts_document) != {"schema_version", "incident_id", "receipts"}
or receipts_document["schema_version"]
!= "aidefend.canonical-fix-receipts/v2"
or receipts_document["incident_id"] != expected_incident
or not isinstance(receipts_document["receipts"], list)
or not receipts_document["receipts"]
):
raise ValueError("canonical fix-receipt set is empty or misbound")
receipts = {}
for receipt in receipts_document["receipts"]:
if not isinstance(receipt, dict) or set(receipt) != receipt_fields:
raise ValueError("canonical fix-receipt schema differs")
receipt_id = receipt["receipt_id"]
if (
not isinstance(receipt_id, str)
or not receipt_id
or receipt_id in receipts
or receipt["incident_id"] != expected_incident
or not isinstance(receipt["owner_control"], str)
or not receipt["owner_control"]
):
raise ValueError("canonical fix-receipt identity differs")
require_sha256(receipt["fix_artifact_sha256"], "fix artifact")
require_sha256(receipt["target_population_sha256"], "receipt population")
receipts[receipt_id] = receipt
source = Path(os.environ["E004_INCIDENT_TARGETS_SNAPSHOT"])
target_text = source.read_bytes().decode("utf-8", errors="strict")
target_lines = target_text.splitlines()
if not target_text.endswith("\n") or not target_lines or any(not line for line in target_lines):
raise ValueError("incident-target population must be canonical nonempty JSONL")
rows = [strict_json(line.encode("utf-8"), f"incident target {number}")
for number, line in enumerate(target_lines, 1)]
if not rows:
raise ValueError("affected target/path population is empty")
required = {
"incident_id", "target_id", "asset_version", "exploit_path_id",
"enforcement_point_id", "expected_readback_adapter_id", "owner_control",
"fix_receipt_id", "expected_effective_state_sha256",
}
keys = set()
subsets: dict[str, list[dict[str, str]]] = defaultdict(list)
for row in rows:
if not isinstance(row, dict) or set(row) != required:
raise ValueError("target schema differs")
if row["incident_id"] != expected_incident:
raise ValueError("target incident binding differs")
if any(not isinstance(row[field], str) or not row[field] for field in required):
raise ValueError("target fields must be nonempty strings")
if any("*" in row[field] or "?" in row[field] for field in (
"target_id", "asset_version", "exploit_path_id", "enforcement_point_id"
)):
raise ValueError("target identities must not contain wildcards")
require_sha256(
row["expected_effective_state_sha256"],
"expected effective state",
)
key = (row["target_id"], row["asset_version"], row["exploit_path_id"])
if key in keys:
raise ValueError("duplicate affected target identity")
keys.add(key)
receipt = receipts.get(row["fix_receipt_id"])
if receipt is None or receipt["owner_control"] != row["owner_control"]:
raise ValueError("target fix-receipt or owner binding differs")
subsets[row["fix_receipt_id"]].append({
"target_id": row["target_id"],
"asset_version": row["asset_version"],
"exploit_path_id": row["exploit_path_id"],
})
if set(subsets) != set(receipts):
raise ValueError("canonical receipt set differs from used target receipts")
for receipt_id, subset in subsets.items():
subset.sort(key=lambda row: (
row["target_id"], row["asset_version"], row["exploit_path_id"]
))
observed = hashlib.sha256(canonical(subset)).hexdigest()
if observed != receipts[receipt_id]["target_population_sha256"]:
raise ValueError("receipt target-population digest differs")
rows.sort(key=lambda row: (
row["target_id"], row["asset_version"], row["exploit_path_id"]
))
manifest = {
"schema_version": "aidefend.incident-closure-population.v3",
"incident_id": expected_incident,
"population_count": len(rows),
"population_sha256": hashlib.sha256(canonical(rows)).hexdigest(),
"targets": rows,
}
output = Path("evidence/closure-population.json")
with output.open("x", encoding="utf-8") as handle:
json.dump(manifest, handle, sort_keys=True, separators=(",", ":"))
handle.write("\n")Later exclusions require a newly signed incident scope and regenerated population; deleting a failed row is not a valid closure operation.
AID-E-004-G003
Independently read back effective fix state across every affected target and enforcement point.
Read effective state, not desired state
For every signed target identity, a separately credentialed adapter queries the exact live enforcement point named by G002. Source IaC, a patch ticket, deployment-tool success, or a cached desired-state document is not an effective-state observation. The observation must bind the incident, target, deployed asset version, exploit path, fix receipt, enforcement point, approved adapter, and UTC observation time.
Use only current, signed observations
The maximum observation age is a required positive value injected from the signed incident closure policy. The normalizer verifies the signed population and signed raw observation set before parsing them, rejects future or stale timestamps, missing or extra rows, duplicate identities, unapproved adapters, incomplete readback, and live-state digests that differ from the approved fix.
set -euo pipefail
: "${E004_CLOSURE_VERIFY_TIMEOUT_SECONDS:?required}"
: "${E004_EFFECTIVE_STATE_MAX_AGE_SECONDS:?required}"
snapshot_root="$(mktemp -d)"
chmod 700 "$snapshot_root"
trap 'rm -rf -- "$snapshot_root"' EXIT
population="$snapshot_root/closure-population.json"
population_bundle="$snapshot_root/closure-population.json.sigstore.json"
observations="$snapshot_root/effective-state-observations.jsonl"
observations_bundle="$snapshot_root/effective-state-observations.jsonl.sigstore.json"
install -m 0400 -- evidence/closure-population.json "$population"
install -m 0400 -- evidence/closure-population.json.sigstore.json "$population_bundle"
install -m 0400 -- evidence/effective-state-observations.jsonl "$observations"
install -m 0400 -- evidence/effective-state-observations.jsonl.sigstore.json "$observations_bundle"
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/closure-scope-authority.pub \
--bundle "$population_bundle" "$population"
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/effective-state-observer.pub \
--bundle "$observations_bundle" "$observations"
E004_CLOSURE_POPULATION_SNAPSHOT="$population" \
E004_EFFECTIVE_STATE_OBSERVATIONS_SNAPSHOT="$observations" \
python closure/normalize_effective_state.py
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign sign-blob --yes --key env://EFFECTIVE_STATE_VERIFIER_KEY \
--bundle evidence/effective-state-results.json.sigstore.json \
evidence/effective-state-results.json
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/effective-state-verifier.pub \
--bundle evidence/effective-state-results.json.sigstore.json \
evidence/effective-state-results.jsonExecutable effective-state normalizer
# File: closure/normalize_effective_state.py
from __future__ import annotations
import hashlib
import json
import math
import os
from datetime import datetime, timezone
from pathlib import Path
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate signed JSON key: {key}")
value[key] = item
return value
def reject_nonfinite(value):
raise ValueError(f"non-finite signed JSON value: {value}")
def strict_json(raw: bytes, label: str):
try:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def positive_seconds(name: str) -> float:
raw = os.environ.get(name)
if raw is None or not raw.strip():
raise RuntimeError(f"required closure-policy field is absent: {name}")
value = float(raw)
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"closure-policy field must be finite and positive: {name}")
return value
def parse_utc(value: object) -> datetime:
if not isinstance(value, str) or not value.endswith("Z"):
raise ValueError("observed_at must be an RFC 3339 UTC timestamp")
try:
parsed = datetime.fromisoformat(value[:-1] + "+00:00")
except ValueError as exc:
raise ValueError("observed_at is not a valid RFC 3339 timestamp") from exc
if parsed.tzinfo is None or parsed.utcoffset() != timezone.utc.utcoffset(parsed):
raise ValueError("observed_at must use UTC")
return parsed
max_age = positive_seconds("E004_EFFECTIVE_STATE_MAX_AGE_SECONDS")
now = datetime.now(timezone.utc)
population = strict_json(
Path(os.environ["E004_CLOSURE_POPULATION_SNAPSHOT"]).read_bytes(),
"closure population",
)
targets = population.get("targets") if isinstance(population, dict) else None
if (
not isinstance(population, dict)
or set(population)
!= {"schema_version", "incident_id", "population_count",
"population_sha256", "targets"}
or population["schema_version"]
!= "aidefend.incident-closure-population.v3"
or not isinstance(targets, list)
or not targets
or population["population_count"] != len(targets)
):
raise ValueError("signed closure population is empty or malformed")
expected = {
(row["target_id"], row["asset_version"], row["exploit_path_id"]): row
for row in targets
}
if len(expected) != len(targets):
raise ValueError("duplicate target identity in closure population")
observation_path = Path(os.environ["E004_EFFECTIVE_STATE_OBSERVATIONS_SNAPSHOT"])
observation_bytes = observation_path.read_bytes()
observation_text = observation_bytes.decode("utf-8", errors="strict")
observation_lines = observation_text.splitlines()
if (not observation_text.endswith("\n") or not observation_lines
or any(not line for line in observation_lines)):
raise ValueError("effective-state observations must be canonical nonempty JSONL")
observations = [strict_json(line.encode("utf-8"), f"observation {number}")
for number, line in enumerate(observation_lines, 1)]
if not observations:
raise ValueError("effective-state observation population is empty")
observation_fields = {
"incident_id", "target_id", "asset_version", "exploit_path_id",
"fix_receipt_id", "enforcement_point_id", "adapter_id", "observed_at",
"observation_complete", "effective_state",
}
actual = {}
for row in observations:
if not isinstance(row, dict) or set(row) != observation_fields:
raise ValueError("effective-state observation schema differs")
key = (row["target_id"], row["asset_version"], row["exploit_path_id"])
wanted = expected.get(key)
observed_at = parse_utc(row["observed_at"])
age = (now - observed_at).total_seconds()
if age < 0 or age > max_age:
raise RuntimeError("effective-state observation is future-dated or stale")
if (
key in actual
or wanted is None
or row["incident_id"] != population["incident_id"]
or row["fix_receipt_id"] != wanted["fix_receipt_id"]
or row["enforcement_point_id"] != wanted["enforcement_point_id"]
or row["adapter_id"] != wanted["expected_readback_adapter_id"]
or row["observation_complete"] is not True
or not isinstance(row["effective_state"], dict)
):
raise RuntimeError("effective-state observation is incomplete or misbound")
observed_digest = hashlib.sha256(canonical(row["effective_state"])).hexdigest()
if observed_digest != wanted["expected_effective_state_sha256"]:
raise RuntimeError("live effective state differs from the approved fix")
actual[key] = {
"target_id": key[0],
"asset_version": key[1],
"exploit_path_id": key[2],
"fix_receipt_id": row["fix_receipt_id"],
"enforcement_point_id": row["enforcement_point_id"],
"adapter_id": row["adapter_id"],
"observed_at": row["observed_at"],
"observed_effective_state_sha256": observed_digest,
"observation_complete": True,
"matches_approved_fix": True,
}
if set(actual) != set(expected):
raise RuntimeError("effective-state observation population is incomplete")
results = [actual[key] for key in sorted(actual)]
document = {
"schema_version": "aidefend.effective-state-results/v2",
"incident_id": population["incident_id"],
"population_sha256": population["population_sha256"],
"source_observations_sha256": hashlib.sha256(observation_bytes).hexdigest(),
"results": results,
}
with Path("evidence/effective-state-results.json").open(
"x", encoding="utf-8"
) as handle:
json.dump(document, handle, sort_keys=True, separators=(",", ":"))
handle.write("\n")Unsupported live introspection, an unavailable authority, partial rollout, stale cache, version skew, adapter mismatch, signature failure, or an incomplete or stale readback blocks G008 from issuing closure.
AID-E-004-G004
Reconstruct a safe, immutable replay bundle for the exact exploited path.
Preserve exploit semantics safely
Derive the replay bundle from immutable forensic evidence and bind the exact request bytes, encoding and decoding sequence, modality, protocol order, authentication context, routing preconditions, model or tool inputs, expected vulnerable sink, and prohibited effect. Replace real secrets, personal data, destructive commands, destinations, and payment or identity side effects with signed synthetic fixtures that preserve the exploit's control-flow semantics.
Qualify the harness
Run from an isolated, production-representative vantage point with no production authority. Instrument every relevant process, network, file, data, model, tool, and external-action sink; pre-register the expected path evidence and abort conditions. Prove the harness reaches the intended path with a safe control fixture and sign the bundle, environment, sensor, and adapter digests. If sanitization changes the exploit mechanism or path reachability cannot be demonstrated, record insufficient evidence rather than an artificial denial.
Executable immutable replay bundle
set -euo pipefail
umask 077
jq -S . evidence/closure-population.json > replay/population.canonical.json
jq -S . evidence/exploit-request.json > replay/exploit-request.canonical.json
tar --sort=name --mtime=@0 --owner=0 --group=0 --numeric-owner -cf replay/exploit-replay.tar replay/population.canonical.json replay/exploit-request.canonical.json replay/expected-effect.json replay/runner-lock.txt
sha256sum replay/exploit-replay.tar > replay/exploit-replay.tar.sha256
cosign sign-blob --yes --key env://REPLAY_BUNDLE_KEY --bundle replay/exploit-replay.tar.sigstore.json replay/exploit-replay.tar
cosign verify-blob --key keys/replay-bundle.pub --bundle replay/exploit-replay.tar.sigstore.json replay/exploit-replay.tarRemove live credentials, personal data not required by the reproducer, and destructive side effects. Bind sanitized substitutions to the original evidence by digest and preserve the exact runner/container lock. G005 and G008 verify the bundle signature and recompute its SHA-256 digest against the signed exact-case manifest before using any replay result.
AID-E-004-G005
Replay the exact exploit against every applicable affected path and observe the protected effect.
Execute the complete, incident-bound replay matrix
Use this path only when G004 can reconstruct the exploited request safely in an isolated, production-representative environment with independent protected-sink observation. The exact-case authority signs a manifest containing the incident ID, G002 population digest, G004 replay-bundle digest, explicit HTTPS replay and observer origins, allowed synthetic header names, and exactly one case for every affected target identity. This prevents a valid case list from another incident, population, bundle, or environment from being reused.
Constrain destinations and inputs before execution
The runner verifies the signed bundle, case manifest, and closure population before parsing them. It rejects origins outside the signed allowlists, URL credentials, redirects, unapproved or duplicate header names, invalid methods, oversized bodies, duplicate cases, missing targets, and fix-receipt drift. Request, observation, and body-size bounds are required values injected from the signed replay policy; none is a universal framework default.
Judge the protected effect, not the HTTP status
A case passes only when it reached the intended control path, the independent observation completed after its window closed, and the prohibited protected effect remained absent. HTTP status is retained only as transport diagnostics. Any production authority or destination, loss of isolation, observer failure, route mismatch, or unbounded side effect aborts the run.
# File: closure/replay_exact.py
from __future__ import annotations
import hashlib
import json
import math
import os
import re
import ssl
import subprocess
import tempfile
import uuid
from pathlib import Path
from urllib.parse import urlsplit
import httpx
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate signed JSON key: {key}")
value[key] = item
return value
def reject_nonfinite(value):
raise ValueError(f"non-finite signed JSON value: {value}")
def strict_json(raw: bytes, label: str):
try:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
def verified_signed_bytes(
payload_path: Path,
bundle_path: Path,
trust_key: str | Path,
timeout_seconds: float,
label: str,
) -> bytes:
payload = payload_path.read_bytes()
bundle = bundle_path.read_bytes()
if not payload or not bundle:
raise ValueError(f"{label} payload or bundle is empty")
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
bundle_snapshot = root / "payload.sigstore.json"
payload_snapshot.write_bytes(payload)
bundle_snapshot.write_bytes(bundle)
os.chmod(payload_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run(
["cosign", "verify-blob", "--key", str(trust_key),
"--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True, timeout=timeout_seconds,
)
verified = payload_snapshot.read_bytes()
verified_bundle = bundle_snapshot.read_bytes()
if verified != payload or verified_bundle != bundle:
raise OSError(f"{label} verified payload or bundle snapshot changed")
return verified
def positive_seconds(name: str) -> float:
raw = os.environ.get(name)
if raw is None or not raw.strip():
raise RuntimeError(f"required replay-policy field is absent: {name}")
value = float(raw)
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"replay-policy field must be finite and positive: {name}")
return value
def positive_integer(name: str) -> int:
raw = os.environ.get(name)
if raw is None or not raw.strip() or not raw.isdecimal():
raise RuntimeError(f"required replay-policy integer is absent: {name}")
value = int(raw)
if value <= 0:
raise RuntimeError(f"replay-policy integer must be positive: {name}")
return value
def verify(payload: Path, bundle: Path, key: str) -> bytes:
return verified_signed_bytes(
payload, bundle, key, verify_timeout, str(payload)
)
def require_sha256(value: object, label: str) -> str:
if (
not isinstance(value, str)
or len(value) != 64
or set(value) - set("0123456789abcdef")
or value == "0" * 64
):
raise ValueError(f"{label} must be a nonzero lowercase SHA-256 digest")
return value
def normalized_origin(value: object, label: str) -> str:
if not isinstance(value, str) or not value:
raise ValueError(f"{label} must be a nonempty HTTPS origin")
parsed = urlsplit(value)
if (
parsed.scheme != "https"
or parsed.hostname is None
or parsed.username is not None
or parsed.password is not None
or parsed.path not in ("", "/")
or parsed.query
or parsed.fragment
):
raise ValueError(f"{label} must be an origin without credentials or path")
try:
port = parsed.port or 443
except ValueError as exc:
raise ValueError(f"{label} has an invalid port") from exc
host = parsed.hostname.lower()
authority = f"[{host}]" if ":" in host else host
return f"https://{authority}:{port}"
def request_origin(value: object, label: str) -> str:
if not isinstance(value, str) or not value:
raise ValueError(f"{label} must be a nonempty HTTPS URL")
parsed = urlsplit(value)
if (
parsed.scheme != "https"
or parsed.hostname is None
or parsed.username is not None
or parsed.password is not None
or parsed.fragment
):
raise ValueError(f"{label} is not an allowed HTTPS URL")
try:
port = parsed.port or 443
except ValueError as exc:
raise ValueError(f"{label} has an invalid port") from exc
host = parsed.hostname.lower()
authority = f"[{host}]" if ":" in host else host
return f"https://{authority}:{port}"
request_timeout = positive_seconds("E004_REPLAY_REQUEST_TIMEOUT_SECONDS")
observation_timeout = positive_seconds("E004_EFFECT_OBSERVATION_TIMEOUT_SECONDS")
verify_timeout = positive_seconds("E004_CLOSURE_VERIFY_TIMEOUT_SECONDS")
max_body_bytes = positive_integer("E004_REPLAY_MAX_BODY_BYTES")
max_observation_response_bytes = positive_integer("E004_OBSERVATION_MAX_RESPONSE_BYTES")
observation_response_chunk_bytes = positive_integer("E004_OBSERVATION_RESPONSE_CHUNK_BYTES")
if observation_response_chunk_bytes > max_observation_response_bytes:
raise RuntimeError("signed observation response bounds are invalid")
BUNDLE = Path("replay/exploit-replay.tar")
MANIFEST = Path("replay/exact-cases.json")
POPULATION = Path("evidence/closure-population.json")
RESULTS = Path("evidence/exact-replay.json")
bundle_raw = verify(BUNDLE, Path(str(BUNDLE) + ".sigstore.json"), "keys/replay-bundle.pub")
manifest_raw = verify(MANIFEST, Path(str(MANIFEST) + ".sigstore.json"), "keys/exact-case-authority.pub")
population_raw = verify(POPULATION, Path(str(POPULATION) + ".sigstore.json"),
"keys/closure-scope-authority.pub")
population = strict_json(population_raw, "closure population")
targets = population.get("targets") if isinstance(population, dict) else None
if (
not isinstance(population, dict)
or population.get("schema_version")
!= "aidefend.incident-closure-population.v3"
or not isinstance(targets, list)
or not targets
or population.get("population_count") != len(targets)
):
raise ValueError("signed affected population is empty or malformed")
expected = {
(row["target_id"], row["asset_version"], row["exploit_path_id"]): row
for row in targets
}
if len(expected) != len(targets):
raise ValueError("duplicate affected target identity")
manifest = strict_json(manifest_raw, "exact-case manifest")
manifest_fields = {
"schema_version", "incident_id", "population_sha256",
"replay_bundle_sha256", "allowed_replay_origins",
"allowed_observer_origins", "allowed_header_names", "cases",
}
case_fields = {
"case_id", "target_id", "asset_version", "exploit_path_id",
"fix_receipt_id", "method", "url", "safe_headers", "body",
"effect_observer_url",
}
if (
not isinstance(manifest, dict)
or set(manifest) != manifest_fields
or manifest["schema_version"] != "aidefend.exact-replay-cases/v2"
or manifest["incident_id"] != population["incident_id"]
or manifest["population_sha256"] != population["population_sha256"]
or not isinstance(manifest["cases"], list)
or not manifest["cases"]
):
raise ValueError("signed exact-case manifest is empty or misbound")
bundle_sha256 = hashlib.sha256(bundle_raw).hexdigest()
if require_sha256(manifest["replay_bundle_sha256"], "replay bundle") != bundle_sha256:
raise ValueError("exact-case manifest does not bind the verified replay bundle")
def load_origins(field: str) -> set[str]:
values = manifest[field]
if not isinstance(values, list) or not values:
raise ValueError(f"{field} must be a nonempty origin list")
normalized = [normalized_origin(value, field) for value in values]
if normalized != values or len(set(normalized)) != len(normalized):
raise ValueError(f"{field} must contain sorted unique normalized origins")
return set(normalized)
replay_origins = load_origins("allowed_replay_origins")
observer_origins = load_origins("allowed_observer_origins")
allowed_headers = manifest["allowed_header_names"]
if (
not isinstance(allowed_headers, list)
or allowed_headers != sorted(set(allowed_headers))
or any(
not isinstance(name, str)
or not name
or name != name.lower()
or any(not character.isalnum() and character not in "-_" for character in name)
for name in allowed_headers
)
or "x-aidefend-replay-id" in allowed_headers
):
raise ValueError("allowed_header_names must be sorted unique lowercase names")
allowed_header_set = set(allowed_headers)
case_id_pattern = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
seen, case_ids, prepared_cases = set(), set(), []
for case in manifest["cases"]:
if not isinstance(case, dict) or set(case) != case_fields:
raise ValueError("exact replay case schema differs")
identity = (case["target_id"], case["asset_version"], case["exploit_path_id"])
wanted = expected.get(identity)
case_id = case["case_id"]
method = case["method"]
headers = case["safe_headers"]
if (
identity in seen
or wanted is None
or not isinstance(case_id, str)
or case_id_pattern.fullmatch(case_id) is None
or case_id in case_ids
or case["fix_receipt_id"] != wanted["fix_receipt_id"]
or not isinstance(method, str)
or re.fullmatch(r"[A-Z]+", method) is None
or not isinstance(headers, dict)
or not isinstance(case["body"], str)
or len(case["body"].encode("utf-8")) > max_body_bytes
or request_origin(case["url"], "replay URL") not in replay_origins
or request_origin(case["effect_observer_url"], "observer URL")
not in observer_origins
):
raise ValueError("exact replay identity, input, or fix binding differs")
normalized_names = [str(name).lower() for name in headers]
if (
len(set(normalized_names)) != len(normalized_names)
or set(normalized_names) - allowed_header_set
or any(
not isinstance(name, str)
or not isinstance(value, str)
or "\r" in value
or "\n" in value
for name, value in headers.items()
)
):
raise ValueError("replay headers are duplicated, unapproved, or unsafe")
seen.add(identity)
case_ids.add(case_id)
prepared_cases.append(case)
if seen != set(expected):
raise RuntimeError("exact replay population differs from affected population")
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate observer JSON key: {key}")
value[key] = item
return value
def reject_nonfinite(value):
raise ValueError(f"non-finite observer JSON is forbidden: {value}")
def read_observer_json(response: httpx.Response) -> dict:
response.raise_for_status()
chunks, observed = [], 0
for chunk in response.iter_bytes(observation_response_chunk_bytes):
observed += len(chunk)
if observed > max_observation_response_bytes:
raise RuntimeError("observer response exceeds signed bound")
chunks.append(chunk)
value = json.loads(b"".join(chunks).decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=reject_nonfinite)
if not isinstance(value, dict):
raise RuntimeError("observer response is not an object")
return value
replay_tls = ssl.create_default_context(cafile="keys/closure-ca.pem")
replay_tls.load_cert_chain("keys/replay-client.pem", "keys/replay-client-key.pem")
observer_tls = ssl.create_default_context(cafile="keys/effect-observer-ca.pem")
observer_tls.load_cert_chain("keys/effect-observer-client.pem", "keys/effect-observer-client-key.pem")
results = []
replay_ids = set()
with (
httpx.Client(
verify=replay_tls,
timeout=request_timeout,
follow_redirects=False,
) as replay_client,
httpx.Client(
verify=observer_tls,
timeout=observation_timeout,
follow_redirects=False,
) as observer,
):
for case in prepared_cases:
replay_id = str(uuid.uuid4())
if replay_id in replay_ids:
raise RuntimeError("duplicate generated replay ID")
replay_ids.add(replay_id)
headers = dict(case["safe_headers"])
headers["X-AIDEFEND-Replay-ID"] = replay_id
with replay_client.stream(
case["method"], case["url"],
headers=headers, content=case["body"],
) as response:
status_code = response.status_code
with observer.stream(
"GET", case["effect_observer_url"],
params={"replay_id": replay_id, "wait_for_window": "true"},
headers={"Accept-Encoding": "identity"},
) as observed_response:
observed = read_observer_json(observed_response)
expected_fields = {
"replay_id", "case_id", "target_id", "asset_version",
"exploit_path_id", "control_path_reached", "observation_complete",
"observation_window_closed", "prohibited_effect_observed",
}
if not isinstance(observed, dict) or set(observed) != expected_fields:
raise RuntimeError("protected-effect observation schema differs")
binding = {
"replay_id": replay_id,
"case_id": case["case_id"],
"target_id": case["target_id"],
"asset_version": case["asset_version"],
"exploit_path_id": case["exploit_path_id"],
}
if any(observed[field] != value for field, value in binding.items()):
raise RuntimeError("protected-effect observation binding differs")
passed = (
observed["control_path_reached"] is True
and observed["observation_complete"] is True
and observed["observation_window_closed"] is True
and observed["prohibited_effect_observed"] is False
)
results.append({
**binding,
"fix_receipt_id": case["fix_receipt_id"],
"status_code": status_code,
"control_path_reached": observed["control_path_reached"],
"observation_complete": observed["observation_complete"],
"observation_window_closed": observed["observation_window_closed"],
"prohibited_effect_observed": observed["prohibited_effect_observed"],
"outcome": "PASS" if passed else "FAIL",
})
document = {
"schema_version": "aidefend.exact-replay-results/v2",
"incident_id": population["incident_id"],
"population_sha256": population["population_sha256"],
"replay_bundle_sha256": bundle_sha256,
"results": results,
}
with RESULTS.open("x", encoding="utf-8") as handle:
json.dump(document, handle, sort_keys=True, separators=(",", ":"))
handle.write("\n")
subprocess.run(
["cosign", "sign-blob", "--yes", "--key",
os.environ["E004_EXACT_REPLAY_RESULTS_SIGNING_KEY"],
"--bundle", str(RESULTS) + ".sigstore.json", str(RESULTS)],
check=True, timeout=verify_timeout,
)
verify(RESULTS, Path(str(RESULTS) + ".sigstore.json"),
"keys/exact-replay-verifier.pub")
if any(row["outcome"] != "PASS" for row in results):
raise SystemExit("exact exploit path remains open or unobserved")Preserve the signed case manifest, replay bundle, exact result document, and raw request, response, control-path, and protected-sink records for G008. A transport denial without protected-sink observation is never closure proof.
AID-E-004-G006
Replay policy-defined bypass, negative, stale-state, and partial-rollout cases without expanding the fix scope.
Pre-register the incident-bounded bypass population
The closure-scope authority signs replay/bypass-case-manifest.json with the incident ID, G002 population digest, and every case ID, target ID, asset version, exploit-path ID, and pinned pytest node ID. Include applicable alternate encodings, route aliases, stale-state, partial-rollout, retry, concurrency, and dependency-outage cases plus nearby non-exploit controls. Every affected target/path needs at least one declared case; later exclusions require a newly signed manifest.
{
"schema_version": "aidefend.bypass-case-manifest/v2",
"incident_id": "INC-2026-0718-0042",
"population_sha256": "c61e73c8f22a997daf0f9a4acbb7e0c45bc7184a2b81cb184034a3e879b330e2",
"cases": [
{
"case_id": "alternate-route-target-a",
"target_id": "target-a",
"asset_version": "release-sha256:9e7d3a4c1b28f506d87211c685f59ac73851ee3c309f82de4c76f0ab1d29e546",
"exploit_path_id": "path-a",
"pytest_nodeid": "closure_tests/test_bypass.py::test_alternate_route_target_a"
}
]
}Execute every signed case and retain sink observations
Each test receives only its fixed result path and must emit a strict result bound to the case, target, and exploit path. A PASS requires the intended control path to be reached, the observation to complete, and the prohibited effect to remain absent. The runner copies the verified manifest and signature into evidence, writes the exact result population, signs it, and returns nonzero after preserving any failed or unobserved case.
# File: closure/run_bypass_population.py
from __future__ import annotations
import json, math, os, re, shutil, subprocess
import tempfile
from pathlib import Path
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate signed JSON key: {key}")
value[key] = item
return value
def reject_nonfinite(value):
raise ValueError(f"non-finite signed JSON value: {value}")
def strict_json(raw: bytes, label: str):
try:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
def verified_signed_artifact(
payload_path: Path,
bundle_path: Path,
trust_key: str | Path,
timeout_seconds: float,
label: str,
) -> tuple[bytes, bytes]:
payload = payload_path.read_bytes()
bundle = bundle_path.read_bytes()
if not payload or not bundle:
raise ValueError(f"{label} payload or bundle is empty")
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
bundle_snapshot = root / "payload.sigstore.json"
payload_snapshot.write_bytes(payload)
bundle_snapshot.write_bytes(bundle)
os.chmod(payload_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run(
["cosign", "verify-blob", "--key", str(trust_key),
"--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True, timeout=timeout_seconds,
)
verified = payload_snapshot.read_bytes()
verified_bundle = bundle_snapshot.read_bytes()
if verified != payload or verified_bundle != bundle:
raise OSError(f"{label} verified payload or bundle snapshot changed")
return verified, verified_bundle
SOURCE_MANIFEST = Path("replay/bypass-case-manifest.json")
SOURCE_BUNDLE = Path("replay/bypass-case-manifest.json.sigstore.json")
POPULATION = Path("evidence/closure-population.json")
EVIDENCE_MANIFEST = Path("evidence/bypass-case-manifest.json")
EVIDENCE_BUNDLE = Path("evidence/bypass-case-manifest.json.sigstore.json")
RESULTS = Path("evidence/bypass-results.jsonl")
RESULT_DIR = Path("evidence/bypass-case-results")
CASE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
timeout = float(os.environ["E004_BYPASS_CASE_TIMEOUT_SECONDS"])
if not math.isfinite(timeout) or timeout <= 0:
raise RuntimeError("signed bypass-case timeout must be finite and positive")
def verify(payload: Path, bundle: Path, key: str) -> tuple[bytes, bytes]:
return verified_signed_artifact(payload, bundle, key, timeout, str(payload))
manifest_raw, manifest_bundle_raw = verify(SOURCE_MANIFEST, SOURCE_BUNDLE, "keys/bypass-case-authority.pub")
population_raw, _population_bundle_raw = verify(POPULATION, Path(str(POPULATION) + ".sigstore.json"),
"keys/closure-scope-authority.pub")
manifest = strict_json(manifest_raw, "bypass-case manifest")
population = strict_json(population_raw, "closure population")
targets = population.get("targets") if isinstance(population, dict) else None
if (
population.get("schema_version")
!= "aidefend.incident-closure-population.v3"
or not isinstance(targets, list)
or not targets
or population.get("population_count") != len(targets)
):
raise ValueError("signed closure population is empty or malformed")
manifest_fields = {"schema_version", "incident_id", "population_sha256", "cases"}
case_fields = {"case_id", "target_id", "asset_version", "exploit_path_id", "pytest_nodeid"}
if (
not isinstance(manifest, dict)
or set(manifest) != manifest_fields
or manifest["schema_version"] != "aidefend.bypass-case-manifest/v2"
or manifest["incident_id"] != population.get("incident_id")
or manifest["population_sha256"] != population.get("population_sha256")
or not isinstance(manifest["cases"], list)
or not manifest["cases"]
):
raise ValueError("signed bypass-case manifest is empty or misbound")
target_pairs = {
(row["target_id"], row["asset_version"], row["exploit_path_id"])
for row in targets
}
if not target_pairs:
raise ValueError("affected target/path population is empty")
case_ids, nodeids, case_pairs = set(), set(), set()
for case in manifest["cases"]:
if not isinstance(case, dict) or set(case) != case_fields:
raise ValueError("bypass case schema differs")
case_id = case["case_id"]
nodeid = case["pytest_nodeid"]
pair = (case["target_id"], case["asset_version"], case["exploit_path_id"])
if (
not isinstance(case_id, str)
or CASE_ID.fullmatch(case_id) is None
or case_id in case_ids
or not isinstance(nodeid, str)
or not nodeid.startswith("closure_tests/")
or "::" not in nodeid
or ".." in nodeid
or any(token in nodeid for token in ("*", "?", "[", "]"))
or nodeid in nodeids
or pair not in target_pairs
):
raise ValueError("bypass case identity, node ID, or target binding is invalid")
case_ids.add(case_id)
nodeids.add(nodeid)
case_pairs.add(pair)
if case_pairs != target_pairs:
raise ValueError("bypass manifest omits an affected target/path")
with EVIDENCE_MANIFEST.open("xb") as handle:
handle.write(manifest_raw)
with EVIDENCE_BUNDLE.open("xb") as handle:
handle.write(manifest_bundle_raw)
RESULT_DIR.mkdir(mode=0o700)
rows = []
result_fields = {
"case_id", "target_id", "asset_version", "exploit_path_id", "control_path_reached",
"observation_complete", "prohibited_effect_observed", "outcome",
}
for case in manifest["cases"]:
result_path = RESULT_DIR / f"{case['case_id']}.json"
environment = dict(os.environ)
environment["AIDEFEND_BYPASS_RESULT_PATH"] = str(result_path)
completed = subprocess.run(
["pytest", "-q", case["pytest_nodeid"]],
check=False, env=environment, timeout=timeout,
)
if not result_path.is_file():
raise RuntimeError(f"bypass case emitted no result: {case['case_id']}")
result = strict_json(result_path.read_bytes(), f"bypass result {case['case_id']}")
if not isinstance(result, dict) or set(result) != result_fields:
raise RuntimeError("bypass result schema differs")
for field in ("case_id", "target_id", "asset_version", "exploit_path_id"):
if result[field] != case[field]:
raise RuntimeError("bypass result binding differs")
if completed.returncode == 0 and result["outcome"] != "PASS":
raise RuntimeError("pytest success conflicts with bypass result")
if completed.returncode != 0 and result["outcome"] == "PASS":
raise RuntimeError("pytest failure conflicts with bypass result")
rows.append(result)
with RESULTS.open("x", encoding="utf-8") as handle:
for row in rows:
handle.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n")
subprocess.run(
["cosign", "sign-blob", "--yes", "--key",
os.environ["E004_BYPASS_RESULTS_SIGNING_KEY"],
"--bundle", str(RESULTS) + ".sigstore.json", str(RESULTS)],
check=True, timeout=timeout,
)
verify(RESULTS, Path(str(RESULTS) + ".sigstore.json"),
"keys/bypass-results-verifier.pub")
if any(
row["outcome"] != "PASS"
or row["control_path_reached"] is not True
or row["observation_complete"] is not True
or row["prohibited_effect_observed"] is not False
for row in rows
):
raise SystemExit("one or more bypass cases failed or were unobserved")The verifier in G008 re-verifies both signatures and requires the result identity set to equal the manifest exactly. Do not append ad hoc cases after seeing results or treat a pytest exit code without a protected-sink observation as proof of closure.
AID-E-004-G007
Run clean functional, authorization, and resilience controls against the fixed affected population.
Pre-register the clean-control population
The closure-scope authority signs replay/clean-case-manifest.json before testing. It binds the incident and G002 population digest to every clean case's stable case ID, target, asset version, exploit path, fix receipt, exact pytest node ID, and expected JUnit (classname, name). Include at least one clean functional, authorization, or resilience case for every affected target identity. Later additions or exclusions require a newly signed manifest.
Execute only the signed cases
The runner verifies the manifest and closure population before invoking pytest. It rejects duplicate IDs, node IDs, or JUnit identities; directory and glob selectors; missing target coverage; and fix-receipt drift. The signed expected JUnit identity set, not an aggregate test count, defines completeness.
# File: closure/run_clean_population.py
from __future__ import annotations
import json
import math
import os
import re
import subprocess
import tempfile
import xml.etree.ElementTree as ET
from pathlib import Path
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate signed JSON key: {key}")
value[key] = item
return value
def reject_nonfinite(value):
raise ValueError(f"non-finite signed JSON value: {value}")
def strict_json(raw: bytes, label: str):
try:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
def verified_signed_artifact(
payload_path: Path,
bundle_path: Path,
trust_key: str | Path,
timeout_seconds: float,
label: str,
) -> tuple[bytes, bytes]:
payload = payload_path.read_bytes()
bundle = bundle_path.read_bytes()
if not payload or not bundle:
raise ValueError(f"{label} payload or bundle is empty")
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
bundle_snapshot = root / "payload.sigstore.json"
payload_snapshot.write_bytes(payload)
bundle_snapshot.write_bytes(bundle)
os.chmod(payload_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run(
["cosign", "verify-blob", "--key", str(trust_key),
"--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True, timeout=timeout_seconds,
)
verified = payload_snapshot.read_bytes()
verified_bundle = bundle_snapshot.read_bytes()
if verified != payload or verified_bundle != bundle:
raise OSError(f"{label} verified payload or bundle snapshot changed")
return verified, verified_bundle
SOURCE_MANIFEST = Path("replay/clean-case-manifest.json")
SOURCE_BUNDLE = Path("replay/clean-case-manifest.json.sigstore.json")
POPULATION = Path("evidence/closure-population.json")
EVIDENCE_MANIFEST = Path("evidence/clean-case-manifest.json")
EVIDENCE_BUNDLE = Path("evidence/clean-case-manifest.json.sigstore.json")
JUNIT = Path("evidence/clean-regression.xml")
CASE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
def positive_seconds(name: str) -> float:
raw = os.environ.get(name)
if raw is None or not raw.strip():
raise RuntimeError(f"required clean-case policy field is absent: {name}")
value = float(raw)
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"clean-case policy field must be finite and positive: {name}")
return value
timeout = positive_seconds("E004_CLEAN_CASE_TIMEOUT_SECONDS")
def verify(payload: Path, bundle: Path, key: str) -> tuple[bytes, bytes]:
return verified_signed_artifact(payload, bundle, key, timeout, str(payload))
manifest_raw, manifest_bundle_raw = verify(SOURCE_MANIFEST, SOURCE_BUNDLE, "keys/clean-case-authority.pub")
population_raw, _population_bundle_raw = verify(
POPULATION,
Path(str(POPULATION) + ".sigstore.json"),
"keys/closure-scope-authority.pub",
)
manifest = strict_json(manifest_raw, "clean-case manifest")
population = strict_json(population_raw, "closure population")
manifest_fields = {"schema_version", "incident_id", "population_sha256", "cases"}
case_fields = {
"case_id", "target_id", "asset_version", "exploit_path_id",
"fix_receipt_id", "pytest_nodeid", "junit_classname", "junit_name",
}
if (
not isinstance(manifest, dict)
or set(manifest) != manifest_fields
or manifest["schema_version"] != "aidefend.clean-case-manifest/v1"
or manifest["incident_id"] != population.get("incident_id")
or manifest["population_sha256"] != population.get("population_sha256")
or not isinstance(manifest["cases"], list)
or not manifest["cases"]
):
raise ValueError("signed clean-case manifest is empty or misbound")
targets = population.get("targets")
if (
population.get("schema_version")
!= "aidefend.incident-closure-population.v3"
or not isinstance(targets, list)
or not targets
or population.get("population_count") != len(targets)
):
raise ValueError("signed closure population is empty or malformed")
expected = {
(row["target_id"], row["asset_version"], row["exploit_path_id"]): row
for row in targets
}
if len(expected) != len(targets):
raise ValueError("duplicate affected target identity")
case_ids: set[str] = set()
nodeids: set[str] = set()
junit_ids: set[tuple[str, str]] = set()
case_pairs: set[tuple[str, str, str]] = set()
for case in manifest["cases"]:
if not isinstance(case, dict) or set(case) != case_fields:
raise ValueError("clean case schema differs")
pair = (case["target_id"], case["asset_version"], case["exploit_path_id"])
wanted = expected.get(pair)
case_id = case["case_id"]
nodeid = case["pytest_nodeid"]
junit_id = (case["junit_classname"], case["junit_name"])
if (
not isinstance(case_id, str)
or CASE_ID.fullmatch(case_id) is None
or case_id in case_ids
or not isinstance(nodeid, str)
or not nodeid.startswith("closure_tests/")
or "::" not in nodeid
or ".." in nodeid
or any(token in nodeid for token in ("*", "?", "[", "]"))
or nodeid in nodeids
or any(not isinstance(value, str) or not value for value in junit_id)
or junit_id in junit_ids
or wanted is None
or case["fix_receipt_id"] != wanted["fix_receipt_id"]
):
raise ValueError("clean case identity, selector, or target binding is invalid")
case_ids.add(case_id)
nodeids.add(nodeid)
junit_ids.add(junit_id)
case_pairs.add(pair)
if case_pairs != set(expected):
raise ValueError("clean-case manifest omits an affected target identity")
with EVIDENCE_MANIFEST.open("xb") as handle:
handle.write(manifest_raw)
with EVIDENCE_BUNDLE.open("xb") as handle:
handle.write(manifest_bundle_raw)
completed = subprocess.run(
["pytest", "-q", *[case["pytest_nodeid"] for case in manifest["cases"]],
"--junitxml=" + str(JUNIT)],
check=False,
timeout=timeout,
)
if not JUNIT.is_file() or JUNIT.stat().st_size == 0:
raise RuntimeError("pytest emitted no clean-control JUnit result")
root = ET.parse(JUNIT).getroot()
testcases = list(root.iter("testcase"))
observed_ids = [
(case.get("classname", ""), case.get("name", "")) for case in testcases
]
if (
completed.returncode != 0
or not testcases
or len(observed_ids) != len(set(observed_ids))
or set(observed_ids) != junit_ids
or any(
case.find("failure") is not None
or case.find("error") is not None
or case.find("skipped") is not None
for case in testcases
)
):
raise RuntimeError("clean-control JUnit population is incomplete, failed, or skipped")
subprocess.run(
[
"cosign", "sign-blob", "--yes", "--key",
os.environ["E004_CLEAN_RESULTS_SIGNING_KEY"],
"--bundle", str(JUNIT) + ".sigstore.json", str(JUNIT),
],
check=True,
timeout=timeout,
)
verify(JUNIT, Path(str(JUNIT) + ".sigstore.json"), "keys/clean-results-verifier.pub")Use the exact result population
G008 re-verifies the copied manifest and JUnit signatures and requires the observed (classname, name) set to equal the signed expected set. A missing, extra, duplicate, failed, errored, or skipped case blocks closure. Clean success cannot excuse an exact or bypass replay failure.
AID-E-004-G008
Compute and sign an independent exploit-path closure finding from raw readback and replay evidence.
Recompute the incident closure result independently
Use this final verifier only after G001 through G007 have produced their incident-bound artifacts. A verifier that cannot apply the fix or execute the replay first verifies every input signature, then independently recomputes the population digest, receipt-to-target subsets, exact asset-version identities, effective-state joins, replay-case and result populations, protected-sink predicates, and clean JUnit identities. It never inherits a deployment, scanner, test-runner, or incident-owner status field.
Keep the result local to the measured exploit path
A PASS means that the canonical fix receipts cover the complete signed incident target population, the expected fix is live at every declared enforcement point, every exact and bypass case reached an observed protected path without the prohibited effect, and every signed clean control passed without omission or skip. It closes only that incident-defined target/version/path population. It is not a product certification, a generic applicability record, or a framework-wide evidence score.
Executable independent closure verifier
# File: closure/verdict.py
from __future__ import annotations
import hashlib
import json
import math
import os
import subprocess
import tempfile
import xml.etree.ElementTree as ET
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate signed JSON key: {key}")
value[key] = item
return value
def reject_nonfinite(value):
raise ValueError(f"non-finite signed JSON value: {value}")
def strict_json(raw: bytes, label: str):
try:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
def verified_signed_bytes(
payload_path: Path,
bundle_path: Path,
trust_key: str | Path,
timeout_seconds: float,
label: str,
) -> bytes:
payload = payload_path.read_bytes()
bundle = bundle_path.read_bytes()
if not payload or not bundle:
raise ValueError(f"{label} payload or bundle is empty")
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
bundle_snapshot = root / "payload.sigstore.json"
payload_snapshot.write_bytes(payload)
bundle_snapshot.write_bytes(bundle)
os.chmod(payload_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run(
["cosign", "verify-blob", "--key", str(trust_key),
"--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True, timeout=timeout_seconds,
)
verified = payload_snapshot.read_bytes()
verified_bundle = bundle_snapshot.read_bytes()
if verified != payload or verified_bundle != bundle:
raise OSError(f"{label} verified payload or bundle snapshot changed")
return verified
def positive_seconds(name: str) -> float:
raw = os.environ.get(name)
if raw is None or not raw.strip():
raise RuntimeError(f"required closure-policy field is absent: {name}")
value = float(raw)
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"closure-policy field must be finite and positive: {name}")
return value
verify_timeout = positive_seconds("E004_CLOSURE_VERIFY_TIMEOUT_SECONDS")
max_effective_age = positive_seconds("E004_EFFECTIVE_STATE_MAX_AGE_SECONDS")
def verify(payload: Path, bundle: Path, key: str) -> bytes:
return verified_signed_bytes(
payload, bundle, key, verify_timeout, str(payload)
)
signed_inputs = [
(Path("replay/exploit-replay.tar"),
Path("replay/exploit-replay.tar.sigstore.json"), "keys/replay-bundle.pub"),
(Path("replay/exact-cases.json"),
Path("replay/exact-cases.json.sigstore.json"), "keys/exact-case-authority.pub"),
(Path("evidence/canonical-fix-receipts.json"),
Path("evidence/canonical-fix-receipts.json.sigstore.json"),
"keys/fix-receipt-set-verifier.pub"),
(Path("evidence/closure-population.json"),
Path("evidence/closure-population.json.sigstore.json"),
"keys/closure-scope-authority.pub"),
(Path("evidence/effective-state-results.json"),
Path("evidence/effective-state-results.json.sigstore.json"),
"keys/effective-state-verifier.pub"),
(Path("evidence/exact-replay.json"),
Path("evidence/exact-replay.json.sigstore.json"),
"keys/exact-replay-verifier.pub"),
(Path("evidence/bypass-case-manifest.json"),
Path("evidence/bypass-case-manifest.json.sigstore.json"),
"keys/bypass-case-authority.pub"),
(Path("evidence/bypass-results.jsonl"),
Path("evidence/bypass-results.jsonl.sigstore.json"),
"keys/bypass-results-verifier.pub"),
(Path("evidence/clean-case-manifest.json"),
Path("evidence/clean-case-manifest.json.sigstore.json"),
"keys/clean-case-authority.pub"),
(Path("evidence/clean-regression.xml"),
Path("evidence/clean-regression.xml.sigstore.json"),
"keys/clean-results-verifier.pub"),
]
verified_inputs = {
payload: verify(payload, bundle, key)
for payload, bundle, key in signed_inputs
}
def input_bytes(path: Path) -> bytes:
try:
return verified_inputs[path]
except KeyError as exc:
raise ValueError(f"input was not Cosign-bundle verified: {path}") from exc
def read_json(path: Path):
return strict_json(input_bytes(path), str(path))
def read_jsonl(path: Path) -> list[dict]:
text = input_bytes(path).decode("utf-8", errors="strict")
lines = text.splitlines()
if not text.endswith("\n") or not lines or any(not line for line in lines):
raise ValueError(f"{path}: canonical nonempty JSONL population required")
return [strict_json(line.encode("utf-8"), f"{path}:{number}")
for number, line in enumerate(lines, 1)]
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def require_sha256(value: object, label: str) -> str:
if (
not isinstance(value, str)
or len(value) != 64
or set(value) - set("0123456789abcdef")
or value == "0" * 64
):
raise ValueError(f"{label} must be a nonzero lowercase SHA-256 digest")
return value
def require_text(value: object, label: str) -> str:
if not isinstance(value, str) or not value:
raise ValueError(f"{label} must be a nonempty string")
return value
def parse_utc(value: object) -> datetime:
text = require_text(value, "observed_at")
if not text.endswith("Z"):
raise ValueError("observed_at must be an RFC 3339 UTC timestamp")
try:
parsed = datetime.fromisoformat(text[:-1] + "+00:00")
except ValueError as exc:
raise ValueError("observed_at is not a valid RFC 3339 timestamp") from exc
if parsed.tzinfo is None or parsed.utcoffset() != timezone.utc.utcoffset(parsed):
raise ValueError("observed_at must use UTC")
return parsed
population_path = Path("evidence/closure-population.json")
population = read_json(population_path)
population_fields = {"schema_version", "incident_id", "population_count", "population_sha256", "targets"}
target_fields = {
"incident_id", "target_id", "asset_version", "exploit_path_id",
"enforcement_point_id", "expected_readback_adapter_id", "owner_control",
"fix_receipt_id", "expected_effective_state_sha256",
}
if (
not isinstance(population, dict)
or set(population) != population_fields
or population["schema_version"] != "aidefend.incident-closure-population.v3"
or not isinstance(population["targets"], list)
or not population["targets"]
or population["population_count"] != len(population["targets"])
):
raise ValueError("closure population is empty or malformed")
incident_id = require_text(population["incident_id"], "incident_id")
targets = population["targets"]
sorted_targets = sorted(targets, key=lambda row: (
row.get("target_id", ""), row.get("asset_version", ""), row.get("exploit_path_id", "")
))
if targets != sorted_targets:
raise ValueError("closure population is not in canonical identity order")
expected = {}
for row in targets:
if not isinstance(row, dict) or set(row) != target_fields:
raise ValueError("closure target schema differs")
if row["incident_id"] != incident_id:
raise ValueError("closure target incident binding differs")
for field in target_fields:
require_text(row[field], f"target.{field}")
require_sha256(row["expected_effective_state_sha256"], "expected effective state")
key = (row["target_id"], row["asset_version"], row["exploit_path_id"])
if key in expected:
raise ValueError("duplicate closure target identity")
expected[key] = row
computed_population_sha256 = hashlib.sha256(canonical(targets)).hexdigest()
if require_sha256(population["population_sha256"], "population") != computed_population_sha256:
raise ValueError("closure population digest differs")
receipts_document = read_json(Path("evidence/canonical-fix-receipts.json"))
receipt_fields = {"receipt_id", "incident_id", "owner_control", "fix_artifact_sha256", "target_population_sha256"}
if (
not isinstance(receipts_document, dict)
or set(receipts_document) != {"schema_version", "incident_id", "receipts"}
or receipts_document["schema_version"] != "aidefend.canonical-fix-receipts/v2"
or receipts_document["incident_id"] != incident_id
or not isinstance(receipts_document["receipts"], list)
or not receipts_document["receipts"]
):
raise ValueError("canonical fix-receipt set is empty or misbound")
receipts = {}
for receipt in receipts_document["receipts"]:
if not isinstance(receipt, dict) or set(receipt) != receipt_fields:
raise ValueError("canonical fix-receipt schema differs")
receipt_id = require_text(receipt["receipt_id"], "receipt_id")
if receipt_id in receipts or receipt["incident_id"] != incident_id:
raise ValueError("duplicate or cross-incident fix receipt")
require_text(receipt["owner_control"], "owner_control")
require_sha256(receipt["fix_artifact_sha256"], "fix artifact")
require_sha256(receipt["target_population_sha256"], "receipt population")
receipts[receipt_id] = receipt
subsets: dict[str, list[dict[str, str]]] = defaultdict(list)
for key, target in expected.items():
receipt = receipts.get(target["fix_receipt_id"])
if receipt is None or receipt["owner_control"] != target["owner_control"]:
raise ValueError("target fix-receipt or owner binding differs")
subsets[target["fix_receipt_id"]].append({
"target_id": key[0], "asset_version": key[1], "exploit_path_id": key[2],
})
if set(subsets) != set(receipts):
raise ValueError("canonical receipt set differs from used target receipts")
for receipt_id, subset in subsets.items():
subset.sort(key=lambda row: (row["target_id"], row["asset_version"], row["exploit_path_id"]))
if hashlib.sha256(canonical(subset)).hexdigest() != receipts[receipt_id]["target_population_sha256"]:
raise ValueError("receipt target-population digest differs")
effective_document = read_json(Path("evidence/effective-state-results.json"))
effective_fields = {"schema_version", "incident_id", "population_sha256", "source_observations_sha256", "results"}
effective_result_fields = {
"target_id", "asset_version", "exploit_path_id", "fix_receipt_id",
"enforcement_point_id", "adapter_id", "observed_at",
"observed_effective_state_sha256", "observation_complete", "matches_approved_fix",
}
if (
not isinstance(effective_document, dict)
or set(effective_document) != effective_fields
or effective_document["schema_version"] != "aidefend.effective-state-results/v2"
or effective_document["incident_id"] != incident_id
or effective_document["population_sha256"] != computed_population_sha256
or not isinstance(effective_document["results"], list)
or not effective_document["results"]
):
raise ValueError("effective-state result document is empty or misbound")
require_sha256(effective_document["source_observations_sha256"], "source observations")
effective = {}
now = datetime.now(timezone.utc)
for row in effective_document["results"]:
if not isinstance(row, dict) or set(row) != effective_result_fields:
raise ValueError("effective-state result schema differs")
key = (row["target_id"], row["asset_version"], row["exploit_path_id"])
target = expected.get(key)
age = (now - parse_utc(row["observed_at"])).total_seconds()
if (
key in effective or target is None
or row["fix_receipt_id"] != target["fix_receipt_id"]
or row["enforcement_point_id"] != target["enforcement_point_id"]
or row["adapter_id"] != target["expected_readback_adapter_id"]
or row["observed_effective_state_sha256"] != target["expected_effective_state_sha256"]
or row["observation_complete"] is not True
or row["matches_approved_fix"] is not True
or age < 0 or age > max_effective_age
):
raise RuntimeError("effective-state result is stale, incomplete, or misbound")
effective[key] = row
if set(effective) != set(expected):
raise RuntimeError("effective-state result population differs")
bundle_path = Path("replay/exploit-replay.tar")
bundle_sha256 = hashlib.sha256(input_bytes(bundle_path)).hexdigest()
exact_cases_document = read_json(Path("replay/exact-cases.json"))
exact_manifest_fields = {
"schema_version", "incident_id", "population_sha256", "replay_bundle_sha256",
"allowed_replay_origins", "allowed_observer_origins", "allowed_header_names", "cases",
}
exact_case_fields = {
"case_id", "target_id", "asset_version", "exploit_path_id", "fix_receipt_id",
"method", "url", "safe_headers", "body", "effect_observer_url",
}
if (
not isinstance(exact_cases_document, dict)
or set(exact_cases_document) != exact_manifest_fields
or exact_cases_document["schema_version"] != "aidefend.exact-replay-cases/v2"
or exact_cases_document["incident_id"] != incident_id
or exact_cases_document["population_sha256"] != computed_population_sha256
or require_sha256(exact_cases_document["replay_bundle_sha256"], "exact replay bundle") != bundle_sha256
or not isinstance(exact_cases_document["cases"], list)
or not exact_cases_document["cases"]
):
raise ValueError("exact-case manifest is empty or misbound")
for field in ("allowed_replay_origins", "allowed_observer_origins"):
values = exact_cases_document[field]
if not isinstance(values, list) or not values or len(values) != len(set(values)):
raise ValueError(f"{field} is empty or duplicated")
allowed_headers = exact_cases_document["allowed_header_names"]
if not isinstance(allowed_headers, list) or len(allowed_headers) != len(set(allowed_headers)):
raise ValueError("allowed_header_names is malformed")
exact_cases = {}
for case in exact_cases_document["cases"]:
if not isinstance(case, dict) or set(case) != exact_case_fields:
raise ValueError("exact replay case schema differs")
case_id = require_text(case["case_id"], "exact case_id")
key = (case["target_id"], case["asset_version"], case["exploit_path_id"])
target = expected.get(key)
if (
case_id in exact_cases or target is None
or case["fix_receipt_id"] != target["fix_receipt_id"]
or any(not isinstance(case[field], str) or not case[field] for field in ("method", "url", "effect_observer_url"))
or not isinstance(case["safe_headers"], dict)
or not isinstance(case["body"], str)
):
raise ValueError("exact replay case identity or fix binding differs")
exact_cases[case_id] = (key, target["fix_receipt_id"])
if {value[0] for value in exact_cases.values()} != set(expected):
raise ValueError("exact-case population differs from affected population")
exact_document = read_json(Path("evidence/exact-replay.json"))
exact_document_fields = {"schema_version", "incident_id", "population_sha256", "replay_bundle_sha256", "results"}
exact_result_fields = {
"replay_id", "case_id", "target_id", "asset_version", "exploit_path_id",
"fix_receipt_id", "status_code", "control_path_reached", "observation_complete",
"observation_window_closed", "prohibited_effect_observed", "outcome",
}
if (
not isinstance(exact_document, dict)
or set(exact_document) != exact_document_fields
or exact_document["schema_version"] != "aidefend.exact-replay-results/v2"
or exact_document["incident_id"] != incident_id
or exact_document["population_sha256"] != computed_population_sha256
or exact_document["replay_bundle_sha256"] != bundle_sha256
or not isinstance(exact_document["results"], list)
or not exact_document["results"]
):
raise ValueError("exact replay result document is empty or misbound")
exact_results, replay_ids = {}, set()
for row in exact_document["results"]:
if not isinstance(row, dict) or set(row) != exact_result_fields:
raise ValueError("exact replay result schema differs")
case_id = require_text(row["case_id"], "exact result case_id")
replay_id = require_text(row["replay_id"], "replay_id")
expected_case = exact_cases.get(case_id)
key = (row["target_id"], row["asset_version"], row["exploit_path_id"])
if (
case_id in exact_results or replay_id in replay_ids or expected_case is None
or (key, row["fix_receipt_id"]) != expected_case
or type(row["status_code"]) is not int or not 100 <= row["status_code"] <= 599
or row["control_path_reached"] is not True
or row["observation_complete"] is not True
or row["observation_window_closed"] is not True
or row["prohibited_effect_observed"] is not False
or row["outcome"] != "PASS"
):
raise RuntimeError("exact replay result is failed, incomplete, or misbound")
exact_results[case_id] = row
replay_ids.add(replay_id)
if set(exact_results) != set(exact_cases):
raise RuntimeError("exact replay result population differs from signed cases")
bypass_manifest = read_json(Path("evidence/bypass-case-manifest.json"))
bypass_manifest_fields = {"schema_version", "incident_id", "population_sha256", "cases"}
bypass_case_fields = {"case_id", "target_id", "asset_version", "exploit_path_id", "pytest_nodeid"}
if (
not isinstance(bypass_manifest, dict)
or set(bypass_manifest) != bypass_manifest_fields
or bypass_manifest["schema_version"] != "aidefend.bypass-case-manifest/v2"
or bypass_manifest["incident_id"] != incident_id
or bypass_manifest["population_sha256"] != computed_population_sha256
or not isinstance(bypass_manifest["cases"], list)
or not bypass_manifest["cases"]
):
raise ValueError("bypass manifest is empty or misbound")
bypass_cases, bypass_nodes, bypass_pairs = {}, set(), set()
for case in bypass_manifest["cases"]:
if not isinstance(case, dict) or set(case) != bypass_case_fields:
raise ValueError("bypass case schema differs")
case_id = require_text(case["case_id"], "bypass case_id")
nodeid = require_text(case["pytest_nodeid"], "bypass pytest_nodeid")
key = (case["target_id"], case["asset_version"], case["exploit_path_id"])
if (
case_id in bypass_cases or nodeid in bypass_nodes or key not in expected
or not nodeid.startswith("closure_tests/") or "::" not in nodeid or ".." in nodeid
or any(token in nodeid for token in ("*", "?", "[", "]"))
):
raise ValueError("bypass case identity, selector, or target differs")
bypass_cases[case_id] = key
bypass_nodes.add(nodeid)
bypass_pairs.add(key)
if bypass_pairs != set(expected):
raise ValueError("bypass manifest omits an affected target identity")
bypass_rows = read_jsonl(Path("evidence/bypass-results.jsonl"))
bypass_result_fields = {
"case_id", "target_id", "asset_version", "exploit_path_id",
"control_path_reached", "observation_complete", "prohibited_effect_observed", "outcome",
}
bypass_results = {}
for row in bypass_rows:
if not isinstance(row, dict) or set(row) != bypass_result_fields:
raise ValueError("bypass result schema differs")
case_id = require_text(row["case_id"], "bypass result case_id")
key = (row["target_id"], row["asset_version"], row["exploit_path_id"])
if (
case_id in bypass_results or bypass_cases.get(case_id) != key
or row["control_path_reached"] is not True
or row["observation_complete"] is not True
or row["prohibited_effect_observed"] is not False
or row["outcome"] != "PASS"
):
raise RuntimeError("bypass replay result is failed, incomplete, or misbound")
bypass_results[case_id] = row
if set(bypass_results) != set(bypass_cases):
raise RuntimeError("bypass result population differs from signed cases")
clean_manifest = read_json(Path("evidence/clean-case-manifest.json"))
clean_manifest_fields = {"schema_version", "incident_id", "population_sha256", "cases"}
clean_case_fields = {
"case_id", "target_id", "asset_version", "exploit_path_id", "fix_receipt_id",
"pytest_nodeid", "junit_classname", "junit_name",
}
if (
not isinstance(clean_manifest, dict)
or set(clean_manifest) != clean_manifest_fields
or clean_manifest["schema_version"] != "aidefend.clean-case-manifest/v1"
or clean_manifest["incident_id"] != incident_id
or clean_manifest["population_sha256"] != computed_population_sha256
or not isinstance(clean_manifest["cases"], list)
or not clean_manifest["cases"]
):
raise ValueError("clean-case manifest is empty or misbound")
clean_ids, clean_nodes, junit_ids, clean_pairs = set(), set(), set(), set()
for case in clean_manifest["cases"]:
if not isinstance(case, dict) or set(case) != clean_case_fields:
raise ValueError("clean case schema differs")
case_id = require_text(case["case_id"], "clean case_id")
nodeid = require_text(case["pytest_nodeid"], "clean pytest_nodeid")
junit_id = (require_text(case["junit_classname"], "JUnit classname"), require_text(case["junit_name"], "JUnit name"))
key = (case["target_id"], case["asset_version"], case["exploit_path_id"])
target = expected.get(key)
if (
case_id in clean_ids or nodeid in clean_nodes or junit_id in junit_ids or target is None
or case["fix_receipt_id"] != target["fix_receipt_id"]
or not nodeid.startswith("closure_tests/") or "::" not in nodeid or ".." in nodeid
or any(token in nodeid for token in ("*", "?", "[", "]"))
):
raise ValueError("clean case identity, selector, or fix binding differs")
clean_ids.add(case_id)
clean_nodes.add(nodeid)
junit_ids.add(junit_id)
clean_pairs.add(key)
if clean_pairs != set(expected):
raise ValueError("clean-case manifest omits an affected target identity")
def local_name(tag: str) -> str:
return tag.rsplit("}", 1)[-1]
clean_root = ET.fromstring(input_bytes(Path("evidence/clean-regression.xml")))
testcases = [element for element in clean_root.iter() if local_name(element.tag) == "testcase"]
observed_junit = [(case.get("classname", ""), case.get("name", "")) for case in testcases]
if (
not testcases or len(observed_junit) != len(set(observed_junit)) or set(observed_junit) != junit_ids
or any(any(local_name(child.tag) in {"failure", "error", "skipped"} for child in list(case)) for case in testcases)
):
raise RuntimeError("clean-control JUnit population is incomplete, failed, or skipped")
payload_paths = [payload for payload, _, _ in signed_inputs]
evidence_sha256 = {
str(path).replace("\\", "/"): hashlib.sha256(input_bytes(path)).hexdigest()
for path in payload_paths
}
verdict = {
"schema_version": "aidefend.exploit-path-closure/v3",
"incident_id": incident_id,
"population_sha256": computed_population_sha256,
"population_count": len(targets),
"canonical_fix_receipt_count": len(receipts),
"effective_state_count": len(effective),
"exact_replay_count": len(exact_results),
"bypass_replay_count": len(bypass_results),
"clean_control_count": len(testcases),
"evidence_sha256": evidence_sha256,
"verified_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
"outcome": "PASS",
}
with Path("evidence/closure-verdict.json").open("x", encoding="utf-8") as handle:
json.dump(verdict, handle, sort_keys=True, separators=(",", ":"))
handle.write("\n")set -euo pipefail
: "${E004_CLOSURE_VERIFY_TIMEOUT_SECONDS:?required}"
[[ "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" =~ ^[0-9]+([.][0-9]+)?$ ]]
awk -v value="$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" 'BEGIN { exit !(value > 0) }'
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign sign-blob --yes --key env://CLOSURE_VERIFIER_KEY \
--bundle evidence/closure-verdict.json.sigstore.json \
evidence/closure-verdict.json
timeout "$E004_CLOSURE_VERIFY_TIMEOUT_SECONDS" \
cosign verify-blob --key keys/closure-verifier.pub \
--bundle evidence/closure-verdict.json.sigstore.json \
evidence/closure-verdict.jsonThe closure verifier identity must be independent of the fix owner and replay runners. Hand the signed incident-proven remediation to AID-R-004 for wider fleet propagation; E-004 itself does not deploy it.
Open-source tools
- Greenbone OpenVAS (authenticated vulnerability re-scan and affected-target verification)
- Nuclei (incident-specific exploit-path replay templates)
- Trivy (deployed artifact, package, SBOM, and configuration readback)
- HTTPX (Python client; independent exploit and clean-control replay harness)
- Cosign (signed fix and closure-receipt verification)
Commercial / hosted tools
- Wiz Cloud Security Platform (cloud workload exposure and effective-state verification)
- Palo Alto Networks Prisma Cloud (cloud workload vulnerability and configuration verification)
- Microsoft Defender for Cloud (cloud workload vulnerability and posture verification)
- Tenable Vulnerability Management (authenticated vulnerability re-scan)
- Rapid7 InsightVM (authenticated vulnerability re-scan)
- Snyk Open Source (dependency fix and artifact verification)
- Mend SCA (dependency fix and artifact verification)
Threat mappings
MITRE ATLAS
- AML.T0011 User Execution
- AML.T0010 AI Supply Chain Compromise (independent readback and exploit replay prove the signed supply-chain fix is effective across the affected release population)
- AML.T0010.001 AI Supply Chain Compromise: AI Software (the verifier proves the approved AI-software fix is deployed and the exact exploit no longer succeeds)
- AML.T0011.001 User Execution: Malicious Package (the verifier proves the package removal, deny, or admission fix blocks the exact malicious-package path)
- AML.T0072 Cyber Communication Channel (independent replay proves the fixed path can no longer establish the incident's command-and-control channel)
- AML.T0031 Erode AI Model Integrity (adversarial replay proves the repaired input path no longer reaches the integrity-eroding model effect)
- AML.T0105 Escape to Host (independent boundary replay proves the approved isolation fix prevents the exact host-escape path)
- AML.T0106 Exploitation for Credential Access (version readback and exact exploit replay prove the credential-access vulnerability is closed)
- AML.T0107 Exploitation for Defense Evasion (effective control readback and bypass replay prove the defense-evasion vulnerability is closed)
- AML.T0010.004 AI Supply Chain Compromise: Container Registry (effective-registry readback and replay prove the overwritten-image path is closed)
- AML.T0081 Modify AI Agent Configuration (effective configuration readback and tamper replay prove the exploited write path is closed)
MAESTRO
- Compromised Framework Components (L3) (effective component-version readback and exploit replay prove the signed fix is consumed)
- Compromised Container Images (L4) (digest readback and deployment replay prove every affected workload consumes the approved image)
- Supply Chain Attacks (Cross-Layer) (signed fix binding and cross-layer replay prove the exploited dependency path is closed)
- Input Validation Attacks (L3) (negative, bypass, and exact-exploit replay prove the repaired validation path rejects the incident payload)
- Infrastructure-as-Code (IaC) Manipulation (L4) (effective control-plane readback and tamper replay prove the repaired IaC path is enforced)
OWASP LLM Top 10 2026
- LLM04:2026 Supply Chain (incident-bound independent artifact readback and exploit replay verify the signed component fix)
- LLM03:2026 Excessive Agency (incident-bound effective tool-policy readback and exploit replay verify the repaired authority boundary)
OWASP ML Top 10 2023
- ML06:2023 AI Supply Chain Attacks (independent artifact readback and exploit replay prove the vulnerable library path is closed)
OWASP Top 10 for Agentic Applications 2026
- ASI04:2026 Agentic Supply Chain Vulnerabilities (dependency and deployment readback plus replay prove the signed supply-chain fix is effective)
- ASI05:2026 Unexpected Code Execution (RCE) (exact exploit replay proves the repaired path can no longer reach code execution)
- ASI02:2026 Tool Misuse and Exploitation (effective tool-policy readback and bypass replay prove the exploited tool path is closed)
- ASI03:2026 Identity and Privilege Abuse (effective identity-policy readback and incident replay prove the exploited privilege path is closed)
NIST Adversarial Machine Learning 2025
- NISTAML.051 Model Poisoning (Supply Chain) (artifact readback and replay prove the repaired model-supply-chain path rejects the incident artifact)
Cisco Integrated AI Security and Safety Framework
- AITech-9.3 Dependency / Plugin Compromise (incident-bound independent fix readback and exploit replay verify closure of the compromised dependency path)
- AITech-12.1 Tool Exploitation (incident-bound tool-policy readback and exploit replay verify the repaired tool path denies the protected effect)
Google Secure AI Framework 2.0 - Risks
- MST: Model Source Tampering (source and dependency readback plus tamper replay prove the approved fix prevents re-exploitation)
- MDT: Model Deployment Tampering (effective deployment-state readback and tamper replay prove the repaired path is enforced)
- IIC: Insecure Integrated Component (component-version readback and exploit replay prove the integrated-component fix is effective)
Databricks AI Security Framework 3.0
- Platform 12.1: Lack of vulnerability management (independent fix readback and exploit replay provide closure evidence for the incident vulnerability)
- Algorithms 5.4: Malicious libraries (library digest readback and replay prove the malicious-library path is removed or blocked)
- Model 7.3: ML Supply chain vulnerabilities (artifact and deployment readback prove the signed supply-chain fix reaches the affected population)
- Agents - Tools MCP Server 13.20: Insecure Server Configuration (effective server-configuration readback and bypass replay prove the exploited setting is repaired)
- Agents - Tools MCP Server 13.21: Supply Chain Attacks (MCP component readback and exploit replay prove the approved supply-chain fix is effective)