AID-D-018: Production AI-Security Detection Efficacy & Scenario-Coverage Validation
Continuously validate the efficacy and scenario coverage of the exact AI-security detectors deployed in production by replaying signed benign and attack scenarios through their real observation paths, measuring segmented recall, precision, false-positive rate, errors, and abstentions, and binding results to detector versions, digests, policies, and a complete eligible inventory.
Pillars: app, model, infra
Lifecycle: validation, operation, improvement
Scope boundary
Owns cross-detector production-path efficacy and fleet-wide scenario coverage by replaying signed benign and attack scenarios through real observation paths and measuring segmented recall, precision, false positives, errors, abstentions, and complete detector-by-scenario coverage.
AID-D-005.002: AID-D-018 determines whether deployed detectors identify required scenarios with acceptable efficacy; AID-D-005.002 determines whether rule assets are packaged, deployed, routed, and technically healthy. Execution health is not semantic detection efficacy.
AID-D-017: AID-D-018 measures detector correctness and scenario coverage; AID-D-017 measures detection, acknowledgement, triage latency, and investigation coverage. Detector quality and operational response service levels require separate result sets.
AID-M-008: AID-D-018 validates the production security detectors and their real observation paths; AID-M-008 benchmarks the protected model or agent against adversarial behavior before or alongside release. The detector under test and the protected AI system under test are different assets.
AID-D-008: AID-D-018 validates scenario efficacy for the production detector fleet; AID-D-008 owns the design, training, calibration, and serving of its fusion detector. The efficacy control may test fusion output but does not retrain or operate the fusion model.
Implementation guidance
- AID-D-018-G001: Replay a signed benign-and-attack scenario population through every eligible production detector and observation path, reconcile the complete detector-by-scenario result matrix, calculate policy-owned metrics per detector and segment, and preserve the signed inputs and result matrix for replay.
AID-D-018-G001
Replay a signed benign-and-attack scenario population through every eligible production detector and observation path, reconcile the complete detector-by-scenario result matrix, calculate policy-owned metrics per detector and segment, and preserve the signed inputs and result matrix for replay.
Versioned runtime profile: The evaluator verifies and strictly parses the signed runtime-profile artifact inside main(); the profile supplies the typed command bound, while a separately admitted bootstrap timeout remains available only to verify that profile and sign an ERROR result. Every outcome records the profile version, digest, and verification state; a missing or invalid profile produces signed ERROR evidence without a framework fallback.
Signed assurance inputs
Sign eight evaluator inputs separately, including the governed production-deployment snapshot with independently authorized Sigstore keys: an expected detector-and-segment deployment manifest; the observed detector inventory; semantic benign/attack scenario JSONL; production-path result JSONL; an exact result-evidence object manifest; a runner/completeness receipt; and the metric policy. The policy binds the expected-manifest digest, exact attack-family and semantic-case population for every detector/segment, detector-specific thresholds, and required source checkpoints; observed inventory and test files cannot redefine that expected population. The independently signed expected manifest binds environment, deployment snapshot, detector ID, version, artifact/config digests, observation point, segments, and named technical prerequisites. The separately signed observed inventory records the exact prerequisite observations, technical scope state, and any technical N/A reason for that same detector identity; deletion of an expected detector, segment, or prerequisite observation is therefore a coverage gap, not a smaller passing population. The signed policy derives each eligible detector's exact semantic scenario set; scenario files cannot self-declare scope, repeat payloads under new IDs, or replace required attack families with easier duplicates. Execute through the deployed ingress, telemetry, and alert path—not by calling a classifier function directly. The receipt binds every signed input digest, the policy-required exact source-checkpoint population, runner/harness versions, and DEPLOYED_OBSERVATION_PATH mode.
Complete-matrix evaluator
# file: assurance/evaluate_production_detectors.py
import errno
import fcntl
import hashlib
import hmac
import json
import math
import os
import stat
import subprocess
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
from pathlib import Path
DEPLOYMENT_SNAPSHOT = Path("assurance/production-deployment-snapshot.json")
EXPECTED_DEPLOYMENT = Path("assurance/expected-detector-segment-manifest.json")
INVENTORY = Path("assurance/detector-inventory.json")
SCENARIOS = Path("assurance/scenarios.jsonl")
SCENARIO_PAYLOAD_ROOT = Path("assurance/scenario-payloads")
RESULTS = Path("assurance/production-path-results.jsonl")
RESULT_EVIDENCE_MANIFEST = Path("assurance/result-evidence-manifest.json")
RESULT_EVIDENCE_ROOT = Path("assurance/result-evidence")
RECEIPT = Path("assurance/run-receipt.json")
POLICY = Path("policy/detection-efficacy-policy.json")
HASH_IO_BUFFER_SIZE = 1024 * 1024 # I/O buffer size; never an acceptance threshold.
BOOTSTRAP_PROFILE_BYTES_CEILING = 4 * 1024 * 1024
class CoverageGap(RuntimeError):
pass
def required_env(name: str) -> str:
value = os.environ.get(name)
if value is None or not value.strip():
raise RuntimeError(f"required bootstrap field is absent: {name}")
return value.strip()
def positive_float(value: object, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{field} is not numeric")
number = float(value)
if not math.isfinite(number) or number <= 0:
raise ValueError(f"{field} must be finite and positive")
return number
def digest_bytes(payload: bytes) -> str:
return hashlib.sha256(payload).hexdigest()
def canonical(value: object) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def strict_json_loads(raw: bytes | str) -> object:
if isinstance(raw, bytes):
text = raw.decode("utf-8", errors="strict")
elif isinstance(raw, str):
text = raw
else:
raise TypeError("JSON input must be bytes or text")
def reject_constant(value: str) -> None:
raise ValueError(f"non-finite JSON constant is forbidden: {value}")
def reject_duplicate_keys(pairs: list[tuple[str, object]]) -> dict:
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON object key: {key}")
result[key] = value
return result
return json.loads(
text,
parse_constant=reject_constant,
object_pairs_hook=reject_duplicate_keys,
)
def load_json_bytes(raw: bytes, label: str) -> object:
try:
return strict_json_loads(raw)
except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
raise ValueError(f"{label} is not strict JSON") from error
def load_jsonl_bytes(raw: bytes, label: str, *, allow_empty: bool = False) -> list[dict]:
text = raw.decode("utf-8", errors="strict")
lines = text.splitlines()
if (not allow_empty and not lines) or any(not line.strip() for line in lines):
raise ValueError(f"{label} has an empty or blank JSONL record")
records = [strict_json_loads(line) for line in lines]
if any(not isinstance(record, dict) for record in records):
raise ValueError(f"{label} contains a non-object JSONL record")
return records
def require_sha256(value: object, field: str) -> str:
if (not isinstance(value, str) or len(value) != 64
or any(character not in "0123456789abcdef" for character in value)):
raise ValueError(f"{field} must be a lowercase SHA-256 digest")
return value
def require_text(value: object, field: str) -> str:
if not isinstance(value, str) or not value:
raise ValueError(f"{field} must be a non-empty string")
return value
def parse_time(value: object, field: str) -> datetime:
if not isinstance(value, str):
raise ValueError(f"{field} is not a timestamp")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError(f"{field} lacks timezone")
return parsed.astimezone(timezone.utc)
def require_sorted_strings(value: object, field: str, *, nonempty: bool) -> list[str]:
if (not isinstance(value, list)
or (nonempty and not value)
or value != sorted(set(value))
or any(not isinstance(item, str) or not item for item in value)):
raise ValueError(f"{field} must be a sorted unique string array")
return value
def unit(value: object, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{field} is not numeric")
number = float(value)
if not math.isfinite(number) or not 0 <= number <= 1:
raise ValueError(f"{field} is outside [0,1]")
return number
def _open_directory_no_follow(path: Path) -> int:
if (os.open not in os.supports_dir_fd
or not hasattr(os, "O_DIRECTORY")
or not hasattr(os, "O_NOFOLLOW")):
raise RuntimeError("descriptor-relative no-follow reads are unsupported")
path = Path(path)
anchor = Path(path.anchor) if path.is_absolute() else Path(".")
parts = path.parts[1:] if path.is_absolute() else path.parts
if any(part in {"", ".", ".."} for part in parts):
raise ValueError(f"unsafe path component in {path}")
descriptor = os.open(anchor, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
try:
for part in parts:
next_descriptor = os.open(
part,
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
dir_fd=descriptor,
)
os.close(descriptor)
descriptor = next_descriptor
if not stat.S_ISDIR(os.fstat(descriptor).st_mode):
raise ValueError(f"{path} is not a directory")
return descriptor
except Exception:
os.close(descriptor)
raise
def _open_regular_no_follow(path: Path) -> int:
path = Path(path)
if not path.name or path.name in {".", ".."}:
raise ValueError(f"invalid regular-file path: {path}")
parent_descriptor = _open_directory_no_follow(path.parent)
try:
descriptor = os.open(
path.name,
os.O_RDONLY | os.O_NOFOLLOW,
dir_fd=parent_descriptor,
)
except OSError as error:
if error.errno == errno.ENOENT:
raise FileNotFoundError(f"required file is absent: {path}") from error
raise ValueError(f"{path} is a symlink or invalid file entry") from error
finally:
os.close(parent_descriptor)
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
os.close(descriptor)
raise ValueError(f"{path} is not a regular file")
return descriptor
def read_regular_no_follow(path: Path, maximum_bytes: int) -> bytes:
if isinstance(maximum_bytes, bool) or not isinstance(maximum_bytes, int) or maximum_bytes < 1:
raise ValueError("read byte bound must be a positive integer")
descriptor = _open_regular_no_follow(path)
size = os.fstat(descriptor).st_size
if size > maximum_bytes:
os.close(descriptor)
raise ValueError(f"input exceeds byte bound: {path.name}")
payload = bytearray()
with os.fdopen(descriptor, "rb") as handle:
while chunk := handle.read(min(HASH_IO_BUFFER_SIZE, maximum_bytes + 1 - len(payload))):
payload.extend(chunk)
if len(payload) > maximum_bytes:
raise ValueError(f"input exceeds byte bound: {path.name}")
return bytes(payload)
def _require_memfd_sealing() -> None:
required = (
"memfd_create", "MFD_ALLOW_SEALING", "MFD_CLOEXEC",
)
seal_constants = (
"F_ADD_SEALS", "F_GET_SEALS", "F_SEAL_WRITE", "F_SEAL_GROW",
"F_SEAL_SHRINK", "F_SEAL_SEAL",
)
if (any(not hasattr(os, name) for name in required)
or any(not hasattr(fcntl, name) for name in seal_constants)
or not Path("/proc/self/fd").is_dir()):
raise RuntimeError("Linux sealed memfd and /proc/self/fd are required")
def _memfd_path(descriptor: int) -> str:
return f"/proc/self/fd/{descriptor}"
def _create_memfd(label: str, open_descriptors: list[int]) -> int:
_require_memfd_sealing()
descriptor = os.memfd_create(
f"aidefend-d018-{label}",
os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING,
)
open_descriptors.append(descriptor)
return descriptor
def _write_memfd(descriptor: int, payload: bytes) -> None:
offset = 0
while offset < len(payload):
written = os.write(descriptor, payload[offset:])
if written <= 0:
raise OSError("short write to capture memfd")
offset += written
os.fsync(descriptor)
def _seal_memfd(descriptor: int) -> None:
required_seals = (
fcntl.F_SEAL_WRITE
| fcntl.F_SEAL_GROW
| fcntl.F_SEAL_SHRINK
| fcntl.F_SEAL_SEAL
)
fcntl.fcntl(descriptor, fcntl.F_ADD_SEALS, required_seals)
actual_seals = fcntl.fcntl(descriptor, fcntl.F_GET_SEALS)
if actual_seals & required_seals != required_seals:
raise RuntimeError("capture memfd is not fully sealed")
os.lseek(descriptor, 0, os.SEEK_SET)
def _read_memfd(descriptor: int) -> bytes:
size = os.fstat(descriptor).st_size
payload = bytearray()
offset = 0
while offset < size:
chunk = os.pread(descriptor, min(HASH_IO_BUFFER_SIZE, size - offset), offset)
if not chunk:
raise OSError("short read from capture memfd")
payload.extend(chunk)
offset += len(chunk)
return bytes(payload)
def capture_bytes_to_sealed_memfd(
label: str,
payload: bytes,
open_descriptors: list[int],
) -> dict:
descriptor = _create_memfd(label, open_descriptors)
_write_memfd(descriptor, payload)
_seal_memfd(descriptor)
return {
"data": payload,
"fd": descriptor,
"proc_path": _memfd_path(descriptor),
"sha256": digest_bytes(payload),
}
def verify_captured(
data_capture: dict,
bundle_capture: dict,
key_env: str,
timeout_seconds: float,
) -> None:
subprocess.run(
[
"cosign", "verify-blob", "--key", required_env(key_env),
"--bundle", bundle_capture["proc_path"], data_capture["proc_path"],
],
check=True,
capture_output=True,
text=True,
timeout=timeout_seconds,
pass_fds=(data_capture["fd"], bundle_capture["fd"]),
)
if (not hmac.compare_digest(_read_memfd(data_capture["fd"]), data_capture["data"])
or not hmac.compare_digest(
_read_memfd(bundle_capture["fd"]),
bundle_capture["data"],
)):
raise RuntimeError("sealed signature-verification capture changed")
def capture_signed_input(
source_path: Path,
bundle_source_path: Path,
key_env: str,
timeout_seconds: float,
label: str,
maximum_bytes: int,
digest_sink: dict[str, str],
open_descriptors: list[int],
) -> dict:
if source_path == bundle_source_path:
raise ValueError(f"{label} input and signature bundle paths are identical")
data = read_regular_no_follow(source_path, maximum_bytes)
content_sha256 = digest_bytes(data)
digest_sink[label] = content_sha256
bundle = read_regular_no_follow(bundle_source_path, maximum_bytes)
data_capture = capture_bytes_to_sealed_memfd(
f"{label}-data",
data,
open_descriptors,
)
bundle_capture = capture_bytes_to_sealed_memfd(
f"{label}-bundle",
bundle,
open_descriptors,
)
verify_captured(
data_capture,
bundle_capture,
key_env,
timeout_seconds,
)
return {
**data_capture,
"sha256": content_sha256,
"source_path": str(source_path),
}
def load_bootstrap() -> dict:
try:
timeout = float(required_env("D018_BOOTSTRAP_COMMAND_TIMEOUT_SECONDS"))
except ValueError as error:
raise RuntimeError("bootstrap command timeout is not numeric") from error
timeout = positive_float(timeout, "bootstrap.command_timeout_seconds")
evidence_path = Path(required_env("D018_EVIDENCE_PATH"))
evidence_bundle_path = Path(required_env("D018_EVIDENCE_BUNDLE_PATH"))
if evidence_path == evidence_bundle_path:
raise RuntimeError("evidence and bundle paths must differ")
return {
"command_timeout_seconds": timeout,
"evidence_path": evidence_path,
"evidence_bundle_path": evidence_bundle_path,
}
def load_runtime_profile(
identity: dict,
bootstrap_timeout: float,
open_descriptors: list[int],
) -> dict:
profile_path = Path(required_env("D018_RUNTIME_PROFILE_PATH"))
profile_bundle = Path(required_env("D018_RUNTIME_PROFILE_BUNDLE_PATH"))
captured_digests = {}
try:
captured = capture_signed_input(
profile_path,
profile_bundle,
"D018_RUNTIME_PROFILE_VERIFY_KEY",
bootstrap_timeout,
"runtime_profile",
BOOTSTRAP_PROFILE_BYTES_CEILING,
captured_digests,
open_descriptors,
)
finally:
if "runtime_profile" in captured_digests:
identity["runtime_profile_sha256"] = captured_digests["runtime_profile"]
identity["runtime_profile_status"] = "PRESENT_UNVERIFIED"
identity["runtime_profile_status"] = "SIGNATURE_VERIFIED"
profile = load_json_bytes(captured["data"], "runtime profile")
if isinstance(profile, dict) and isinstance(profile.get("profile_version"), str):
identity["runtime_profile_version"] = profile["profile_version"] or None
fields = {
"schema_version", "profile_version", "command_timeout_seconds",
"maximum_input_bytes", "maximum_object_bytes", "maximum_records",
"maximum_matrix_cells",
}
if (not isinstance(profile, dict)
or set(profile) != fields
or profile["schema_version"] != "aidefend.d018-runtime-profile.v1"):
raise ValueError("runtime profile schema differs")
require_text(profile["profile_version"], "runtime_profile.profile_version")
profile["command_timeout_seconds"] = positive_float(
profile["command_timeout_seconds"],
"runtime_profile.command_timeout_seconds",
)
for field in (
"maximum_input_bytes", "maximum_object_bytes", "maximum_records",
"maximum_matrix_cells",
):
if isinstance(profile[field], bool) or not isinstance(profile[field], int) or profile[field] < 1:
raise ValueError(f"runtime_profile.{field} must be a positive integer")
if profile["maximum_object_bytes"] > profile["maximum_input_bytes"]:
raise ValueError("runtime object byte bound exceeds top-level input bound")
identity["runtime_profile_status"] = "VERIFIED"
return profile
def write_exclusive(path: Path, payload: bytes) -> None:
path = Path(path)
if not path.name or path.name in {".", ".."}:
raise ValueError(f"invalid output path: {path}")
parent_descriptor = _open_directory_no_follow(path.parent)
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
descriptor = os.open(path.name, flags, 0o600, dir_fd=parent_descriptor)
finally:
os.close(parent_descriptor)
with os.fdopen(descriptor, "wb") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
def publish_once_or_match(path: Path, payload: bytes) -> None:
"""Publish once; make an interrupted two-file publication safely retryable."""
try:
write_exclusive(path, payload)
except FileExistsError:
existing = read_regular_no_follow(path, max(len(payload), 1))
if not hmac.compare_digest(existing, payload):
raise RuntimeError("existing publication differs from exact signed bytes")
def sign_evidence_to_sealed_memfds(
payload: bytes,
timeout_seconds: float,
open_descriptors: list[int],
) -> bytes:
data_capture = capture_bytes_to_sealed_memfd(
"final-evidence-data",
payload,
open_descriptors,
)
bundle_descriptor = _create_memfd(
"final-evidence-bundle",
open_descriptors,
)
subprocess.run(
[
"cosign", "sign-blob", "--yes",
"--key", "env://D018_EVIDENCE_SIGNING_KEY",
"--bundle", _memfd_path(bundle_descriptor),
data_capture["proc_path"],
],
check=True,
capture_output=True,
text=True,
timeout=timeout_seconds,
pass_fds=(data_capture["fd"], bundle_descriptor),
)
_seal_memfd(bundle_descriptor)
bundle = _read_memfd(bundle_descriptor)
if not bundle:
raise RuntimeError("cosign produced an empty evidence bundle")
bundle_capture = {
"data": bundle,
"fd": bundle_descriptor,
"proc_path": _memfd_path(bundle_descriptor),
"sha256": digest_bytes(bundle),
}
verify_captured(
data_capture,
bundle_capture,
"D018_EVIDENCE_VERIFY_KEY",
timeout_seconds,
)
return bundle
def read_named_object_no_follow(
root: Path, object_name: object, label: str, maximum_bytes: int,
) -> bytes:
if (not isinstance(object_name, str) or not object_name
or object_name in {".", ".."}
or "/" in object_name or chr(92) in object_name
or Path(object_name).name != object_name):
raise ValueError(f"{label} path must be one basename")
try:
return read_regular_no_follow(root / object_name, maximum_bytes)
except FileNotFoundError as error:
raise CoverageGap(f"required {label} object is absent") from error
def validate_expected_deployment(expected: object) -> dict[str, dict]:
fields = {
"schema_version", "manifest_version", "environment_id",
"deployment_snapshot_sha256", "detectors",
}
if (not isinstance(expected, dict)
or set(expected) != fields
or expected["schema_version"] != "aidefend.d018-expected-deployment.v2"
or not isinstance(expected["detectors"], list)):
raise ValueError("expected deployment manifest schema differs")
require_text(expected["manifest_version"], "expected.manifest_version")
require_text(expected["environment_id"], "expected.environment_id")
require_sha256(
expected["deployment_snapshot_sha256"],
"expected.deployment_snapshot_sha256",
)
detector_fields = {
"detector_id", "detector_type", "version", "artifact_sha256",
"config_sha256", "observation_point", "segments",
"technical_prerequisites",
}
detectors = {}
for item in expected["detectors"]:
if not isinstance(item, dict) or set(item) != detector_fields:
raise ValueError("expected detector schema differs")
detector_id = require_text(item["detector_id"], "expected.detector_id")
if detector_id in detectors:
raise ValueError("duplicate expected detector ID")
for field in ("detector_type", "version", "observation_point"):
require_text(item[field], f"{detector_id}.{field}")
for field in ("artifact_sha256", "config_sha256"):
require_sha256(item[field], f"{detector_id}.{field}")
require_sorted_strings(item["segments"], f"{detector_id}.segments", nonempty=True)
require_sorted_strings(
item["technical_prerequisites"],
f"{detector_id}.technical_prerequisites",
nonempty=False,
)
detectors[detector_id] = item
return detectors
def validate_deployment_snapshot(
snapshot: object, snapshot_sha256: str, expected: dict,
) -> None:
fields = {"schema_version", "snapshot_id", "environment_id", "detectors"}
if (
not isinstance(snapshot, dict)
or set(snapshot) != fields
or snapshot["schema_version"] != "aidefend.production-deployment-snapshot.v1"
or not isinstance(snapshot["detectors"], list)
):
raise ValueError("production deployment snapshot schema differs")
require_text(snapshot["snapshot_id"], "deployment_snapshot.snapshot_id")
if (
snapshot["environment_id"] != expected["environment_id"]
or snapshot["detectors"] != expected["detectors"]
or snapshot_sha256 != expected["deployment_snapshot_sha256"]
):
raise CoverageGap("expected detector population differs from governed deployment snapshot")
def validate_inventory(inventory: object, expected: dict, expected_by_id: dict) -> dict:
fields = {
"schema_version", "inventory_version", "environment_id",
"deployment_snapshot_sha256", "collection_complete", "detectors",
}
if (not isinstance(inventory, dict)
or set(inventory) != fields
or inventory["schema_version"] != "aidefend.detector-inventory.v2"
or not isinstance(inventory["collection_complete"], bool)
or not isinstance(inventory["detectors"], list)):
raise ValueError("inventory schema differs")
require_text(inventory["inventory_version"], "inventory.inventory_version")
if (inventory["environment_id"] != expected["environment_id"]
or inventory["deployment_snapshot_sha256"]
!= expected["deployment_snapshot_sha256"]):
raise ValueError("inventory deployment binding differs")
if inventory["collection_complete"] is not True:
raise CoverageGap("detector inventory collection is incomplete")
detector_fields = {
"detector_id", "detector_type", "version", "artifact_sha256",
"config_sha256", "observation_point", "technical_scope",
"prerequisite_results", "not_applicable_reason", "segments",
"technical_prerequisites",
}
actual_by_id = {}
for item in inventory["detectors"]:
if (not isinstance(item, dict) or set(item) != detector_fields
or item["technical_scope"] not in {"ELIGIBLE", "NOT_APPLICABLE"}):
raise ValueError("inventory detector schema differs")
detector_id = require_text(item["detector_id"], "inventory.detector_id")
if detector_id in actual_by_id:
raise ValueError("duplicate inventory detector ID")
actual_by_id[detector_id] = item
missing = sorted(set(expected_by_id) - set(actual_by_id))
unexpected = sorted(set(actual_by_id) - set(expected_by_id))
if missing:
raise CoverageGap(f"expected detector inventory entries are absent: {missing}")
if unexpected:
raise ValueError(f"inventory has unapproved detector entries: {unexpected}")
for detector_id, item in actual_by_id.items():
expected_item = expected_by_id[detector_id]
for field in (
"detector_type", "version", "artifact_sha256", "config_sha256",
"observation_point", "technical_prerequisites",
):
if item[field] != expected_item[field]:
raise ValueError(f"{detector_id}.{field} differs from expected deployment")
require_sorted_strings(item["segments"], f"{detector_id}.segments", nonempty=True)
missing_segments = sorted(set(expected_item["segments"]) - set(item["segments"]))
extra_segments = sorted(set(item["segments"]) - set(expected_item["segments"]))
if missing_segments:
raise CoverageGap(f"{detector_id} expected segments are absent: {missing_segments}")
if extra_segments:
raise ValueError(f"{detector_id} has unapproved segments: {extra_segments}")
prerequisites = item["technical_prerequisites"]
readings = item["prerequisite_results"]
if not isinstance(readings, dict):
raise ValueError(f"{detector_id}.prerequisite_results is not an object")
missing_readings = sorted(set(prerequisites) - set(readings))
extra_readings = sorted(set(readings) - set(prerequisites))
if missing_readings:
raise CoverageGap(
f"{detector_id} technical prerequisite observations are absent: "
f"{missing_readings}"
)
if extra_readings or any(not isinstance(value, bool) for value in readings.values()):
raise ValueError(f"{detector_id} technical prerequisite observations differ")
reason = item["not_applicable_reason"]
if item["technical_scope"] == "ELIGIBLE":
if not all(readings.values()) or reason is not None:
raise ValueError(f"{detector_id} eligible technical scope is inconsistent")
elif not prerequisites or all(readings.values()):
raise ValueError(f"{detector_id} N/A requires a failed technical prerequisite")
else:
require_text(reason, f"{detector_id}.not_applicable_reason")
return actual_by_id
def validate_policy(policy: object, expected_sha256: str, expected_by_id: dict) -> tuple[dict, dict]:
fields = {
"schema_version", "policy_version", "expected_deployment_sha256",
"required_source_checkpoints", "required_scenarios_by_detector_segment",
"thresholds_by_detector", "assurance_window",
}
if (not isinstance(policy, dict)
or set(policy) != fields
or policy["schema_version"] != "aidefend.detection-efficacy-policy.v2"
or policy["expected_deployment_sha256"] != expected_sha256):
raise ValueError("metric policy schema or expected-deployment binding differs")
require_text(policy["policy_version"], "policy.policy_version")
window = policy["assurance_window"]
window_fields = {
"window_id", "window_start", "window_end",
"maximum_future_skew_seconds", "maximum_evidence_age_seconds",
}
if not isinstance(window, dict) or set(window) != window_fields:
raise ValueError("assurance window schema differs")
require_text(window["window_id"], "assurance_window.window_id")
start = parse_time(window["window_start"], "assurance_window.window_start")
end = parse_time(window["window_end"], "assurance_window.window_end")
for field in ("maximum_future_skew_seconds", "maximum_evidence_age_seconds"):
if isinstance(window[field], bool) or not isinstance(window[field], int) or window[field] < 0:
raise ValueError(f"assurance_window.{field} is invalid")
now = datetime.now(timezone.utc)
if (
not start < end
or end > now + timedelta(seconds=window["maximum_future_skew_seconds"])
or now - end > timedelta(seconds=window["maximum_evidence_age_seconds"])
):
raise CoverageGap("assurance window is stale, future-dated, or unordered")
require_sorted_strings(
policy["required_source_checkpoints"],
"policy.required_source_checkpoints",
nonempty=True,
)
requirements = policy["required_scenarios_by_detector_segment"]
thresholds = policy["thresholds_by_detector"]
if (not isinstance(requirements, dict) or set(requirements) != set(expected_by_id)
or not isinstance(thresholds, dict) or set(thresholds) != set(expected_by_id)):
raise ValueError("policy detector population differs from expected deployment")
requirement_fields = {
"semantic_case_id", "ground_truth", "attack_family", "scenario_version",
"payload_sha256", "harness_id", "harness_version",
"harness_artifact_sha256",
}
threshold_fields = {
"minimum_attack_cases", "minimum_benign_cases", "minimum_recall",
"minimum_precision", "maximum_false_positive_rate",
}
normalized_requirements = {}
catalog_identity = {}
for detector_id, expected_detector in expected_by_id.items():
required_segments = requirements[detector_id]
detector_thresholds = thresholds[detector_id]
segments = set(expected_detector["segments"])
if (not isinstance(required_segments, dict) or set(required_segments) != segments
or not isinstance(detector_thresholds, dict)
or set(detector_thresholds) != segments):
raise ValueError(f"{detector_id} policy segment population differs")
for segment in expected_detector["segments"]:
cases = required_segments[segment]
threshold = detector_thresholds[segment]
if not isinstance(cases, list) or not cases:
raise ValueError(f"{detector_id}/{segment} has no required scenarios")
order = []
seen_case_ids = set()
normalized = []
for case in cases:
if (not isinstance(case, dict) or set(case) != requirement_fields
or case["ground_truth"] not in {"ATTACK", "BENIGN"}):
raise ValueError(f"{detector_id}/{segment} scenario requirement differs")
semantic_case_id = require_text(
case["semantic_case_id"],
f"{detector_id}/{segment}.semantic_case_id",
)
attack_family = require_text(
case["attack_family"],
f"{detector_id}/{segment}.attack_family",
)
scenario_version = require_text(
case["scenario_version"],
f"{detector_id}/{segment}.scenario_version",
)
payload_sha256 = require_sha256(
case["payload_sha256"],
f"{detector_id}/{segment}.payload_sha256",
)
harness_id = require_text(
case["harness_id"],
f"{detector_id}/{segment}.harness_id",
)
harness_version = require_text(
case["harness_version"],
f"{detector_id}/{segment}.harness_version",
)
harness_artifact_sha256 = require_sha256(
case["harness_artifact_sha256"],
f"{detector_id}/{segment}.harness_artifact_sha256",
)
if semantic_case_id in seen_case_ids:
raise ValueError(f"{detector_id}/{segment} duplicates a semantic case")
seen_case_ids.add(semantic_case_id)
identity = (segment, semantic_case_id)
classification = (
case["ground_truth"], attack_family, scenario_version,
payload_sha256, harness_id, harness_version,
harness_artifact_sha256,
)
if identity in catalog_identity and catalog_identity[identity] != classification:
raise ValueError(f"{segment}/{semantic_case_id} has conflicting exact binding")
catalog_identity[identity] = classification
order.append((
semantic_case_id, case["ground_truth"], attack_family,
scenario_version, payload_sha256, harness_id, harness_version,
harness_artifact_sha256,
))
normalized.append((
segment, case["ground_truth"], attack_family, semantic_case_id,
scenario_version, payload_sha256, harness_id, harness_version,
harness_artifact_sha256,
))
if order != sorted(order):
raise ValueError(f"{detector_id}/{segment} scenarios are not canonical")
if not isinstance(threshold, dict) or set(threshold) != threshold_fields:
raise ValueError(f"{detector_id}/{segment} threshold schema differs")
for field in ("minimum_attack_cases", "minimum_benign_cases"):
if (isinstance(threshold[field], bool)
or not isinstance(threshold[field], int)
or threshold[field] < 1):
raise ValueError(f"{detector_id}/{segment}.{field} is invalid")
for field in (
"minimum_recall", "minimum_precision", "maximum_false_positive_rate",
):
unit(threshold[field], f"{detector_id}/{segment}.{field}")
attack_count = sum(case[1] == "ATTACK" for case in normalized)
benign_count = sum(case[1] == "BENIGN" for case in normalized)
if (attack_count < threshold["minimum_attack_cases"]
or benign_count < threshold["minimum_benign_cases"]):
raise ValueError(f"{detector_id}/{segment} policy minimum exceeds exact scenarios")
normalized_requirements[(detector_id, segment)] = set(normalized)
required_catalog = {
(
segment, ground_truth, attack_family, semantic_case_id,
scenario_version, payload_sha256, harness_id, harness_version,
harness_artifact_sha256,
)
for cases in normalized_requirements.values()
for (
segment, ground_truth, attack_family, semantic_case_id,
scenario_version, payload_sha256, harness_id, harness_version,
harness_artifact_sha256,
) in cases
}
return normalized_requirements, required_catalog
def validate_scenarios(
scenarios: list[dict],
required_catalog: set,
payload_root: Path,
maximum_object_bytes: int,
) -> dict:
fields = {
"schema_version", "scenario_id", "scenario_version", "segment",
"ground_truth", "attack_family", "semantic_case_id", "payload_sha256",
"payload_object", "harness_id", "harness_version",
"harness_artifact_sha256",
}
by_identity = {}
scenario_ids = set()
payload_digests = set()
payload_objects = set()
semantic_locators = set()
for scenario in scenarios:
if (set(scenario) != fields
or scenario["schema_version"] != "aidefend.detection-scenario.v2"
or scenario["ground_truth"] not in {"ATTACK", "BENIGN"}):
raise ValueError("scenario schema differs")
for field in (
"scenario_id", "scenario_version", "segment", "attack_family",
"semantic_case_id", "payload_object", "harness_id", "harness_version",
):
require_text(scenario[field], f"scenario.{field}")
require_sha256(scenario["payload_sha256"], "scenario.payload_sha256")
require_sha256(
scenario["harness_artifact_sha256"],
"scenario.harness_artifact_sha256",
)
identity = (
scenario["segment"], scenario["ground_truth"],
scenario["attack_family"], scenario["semantic_case_id"],
scenario["scenario_version"], scenario["payload_sha256"],
scenario["harness_id"], scenario["harness_version"],
scenario["harness_artifact_sha256"],
)
semantic_locator = (scenario["segment"], scenario["semantic_case_id"])
if scenario["scenario_id"] in scenario_ids:
raise ValueError("duplicate scenario ID")
if identity in by_identity or semantic_locator in semantic_locators:
raise ValueError("duplicate semantic scenario case")
if scenario["payload_sha256"] in payload_digests:
raise ValueError("duplicate scenario payload digest")
if scenario["payload_object"] in payload_objects:
raise ValueError("duplicate scenario payload object")
payload = read_named_object_no_follow(
payload_root,
scenario["payload_object"],
"scenario payload",
maximum_object_bytes,
)
if not payload:
raise ValueError("scenario payload object is empty")
if not hmac.compare_digest(digest_bytes(payload), scenario["payload_sha256"]):
raise ValueError("scenario payload object digest differs")
scenario_ids.add(scenario["scenario_id"])
semantic_locators.add(semantic_locator)
payload_digests.add(scenario["payload_sha256"])
payload_objects.add(scenario["payload_object"])
by_identity[identity] = scenario
missing = sorted(required_catalog - set(by_identity))
unexpected = sorted(set(by_identity) - required_catalog)
if missing:
raise CoverageGap(f"required semantic scenarios are absent: {missing}")
if unexpected:
raise ValueError(f"scenario catalog has unapproved cases: {unexpected}")
return by_identity
def validate_results(results: list[dict], policy: dict) -> None:
window = policy["assurance_window"]
window_start = parse_time(window["window_start"], "assurance_window.window_start")
window_end = parse_time(window["window_end"], "assurance_window.window_end")
fields = {
"schema_version", "result_id", "scenario_id", "scenario_version",
"payload_sha256", "harness_id", "harness_version",
"harness_artifact_sha256", "detector_id",
"detector_version", "detector_artifact_sha256", "detector_config_sha256",
"observation_point", "production_request_id", "production_event_id",
"outcome", "alert_event_id", "evidence_sha256", "observed_at",
}
result_ids = set()
production_event_ids = set()
alert_event_ids = set()
for result in results:
if (set(result) != fields
or result["schema_version"] != "aidefend.production-detector-result.v2"
or result["outcome"] not in {"DETECTED", "NOT_DETECTED", "ERROR", "ABSTAIN"}):
raise ValueError("production result schema differs")
result_id = require_text(result["result_id"], "result.result_id")
if result_id in result_ids:
raise ValueError("duplicate result ID")
result_ids.add(result_id)
for field in (
"scenario_id", "scenario_version", "harness_id", "harness_version",
"detector_id", "detector_version", "observation_point",
"production_request_id", "production_event_id",
):
require_text(result[field], f"result.{field}")
observed_at = parse_time(result["observed_at"], "result.observed_at")
if not window_start <= observed_at < window_end:
raise CoverageGap("production result falls outside assurance window")
for field in (
"payload_sha256", "harness_artifact_sha256", "detector_artifact_sha256",
"detector_config_sha256", "evidence_sha256",
):
require_sha256(result[field], f"result.{field}")
if result["production_event_id"] in production_event_ids:
raise ValueError("duplicate production observation event ID")
production_event_ids.add(result["production_event_id"])
alert_id = result["alert_event_id"]
if result["outcome"] == "DETECTED":
require_text(alert_id, "result.alert_event_id")
if alert_id in alert_event_ids:
raise ValueError("duplicate alert event ID")
alert_event_ids.add(alert_id)
elif alert_id is not None:
raise ValueError("non-DETECTED outcomes must have a null alert_event_id")
def result_replay_binding(result: dict) -> dict:
fields = (
"result_id", "scenario_id", "scenario_version", "payload_sha256",
"harness_id", "harness_version", "harness_artifact_sha256",
"detector_id", "production_request_id", "production_event_id",
"observed_at",
)
return {field: result[field] for field in fields}
def replay_bindings_sha256(results: list[dict]) -> str:
bindings = sorted(
(result_replay_binding(result) for result in results),
key=lambda item: item["result_id"],
)
return digest_bytes(canonical(bindings))
def validate_result_evidence(
manifest: object,
results: list[dict],
results_sha256: str,
maximum_object_bytes: int,
) -> None:
fields = {"schema_version", "results_sha256", "objects"}
if (not isinstance(manifest, dict)
or set(manifest) != fields
or manifest["schema_version"] != "aidefend.d018-result-evidence-manifest.v2"
or manifest["results_sha256"] != results_sha256
or not isinstance(manifest["objects"], list)):
raise ValueError("result evidence manifest differs")
object_fields = {"result_id", "object_path", "evidence_sha256"}
by_result = {}
paths = set()
for item in manifest["objects"]:
if not isinstance(item, dict) or set(item) != object_fields:
raise ValueError("result evidence object schema differs")
result_id = require_text(item["result_id"], "result_evidence.result_id")
object_path = require_text(item["object_path"], "result_evidence.object_path")
require_sha256(item["evidence_sha256"], "result_evidence.evidence_sha256")
if result_id in by_result or object_path in paths:
raise ValueError("duplicate result evidence ID or object path")
by_result[result_id] = item
paths.add(object_path)
result_by_id = {result["result_id"]: result for result in results}
missing = sorted(set(result_by_id) - set(by_result))
unexpected = sorted(set(by_result) - set(result_by_id))
if missing:
raise CoverageGap(f"result evidence manifest entries are absent: {missing}")
if unexpected:
raise ValueError(f"result evidence manifest has unknown results: {unexpected}")
for result_id, result in result_by_id.items():
item = by_result[result_id]
raw = read_named_object_no_follow(
RESULT_EVIDENCE_ROOT,
item["object_path"],
"production observation evidence",
maximum_object_bytes,
)
computed = digest_bytes(raw)
if (not hmac.compare_digest(computed, item["evidence_sha256"])
or item["evidence_sha256"] != result["evidence_sha256"]):
raise ValueError(f"{result_id} result evidence digest differs")
observation = load_json_bytes(raw, f"{result_id} production observation")
observation_fields = {
"schema_version", "result_id", "scenario_id", "scenario_version",
"payload_sha256", "harness_id", "harness_version",
"harness_artifact_sha256", "detector_id", "detector_version",
"detector_artifact_sha256", "detector_config_sha256",
"observation_point", "production_request_id", "production_event_id",
"observed_at", "outcome", "alert_event_id",
}
if (not isinstance(observation, dict)
or set(observation) != observation_fields
or observation["schema_version"]
!= "aidefend.d018-production-observation.v1"):
raise ValueError(f"{result_id} production observation schema differs")
expected_observation = {
"schema_version": "aidefend.d018-production-observation.v1",
**{
field: result[field]
for field in observation_fields - {"schema_version"}
},
}
if (observation != expected_observation
or not hmac.compare_digest(raw, canonical(observation))):
raise ValueError(f"{result_id} replay-path evidence binding differs")
def validate_receipt(
receipt: object,
profile_sha256: str,
policy: dict,
input_digests: dict[str, str],
scenarios: list[dict],
results: list[dict],
) -> None:
fields = {
"schema_version", "execution_mode", "deployment_snapshot_sha256",
"expected_deployment_sha256", "inventory_sha256", "scenarios_sha256",
"results_sha256", "result_evidence_manifest_sha256", "policy_sha256",
"runtime_profile_sha256", "scenario_count", "result_count", "runner_id",
"runner_version", "runner_artifact_sha256", "source_checkpoints",
"window_id", "window_start", "window_end", "observation_through",
"replay_bindings_sha256", "collection_complete",
}
if (not isinstance(receipt, dict)
or set(receipt) != fields
or receipt["schema_version"] != "aidefend.d018-run-receipt.v4"
or receipt["execution_mode"] != "DEPLOYED_OBSERVATION_PATH"):
raise ValueError("run receipt schema or mode differs")
expected_digests = {
"deployment_snapshot_sha256": input_digests["deployment_snapshot"],
"expected_deployment_sha256": input_digests["expected_deployment"],
"inventory_sha256": input_digests["inventory"],
"scenarios_sha256": input_digests["scenarios"],
"results_sha256": input_digests["results"],
"result_evidence_manifest_sha256": input_digests["result_evidence_manifest"],
"policy_sha256": input_digests["policy"],
"runtime_profile_sha256": profile_sha256,
}
if any(receipt[field] != expected for field, expected in expected_digests.items()):
raise ValueError("run receipt input digest differs")
window = policy["assurance_window"]
if (
receipt["window_id"] != window["window_id"]
or receipt["window_start"] != window["window_start"]
or receipt["window_end"] != window["window_end"]
):
raise ValueError("run receipt assurance window differs from policy")
observation_through = parse_time(receipt["observation_through"], "receipt.observation_through")
window_end = parse_time(receipt["window_end"], "receipt.window_end")
now = datetime.now(timezone.utc)
if (
observation_through < window_end
or observation_through > now + timedelta(seconds=window["maximum_future_skew_seconds"])
):
raise CoverageGap("receipt observation horizon is incomplete or future-dated")
for field in ("scenario_count", "result_count"):
if isinstance(receipt[field], bool) or not isinstance(receipt[field], int) or receipt[field] < 0:
raise ValueError(f"receipt.{field} is invalid")
if (receipt["scenario_count"] != len(scenarios)
or receipt["result_count"] != len(results)):
raise ValueError("run receipt population count differs from signed inputs")
for field in ("runner_id", "runner_version"):
require_text(receipt[field], f"receipt.{field}")
require_sha256(receipt["runner_artifact_sha256"], "receipt.runner_artifact_sha256")
require_sha256(receipt["replay_bindings_sha256"], "receipt.replay_bindings_sha256")
if not hmac.compare_digest(
receipt["replay_bindings_sha256"],
replay_bindings_sha256(results),
):
raise ValueError("run receipt replay binding population differs")
checkpoints = receipt["source_checkpoints"]
if not isinstance(checkpoints, dict):
raise ValueError("run source checkpoints are not an object")
required = set(policy["required_source_checkpoints"])
missing = sorted(required - set(checkpoints))
unexpected = sorted(set(checkpoints) - required)
if missing:
raise CoverageGap(f"required source checkpoints are absent: {missing}")
if unexpected:
raise ValueError(f"run has unapproved source checkpoints: {unexpected}")
checkpoint_fields = {
"first", "last", "expected_count", "observed_count", "gap_free", "complete",
}
for name, checkpoint in checkpoints.items():
if not isinstance(checkpoint, dict) or set(checkpoint) != checkpoint_fields:
raise ValueError(f"{name} checkpoint schema differs")
for field in ("first", "last", "expected_count", "observed_count"):
if (isinstance(checkpoint[field], bool)
or not isinstance(checkpoint[field], int)
or checkpoint[field] < 0):
raise ValueError(f"{name}.{field} is invalid")
if (checkpoint["last"] < checkpoint["first"]
or not isinstance(checkpoint["gap_free"], bool)
or not isinstance(checkpoint["complete"], bool)):
raise ValueError(f"{name} checkpoint range or state differs")
if (checkpoint["complete"] is not True
or checkpoint["gap_free"] is not True
or checkpoint["expected_count"] != checkpoint["observed_count"]):
raise CoverageGap(f"{name} source checkpoint is incomplete")
if not isinstance(receipt["collection_complete"], bool):
raise ValueError("receipt.collection_complete is not boolean")
if receipt["collection_complete"] is not True:
raise CoverageGap("runner collection is incomplete")
def evaluate(
profile: dict,
profile_sha256: str,
open_descriptors: list[int],
input_digests: dict[str, str],
) -> tuple[dict, dict]:
timeout = profile["command_timeout_seconds"]
signed_inputs = (
("deployment_snapshot", DEPLOYMENT_SNAPSHOT, "D018_DEPLOYMENT_SNAPSHOT_VERIFY_KEY"),
("expected_deployment", EXPECTED_DEPLOYMENT, "D018_EXPECTED_DEPLOYMENT_VERIFY_KEY"),
("inventory", INVENTORY, "D018_INVENTORY_VERIFY_KEY"),
("scenarios", SCENARIOS, "D018_SCENARIO_VERIFY_KEY"),
("results", RESULTS, "D018_RUNNER_VERIFY_KEY"),
(
"result_evidence_manifest",
RESULT_EVIDENCE_MANIFEST,
"D018_RESULT_EVIDENCE_VERIFY_KEY",
),
("receipt", RECEIPT, "D018_COMPLETENESS_VERIFY_KEY"),
("policy", POLICY, "D018_POLICY_VERIFY_KEY"),
)
captured = {}
for label, path, key_env in signed_inputs:
captured[label] = capture_signed_input(
path,
Path(str(path) + ".sigstore.json"),
key_env,
timeout,
label,
profile["maximum_input_bytes"],
input_digests,
open_descriptors,
)
deployment_snapshot = load_json_bytes(
captured["deployment_snapshot"]["data"],
"production deployment snapshot",
)
expected = load_json_bytes(captured["expected_deployment"]["data"], "expected deployment")
inventory = load_json_bytes(captured["inventory"]["data"], "detector inventory")
scenarios = load_jsonl_bytes(
captured["scenarios"]["data"],
"scenario catalog",
allow_empty=True,
)
results = load_jsonl_bytes(
captured["results"]["data"],
"production-path results",
allow_empty=True,
)
result_evidence = load_json_bytes(
captured["result_evidence_manifest"]["data"],
"result evidence manifest",
)
receipt = load_json_bytes(captured["receipt"]["data"], "run receipt")
policy = load_json_bytes(captured["policy"]["data"], "metric policy")
bounded_populations = {
"deployment snapshot detectors": (
deployment_snapshot.get("detectors")
if isinstance(deployment_snapshot, dict) else None
),
"expected deployment detectors": (
expected.get("detectors") if isinstance(expected, dict) else None
),
"inventory detectors": (
inventory.get("detectors") if isinstance(inventory, dict) else None
),
"scenario records": scenarios,
"result records": results,
"result evidence objects": (
result_evidence.get("objects")
if isinstance(result_evidence, dict) else None
),
}
for label, population in bounded_populations.items():
if not isinstance(population, list):
raise ValueError(label + " is not a list")
if len(population) > profile["maximum_records"]:
raise ValueError(label + " exceeds the signed record ceiling")
expected_by_id = validate_expected_deployment(expected)
validate_deployment_snapshot(
deployment_snapshot,
captured["deployment_snapshot"]["sha256"],
expected,
)
normalized_requirements, required_catalog = validate_policy(
policy,
captured["expected_deployment"]["sha256"],
expected_by_id,
)
inventory_by_id = validate_inventory(inventory, expected, expected_by_id)
scenarios_by_identity = validate_scenarios(
scenarios,
required_catalog,
SCENARIO_PAYLOAD_ROOT,
profile["maximum_object_bytes"],
)
validate_results(results, policy)
validate_result_evidence(
result_evidence,
results,
captured["results"]["sha256"],
profile["maximum_object_bytes"],
)
validate_receipt(
receipt,
profile_sha256,
policy,
input_digests,
scenarios,
results,
)
eligible = {
detector_id: detector
for detector_id, detector in inventory_by_id.items()
if detector["technical_scope"] == "ELIGIBLE"
}
if not eligible:
if results:
raise ValueError("N/A detector population cannot have production results")
return (
{
"status": "NOT_APPLICABLE",
"reason": "every expected detector lacks one or more declared technical prerequisites",
"groups": [],
"expected_cells": 0,
"observed_cells": 0,
},
{"expected": expected, "inventory": inventory, "policy": policy, "receipt": receipt},
)
expected_cells = {}
for (detector_id, segment), requirements in normalized_requirements.items():
if detector_id not in eligible:
continue
for identity in requirements:
scenario = scenarios_by_identity[identity]
expected_cells[(scenario["scenario_id"], detector_id)] = scenario
if len(expected_cells) > profile["maximum_matrix_cells"]:
raise ValueError("detector-scenario matrix exceeds the signed cell ceiling")
observed_cells = {}
for result in results:
cell = (result["scenario_id"], result["detector_id"])
if cell in observed_cells:
raise ValueError("duplicate detector-scenario result cell")
observed_cells[cell] = result
missing = sorted(set(expected_cells) - set(observed_cells))
unexpected = sorted(set(observed_cells) - set(expected_cells))
if missing:
raise CoverageGap(f"required detector-scenario result cells are absent: {missing}")
if unexpected:
raise ValueError(f"results have unapproved detector-scenario cells: {unexpected}")
grouped = defaultdict(list)
for cell, result in observed_cells.items():
scenario = expected_cells[cell]
detector = eligible[result["detector_id"]]
if (result["scenario_version"] != scenario["scenario_version"]
or result["payload_sha256"] != scenario["payload_sha256"]
or result["harness_id"] != scenario["harness_id"]
or result["harness_version"] != scenario["harness_version"]
or result["harness_artifact_sha256"]
!= scenario["harness_artifact_sha256"]
or result["detector_version"] != detector["version"]
or result["detector_artifact_sha256"] != detector["artifact_sha256"]
or result["detector_config_sha256"] != detector["config_sha256"]
or result["observation_point"] != detector["observation_point"]):
raise ValueError("result scenario, harness, or detector binding differs")
grouped[(result["detector_id"], scenario["segment"])].append((scenario, result))
reports = []
failures = []
for detector_id, detector in sorted(eligible.items()):
for segment in expected_by_id[detector_id]["segments"]:
rows = grouped[(detector_id, segment)]
threshold = policy["thresholds_by_detector"][detector_id][segment]
counts = Counter()
for scenario, result in rows:
if scenario["ground_truth"] == "ATTACK":
counts["tp" if result["outcome"] == "DETECTED" else "fn"] += 1
else:
counts["tn" if result["outcome"] == "NOT_DETECTED" else "fp"] += 1
if result["outcome"] in {"ERROR", "ABSTAIN"}:
counts[result["outcome"].lower()] += 1
attacks = counts["tp"] + counts["fn"]
benign = counts["tn"] + counts["fp"]
if (attacks < threshold["minimum_attack_cases"]
or benign < threshold["minimum_benign_cases"]):
raise CoverageGap(f"{detector_id}/{segment} class population is incomplete")
recall = counts["tp"] / attacks
false_positive_rate = counts["fp"] / benign
precision_denominator = counts["tp"] + counts["fp"]
precision = counts["tp"] / precision_denominator if precision_denominator else 0.0
passed = (
recall >= threshold["minimum_recall"]
and precision >= threshold["minimum_precision"]
and false_positive_rate <= threshold["maximum_false_positive_rate"]
and counts["error"] == 0
and counts["abstain"] == 0
)
report = {
"detector_id": detector_id,
"version": detector["version"],
"artifact_sha256": detector["artifact_sha256"],
"config_sha256": detector["config_sha256"],
"observation_point": detector["observation_point"],
"segment": segment,
"population": len(rows),
"counts": dict(counts),
"recall": recall,
"precision": precision,
"false_positive_rate": false_positive_rate,
"passed": passed,
}
reports.append(report)
if not passed:
failures.append({"detector_id": detector_id, "segment": segment})
result = {
"status": "FAIL" if failures else "PASS",
"reason": (
"one or more detector segments missed signed thresholds"
if failures else "all exact detector/segment scenario populations met signed thresholds"
),
"groups": reports,
"failed_groups": failures,
"expected_cells": len(expected_cells),
"observed_cells": len(observed_cells),
}
return result, {"expected": expected, "inventory": inventory, "policy": policy, "receipt": receipt}
def base_evidence(status: str, reason: str, identity: dict, input_digests: dict) -> dict:
return {
"schema_version": "aidefend.d018-evidence.v2",
"control": "AID-D-018",
"status": status,
"reason": reason,
"runtime_profile_version": identity["runtime_profile_version"],
"runtime_profile_sha256": identity["runtime_profile_sha256"],
"runtime_profile_status": identity["runtime_profile_status"],
"input_digests": input_digests,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
}
def main() -> None:
bootstrap = load_bootstrap()
open_descriptors = []
identity = {
"runtime_profile_version": None,
"runtime_profile_sha256": None,
"runtime_profile_status": "ABSENT",
}
input_digests = {}
profile = None
try:
profile = load_runtime_profile(
identity,
bootstrap["command_timeout_seconds"],
open_descriptors,
)
result, context = evaluate(
profile,
identity["runtime_profile_sha256"],
open_descriptors,
input_digests,
)
evidence = base_evidence(result["status"], result["reason"], identity, input_digests)
evidence.update(result)
evidence.update({
"environment_id": context["expected"]["environment_id"],
"deployment_snapshot_sha256": context["expected"]["deployment_snapshot_sha256"],
"expected_manifest_version": context["expected"]["manifest_version"],
"inventory_version": context["inventory"]["inventory_version"],
"policy_version": context["policy"]["policy_version"],
"runner_id": context["receipt"]["runner_id"],
"runner_version": context["receipt"]["runner_version"],
"runner_artifact_sha256": context["receipt"]["runner_artifact_sha256"],
})
except CoverageGap:
evidence = base_evidence(
"INSUFFICIENT_DATA",
"coverage or currentness evidence is incomplete",
identity,
input_digests,
)
evidence["error_class"] = "CoverageGap"
except Exception as error:
evidence = base_evidence(
"ERROR",
"evaluation dependency or input verification failed",
identity,
input_digests,
)
evidence["error_class"] = type(error).__name__
timeout = (
profile["command_timeout_seconds"]
if profile is not None else bootstrap["command_timeout_seconds"]
)
evidence_path = bootstrap["evidence_path"]
bundle_path = bootstrap["evidence_bundle_path"]
payload = canonical(evidence)
signed_bundle = sign_evidence_to_sealed_memfds(
payload,
timeout,
open_descriptors,
)
publish_once_or_match(bundle_path, signed_bundle)
publish_once_or_match(evidence_path, payload)
publication_limit = (
profile["maximum_input_bytes"]
if profile is not None else BOOTSTRAP_PROFILE_BYTES_CEILING
)
published_digests = {}
published = capture_signed_input(
evidence_path,
bundle_path,
"D018_EVIDENCE_VERIFY_KEY",
timeout,
"published_evidence",
publication_limit,
published_digests,
open_descriptors,
)
if not hmac.compare_digest(published["data"], payload):
raise RuntimeError("published evidence differs from signed private bytes")
strict_json_loads(payload)
print(json.dumps(evidence, sort_keys=True, allow_nan=False))
exit_codes = {
"PASS": 0,
"NOT_APPLICABLE": 0,
"FAIL": 1,
"INSUFFICIENT_DATA": 2,
"ERROR": 3,
}
for descriptor in reversed(open_descriptors):
os.close(descriptor)
raise SystemExit(exit_codes[evidence["status"]])
if __name__ == "__main__":
main()Reference evaluator prerequisites
The reference evaluator variant requires a Linux kernel that supports memfd_create(MFD_ALLOW_SEALING) and the write/grow/shrink/seal locks, mounted /proc/self/fd, Python 3.10 or newer, and a current supported Sigstore Cosign release. It opens every source component with descriptor-relative no-follow traversal, captures each top-level signed input and its bundle exactly once into separate kernel-sealed memfds, gives Cosign only those inherited descriptors, and strictly parses and hashes the same captured bytes. It never reopens a signed source pathname after signature verification. A platform lacking these prerequisites makes this implementation variant ERROR or requires an equivalently trusted immutable-snapshot runner; it never turns the D-018 control into PASS or NOT_APPLICABLE.
The policy pins each approved semantic case to its exact scenario version, payload digest, harness ID/version, and harness artifact digest. The evaluator loads each basename-only payload object through no-follow traversal and verifies its bytes. Every result repeats that binding plus immutable detector identity, production request ID, and production observation-event ID; the signed completeness receipt binds the canonical full replay-binding population. Each canonical production-observation object must match the result exactly. alert_event_id is required only for DETECTED and must be null for every other outcome; operational/error correlation belongs in production_event_id.
Executable signed-fixture and negative acceptance suite
# file: tests/test_d018_negative.py
from __future__ import annotations
import errno
import hashlib
import importlib.util
import json
import math
import os
import shutil
import subprocess
import sys
from pathlib import Path
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
import pytest
pytestmark = pytest.mark.skipif(
sys.platform != "linux",
reason="AID-D-018 requires Linux sealed memfd semantics",
)
EVALUATOR = Path(
os.environ.get(
"D018_EVALUATOR_PATH",
"assurance/evaluate_production_detectors.py",
)
).resolve()
SIGNED_INPUTS = {
"deployment_snapshot": Path("assurance/production-deployment-snapshot.json"),
"expected_deployment": Path("assurance/expected-detector-segment-manifest.json"),
"inventory": Path("assurance/detector-inventory.json"),
"scenarios": Path("assurance/scenarios.jsonl"),
"results": Path("assurance/production-path-results.jsonl"),
"result_evidence_manifest": Path("assurance/result-evidence-manifest.json"),
"receipt": Path("assurance/run-receipt.json"),
"policy": Path("policy/detection-efficacy-policy.json"),
}
VERIFY_ENV_BY_LABEL = {
"deployment_snapshot": "D018_DEPLOYMENT_SNAPSHOT_VERIFY_KEY",
"expected_deployment": "D018_EXPECTED_DEPLOYMENT_VERIFY_KEY",
"inventory": "D018_INVENTORY_VERIFY_KEY",
"scenarios": "D018_SCENARIO_VERIFY_KEY",
"results": "D018_RUNNER_VERIFY_KEY",
"result_evidence_manifest": "D018_RESULT_EVIDENCE_VERIFY_KEY",
"receipt": "D018_COMPLETENESS_VERIFY_KEY",
"policy": "D018_POLICY_VERIFY_KEY",
}
def required_positive_seconds(name: str) -> float:
raw = os.environ.get(name)
if raw is None or not raw.strip():
raise RuntimeError(
f"{name} must be injected from the signed CI runner profile"
)
try:
value = float(raw)
except ValueError as error:
raise RuntimeError(f"{name} must be numeric") from error
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"{name} must be finite and positive")
return value
TEST_PROCESS_TIMEOUT_SECONDS = required_positive_seconds(
"D018_TEST_PROCESS_TIMEOUT_SECONDS"
)
def canonical(value: object) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def sha256(payload: bytes) -> str:
return hashlib.sha256(payload).hexdigest()
def file_sha256(path: Path) -> str:
return sha256(path.read_bytes())
def write_json(path: Path, value: object) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(canonical(value))
def read_json(path: Path) -> dict:
value = json.loads(path.read_text(encoding="utf-8"))
assert isinstance(value, dict)
return value
def write_jsonl(path: Path, records: list[dict]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"" if not records else b"\n".join(map(canonical, records)) + b"\n")
def read_jsonl(path: Path) -> list[dict]:
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
def replay_binding_digest(results: list[dict]) -> str:
fields = (
"result_id", "scenario_id", "scenario_version", "payload_sha256",
"harness_id", "harness_version", "harness_artifact_sha256",
"detector_id", "production_request_id", "production_event_id", "observed_at",
)
bindings = sorted(
({field: result[field] for field in fields} for result in results),
key=lambda item: item["result_id"],
)
return sha256(canonical(bindings))
@pytest.fixture(scope="session")
def cosign() -> str:
executable = shutil.which("cosign")
if executable is None:
pytest.fail("cosign is required; the acceptance suite must not skip signatures")
return executable
@pytest.fixture(scope="session")
def signing_material(tmp_path_factory: pytest.TempPathFactory, cosign: str) -> dict:
root = tmp_path_factory.mktemp("d018-cosign")
prefix = root / "fixture"
public_test_fixture_value = "not-a-secret-public-cosign-test-fixture"
environment = {**os.environ, "COSIGN_PASSWORD": public_test_fixture_value}
subprocess.run(
[cosign, "generate-key-pair", "--output-key-prefix", str(prefix)],
check=True,
capture_output=True,
text=True,
env=environment,
timeout=TEST_PROCESS_TIMEOUT_SECONDS,
)
return {
"private": prefix.with_suffix(".key"),
"public": prefix.with_suffix(".pub"),
"public_test_fixture_value": public_test_fixture_value,
"cosign": cosign,
}
def sign(path: Path, material: dict) -> None:
bundle = Path(str(path) + ".sigstore.json")
bundle.unlink(missing_ok=True)
subprocess.run(
[
material["cosign"], "sign-blob", "--yes", "--use-signing-config=false",
"--key", str(material["private"]), "--bundle", str(bundle), str(path),
],
check=True,
capture_output=True,
text=True,
env={**os.environ, "COSIGN_PASSWORD": material["public_test_fixture_value"]},
timeout=TEST_PROCESS_TIMEOUT_SECONDS,
)
def sign_all(root: Path, material: dict) -> None:
for relative in SIGNED_INPUTS.values():
sign(root / relative, material)
sign(root / "assurance/runtime-profile.json", material)
def build_passing_fixture(root: Path, material: dict) -> None:
assurance = root / "assurance"
(assurance / "scenario-payloads").mkdir(parents=True)
(assurance / "result-evidence").mkdir(parents=True)
(root / "policy").mkdir(parents=True)
(root / "out").mkdir(parents=True)
detector_artifact = sha256(b"detector-artifact")
detector_config = sha256(b"detector-config")
# Test-only freshness values; production reads its signed assurance window.
test_window_lag_seconds = 1
test_window_duration_seconds = 60
window_end = datetime.now(timezone.utc) - timedelta(seconds=test_window_lag_seconds)
window_start = window_end - timedelta(seconds=test_window_duration_seconds)
window = {
"window_id": "fixture-window-1",
"window_start": window_start.isoformat(),
"window_end": window_end.isoformat(),
"maximum_future_skew_seconds": 5,
"maximum_evidence_age_seconds": 3600,
}
deployment_snapshot = "pending-fixture-snapshot"
harness_artifact = sha256(b"harness-artifact")
expected = {
"schema_version": "aidefend.d018-expected-deployment.v2",
"manifest_version": "fixture-1",
"environment_id": "fixture-production",
"deployment_snapshot_sha256": deployment_snapshot,
"detectors": [{
"detector_id": "detector-1",
"detector_type": "prompt-safety-classifier",
"version": "1.0.0",
"artifact_sha256": detector_artifact,
"config_sha256": detector_config,
"observation_point": "production-gateway",
"segments": ["global"],
"technical_prerequisites": ["production_path_attached"],
}],
}
snapshot = {
"schema_version": "aidefend.production-deployment-snapshot.v1",
"snapshot_id": "fixture-snapshot-1",
"environment_id": expected["environment_id"],
"detectors": expected["detectors"],
}
write_json(assurance / "production-deployment-snapshot.json", snapshot)
deployment_snapshot = file_sha256(assurance / "production-deployment-snapshot.json")
expected["deployment_snapshot_sha256"] = deployment_snapshot
write_json(assurance / "expected-detector-segment-manifest.json", expected)
inventory = {
"schema_version": "aidefend.detector-inventory.v2",
"inventory_version": "fixture-1",
"environment_id": "fixture-production",
"deployment_snapshot_sha256": deployment_snapshot,
"collection_complete": True,
"detectors": [{
**expected["detectors"][0],
"technical_scope": "ELIGIBLE",
"prerequisite_results": {"production_path_attached": True},
"not_applicable_reason": None,
}],
}
write_json(assurance / "detector-inventory.json", inventory)
payload_specs = (
("attack-001", "ATTACK", "prompt-injection", b"attack-payload", "DETECTED"),
("benign-001", "BENIGN", "benign-control", b"benign-payload", "NOT_DETECTED"),
)
scenarios = []
results = []
manifest_objects = []
for index, (case_id, truth, family, payload, outcome) in enumerate(payload_specs, start=1):
payload_object = f"{case_id}.bin"
(assurance / "scenario-payloads" / payload_object).write_bytes(payload)
scenario = {
"schema_version": "aidefend.detection-scenario.v2",
"scenario_id": f"scenario-{index}",
"scenario_version": "1.0.0",
"segment": "global",
"ground_truth": truth,
"attack_family": family,
"semantic_case_id": case_id,
"payload_sha256": sha256(payload),
"payload_object": payload_object,
"harness_id": "production-replay-harness",
"harness_version": "1.0.0",
"harness_artifact_sha256": harness_artifact,
}
scenarios.append(scenario)
result = {
"schema_version": "aidefend.production-detector-result.v2",
"result_id": f"result-{index}",
"scenario_id": scenario["scenario_id"],
"scenario_version": scenario["scenario_version"],
"payload_sha256": scenario["payload_sha256"],
"harness_id": scenario["harness_id"],
"harness_version": scenario["harness_version"],
"harness_artifact_sha256": harness_artifact,
"detector_id": "detector-1",
"detector_version": "1.0.0",
"detector_artifact_sha256": detector_artifact,
"detector_config_sha256": detector_config,
"observation_point": "production-gateway",
"production_request_id": f"request-{index}",
"production_event_id": f"observation-{index}",
"observed_at": (window_start + timedelta(seconds=index)).isoformat(),
"outcome": outcome,
"alert_event_id": f"alert-{index}" if outcome == "DETECTED" else None,
}
observation = {
"schema_version": "aidefend.d018-production-observation.v1",
**{key: value for key, value in result.items() if key != "schema_version"},
}
object_path = f"result-{index}.json"
observation_bytes = canonical(observation)
(assurance / "result-evidence" / object_path).write_bytes(observation_bytes)
result["evidence_sha256"] = sha256(observation_bytes)
results.append(result)
manifest_objects.append({
"result_id": result["result_id"],
"object_path": object_path,
"evidence_sha256": result["evidence_sha256"],
})
write_jsonl(assurance / "scenarios.jsonl", scenarios)
write_jsonl(assurance / "production-path-results.jsonl", results)
manifest = {
"schema_version": "aidefend.d018-result-evidence-manifest.v2",
"results_sha256": file_sha256(assurance / "production-path-results.jsonl"),
"objects": manifest_objects,
}
write_json(assurance / "result-evidence-manifest.json", manifest)
requirements = []
for scenario in scenarios:
requirements.append({
"semantic_case_id": scenario["semantic_case_id"],
"ground_truth": scenario["ground_truth"],
"attack_family": scenario["attack_family"],
"scenario_version": scenario["scenario_version"],
"payload_sha256": scenario["payload_sha256"],
"harness_id": scenario["harness_id"],
"harness_version": scenario["harness_version"],
"harness_artifact_sha256": scenario["harness_artifact_sha256"],
})
requirements.sort(
key=lambda item: (
item["semantic_case_id"], item["ground_truth"], item["attack_family"],
item["scenario_version"], item["payload_sha256"], item["harness_id"],
item["harness_version"], item["harness_artifact_sha256"],
)
)
policy = {
"schema_version": "aidefend.detection-efficacy-policy.v2",
"policy_version": "fixture-1",
"assurance_window": window,
"expected_deployment_sha256": file_sha256(
assurance / "expected-detector-segment-manifest.json"
),
"required_source_checkpoints": ["production-observation-stream"],
"required_scenarios_by_detector_segment": {
"detector-1": {"global": requirements},
},
"thresholds_by_detector": {
"detector-1": {"global": {
"minimum_attack_cases": 1,
"minimum_benign_cases": 1,
"minimum_recall": 1.0,
"minimum_precision": 1.0,
"maximum_false_positive_rate": 0.0,
}},
},
}
write_json(root / "policy/detection-efficacy-policy.json", policy)
profile = {
"schema_version": "aidefend.d018-runtime-profile.v1",
"profile_version": "fixture-1",
"command_timeout_seconds": 30,
"maximum_input_bytes": 1048576,
"maximum_object_bytes": 262144,
"maximum_records": 100,
"maximum_matrix_cells": 100,
}
write_json(assurance / "runtime-profile.json", profile)
receipt = {
"schema_version": "aidefend.d018-run-receipt.v4",
"execution_mode": "DEPLOYED_OBSERVATION_PATH",
"deployment_snapshot_sha256": deployment_snapshot,
"window_id": window["window_id"],
"window_start": window["window_start"],
"window_end": window["window_end"],
"observation_through": window["window_end"],
"expected_deployment_sha256": "",
"inventory_sha256": "",
"scenarios_sha256": "",
"results_sha256": "",
"result_evidence_manifest_sha256": "",
"policy_sha256": "",
"runtime_profile_sha256": "",
"scenario_count": 0,
"result_count": 0,
"runner_id": "fixture-production-runner",
"runner_version": "1.0.0",
"runner_artifact_sha256": sha256(b"fixture-runner-artifact"),
"source_checkpoints": {
"production-observation-stream": {
"first": 1,
"last": 2,
"expected_count": 2,
"observed_count": 2,
"gap_free": True,
"complete": True,
},
},
"replay_bindings_sha256": "",
"collection_complete": True,
}
write_json(assurance / "run-receipt.json", receipt)
refresh_receipt(root)
sign_all(root, material)
def refresh_receipt(root: Path) -> None:
assurance = root / "assurance"
path = assurance / "run-receipt.json"
receipt = read_json(path)
scenarios = read_jsonl(assurance / "scenarios.jsonl")
results = read_jsonl(assurance / "production-path-results.jsonl")
receipt.update({
"deployment_snapshot_sha256": file_sha256(assurance / "production-deployment-snapshot.json"),
"expected_deployment_sha256": file_sha256(
assurance / "expected-detector-segment-manifest.json"
),
"inventory_sha256": file_sha256(assurance / "detector-inventory.json"),
"scenarios_sha256": file_sha256(assurance / "scenarios.jsonl"),
"results_sha256": file_sha256(assurance / "production-path-results.jsonl"),
"result_evidence_manifest_sha256": file_sha256(
assurance / "result-evidence-manifest.json"
),
"policy_sha256": file_sha256(root / "policy/detection-efficacy-policy.json"),
"runtime_profile_sha256": file_sha256(assurance / "runtime-profile.json"),
"scenario_count": len(scenarios),
"result_count": len(results),
"replay_bindings_sha256": replay_binding_digest(results),
})
write_json(path, receipt)
def resign_with_receipt(root: Path, material: dict, labels: list[str]) -> None:
refresh_receipt(root)
for label in labels:
sign(root / SIGNED_INPUTS[label], material)
sign(root / SIGNED_INPUTS["receipt"], material)
@pytest.fixture(scope="session")
def passing_template(
tmp_path_factory: pytest.TempPathFactory,
signing_material: dict,
) -> Path:
if not EVALUATOR.is_file():
pytest.fail(f"save the evaluator code block first: {EVALUATOR}")
root = tmp_path_factory.mktemp("d018-passing")
build_passing_fixture(root, signing_material)
return root
@pytest.fixture
def case(tmp_path: Path, passing_template: Path) -> Path:
root = tmp_path / "case"
shutil.copytree(passing_template, root)
return root
def evaluator_environment(root: Path, material: dict) -> dict[str, str]:
public_key = str(material["public"])
environment = {
**os.environ,
"COSIGN_PASSWORD": material["public_test_fixture_value"],
"D018_BOOTSTRAP_COMMAND_TIMEOUT_SECONDS": "30",
"D018_RUNTIME_PROFILE_PATH": str(root / "assurance/runtime-profile.json"),
"D018_RUNTIME_PROFILE_BUNDLE_PATH": str(
root / "assurance/runtime-profile.json.sigstore.json"
),
"D018_RUNTIME_PROFILE_VERIFY_KEY": public_key,
"D018_EVIDENCE_PATH": str(root / "out/evidence.json"),
"D018_EVIDENCE_BUNDLE_PATH": str(root / "out/evidence.sigstore.json"),
"D018_EVIDENCE_SIGNING_KEY": material["private"].read_text(encoding="utf-8"),
"D018_EVIDENCE_VERIFY_KEY": public_key,
}
for variable in VERIFY_ENV_BY_LABEL.values():
environment[variable] = public_key
return environment
def run_case(root: Path, material: dict) -> tuple[subprocess.CompletedProcess, dict]:
evidence_path = root / "out/evidence.json"
bundle_path = root / "out/evidence.sigstore.json"
evidence_path.unlink(missing_ok=True)
bundle_path.unlink(missing_ok=True)
completed = subprocess.run(
[sys.executable, str(EVALUATOR)],
cwd=root,
env=evaluator_environment(root, material),
capture_output=True,
text=True,
timeout=TEST_PROCESS_TIMEOUT_SECONDS,
check=False,
)
assert evidence_path.is_file(), completed.stderr
assert bundle_path.is_file(), completed.stderr
subprocess.run(
[
material["cosign"], "verify-blob", "--key", str(material["public"]),
"--bundle", str(bundle_path), str(evidence_path),
],
check=True,
capture_output=True,
text=True,
timeout=TEST_PROCESS_TIMEOUT_SECONDS,
)
evidence = read_json(evidence_path)
assert evidence["status"] != "PASS" or completed.returncode == 0
return completed, evidence
def assert_outcome(
root: Path,
material: dict,
status: str,
exit_code: int,
) -> dict:
completed, evidence = run_case(root, material)
assert completed.returncode == exit_code, completed.stderr
assert evidence["status"] == status, evidence
return evidence
def load_evaluator_module():
spec = importlib.util.spec_from_file_location("d018_evaluator", EVALUATOR)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_passing_signed_fixture(case: Path, signing_material: dict) -> None:
completed, evidence = run_case(case, signing_material)
assert completed.returncode == 0, completed.stderr
assert evidence["status"] == "PASS"
@pytest.mark.parametrize("label", sorted(SIGNED_INPUTS))
def test_each_signed_input_resists_swap_restore_aba(
label: str,
case: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
module = load_evaluator_module()
monkeypatch.chdir(case)
monkeypatch.setenv(VERIFY_ENV_BY_LABEL[label], "fixture-public-key")
relative = SIGNED_INPUTS[label]
source = case / relative
bundle = Path(str(source) + ".sigstore.json")
original = source.read_bytes()
original_bundle = bundle.read_bytes()
reads = {"source": 0, "bundle": 0}
real_read = module.read_regular_no_follow
def counted_read(path: Path, maximum_bytes: int) -> bytes:
candidate = Path(path)
if candidate == relative:
reads["source"] += 1
if candidate == Path(str(relative) + ".sigstore.json"):
reads["bundle"] += 1
return real_read(candidate, maximum_bytes)
def fake_cosign(arguments: list[str], **kwargs) -> SimpleNamespace:
source.write_bytes(b"attacker replacement")
bundle.write_bytes(b"attacker bundle replacement")
try:
assert Path(arguments[-1]).read_bytes() == original
assert Path(arguments[arguments.index("--bundle") + 1]).read_bytes() == original_bundle
assert len(kwargs["pass_fds"]) == 2
finally:
source.write_bytes(original)
bundle.write_bytes(original_bundle)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(module, "read_regular_no_follow", counted_read)
monkeypatch.setattr(module.subprocess, "run", fake_cosign)
descriptors = []
try:
capture = module.capture_signed_input(
relative,
Path(str(relative) + ".sigstore.json"),
VERIFY_ENV_BY_LABEL[label],
30.0,
label,
1048576,
{},
descriptors,
)
assert capture["data"] == original
assert reads == {"source": 1, "bundle": 1}
for descriptor in descriptors:
with pytest.raises(OSError) as sealed_error:
os.write(descriptor, b"x")
assert sealed_error.value.errno == errno.EPERM
finally:
for descriptor in reversed(descriptors):
os.close(descriptor)
def test_runtime_profile_resists_swap_restore_aba(
case: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
module = load_evaluator_module()
monkeypatch.chdir(case)
profile = case / "assurance/runtime-profile.json"
bundle = Path(str(profile) + ".sigstore.json")
original = profile.read_bytes()
original_bundle = bundle.read_bytes()
monkeypatch.setenv("D018_RUNTIME_PROFILE_PATH", "assurance/runtime-profile.json")
monkeypatch.setenv(
"D018_RUNTIME_PROFILE_BUNDLE_PATH",
"assurance/runtime-profile.json.sigstore.json",
)
monkeypatch.setenv("D018_RUNTIME_PROFILE_VERIFY_KEY", "fixture-public-key")
def fake_cosign(arguments: list[str], **kwargs) -> SimpleNamespace:
profile.write_bytes(b"{}")
bundle.write_bytes(b"attacker")
try:
assert Path(arguments[-1]).read_bytes() == original
assert Path(arguments[arguments.index("--bundle") + 1]).read_bytes() == original_bundle
finally:
profile.write_bytes(original)
bundle.write_bytes(original_bundle)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(module.subprocess, "run", fake_cosign)
descriptors = []
try:
identity = {
"runtime_profile_version": None,
"runtime_profile_sha256": None,
"runtime_profile_status": "ABSENT",
}
loaded = module.load_runtime_profile(identity, 30.0, descriptors)
assert loaded["profile_version"] == "fixture-1"
assert identity["runtime_profile_status"] == "VERIFIED"
finally:
for descriptor in reversed(descriptors):
os.close(descriptor)
def test_missing_inventory_entry_is_insufficient(
case: Path,
signing_material: dict,
) -> None:
inventory_path = case / SIGNED_INPUTS["inventory"]
inventory = read_json(inventory_path)
inventory["detectors"] = []
write_json(inventory_path, inventory)
resign_with_receipt(case, signing_material, ["inventory"])
assert_outcome(case, signing_material, "INSUFFICIENT_DATA", 2)
def test_missing_scenario_is_insufficient(case: Path, signing_material: dict) -> None:
path = case / SIGNED_INPUTS["scenarios"]
write_jsonl(path, read_jsonl(path)[:-1])
resign_with_receipt(case, signing_material, ["scenarios"])
assert_outcome(case, signing_material, "INSUFFICIENT_DATA", 2)
def test_missing_result_cell_is_insufficient(case: Path, signing_material: dict) -> None:
results_path = case / SIGNED_INPUTS["results"]
results = read_jsonl(results_path)[:-1]
write_jsonl(results_path, results)
manifest_path = case / SIGNED_INPUTS["result_evidence_manifest"]
manifest = read_json(manifest_path)
manifest["results_sha256"] = file_sha256(results_path)
manifest["objects"] = manifest["objects"][:-1]
write_json(manifest_path, manifest)
resign_with_receipt(
case,
signing_material,
["results", "result_evidence_manifest"],
)
assert_outcome(case, signing_material, "INSUFFICIENT_DATA", 2)
def test_missing_prerequisite_observation_is_insufficient(
case: Path,
signing_material: dict,
) -> None:
path = case / SIGNED_INPUTS["inventory"]
inventory = read_json(path)
inventory["detectors"][0]["prerequisite_results"] = {}
write_json(path, inventory)
resign_with_receipt(case, signing_material, ["inventory"])
assert_outcome(case, signing_material, "INSUFFICIENT_DATA", 2)
def test_missing_evidence_object_is_insufficient(
case: Path,
signing_material: dict,
) -> None:
(case / "assurance/result-evidence/result-2.json").unlink()
assert_outcome(case, signing_material, "INSUFFICIENT_DATA", 2)
def test_missing_source_checkpoint_is_insufficient(
case: Path,
signing_material: dict,
) -> None:
path = case / SIGNED_INPUTS["receipt"]
receipt = read_json(path)
receipt["source_checkpoints"] = {}
write_json(path, receipt)
sign(path, signing_material)
assert_outcome(case, signing_material, "INSUFFICIENT_DATA", 2)
@pytest.mark.parametrize(
"malformed",
[
b"{",
b'{"schema_version":"x","schema_version":"y"}',
b'{"schema_version":NaN}',
b'{"schema_version":Infinity}',
],
)
def test_malformed_duplicate_and_nonfinite_json_are_error(
malformed: bytes,
case: Path,
signing_material: dict,
) -> None:
path = case / SIGNED_INPUTS["policy"]
path.write_bytes(malformed)
refresh_receipt(case)
sign(path, signing_material)
sign(case / SIGNED_INPUTS["receipt"], signing_material)
assert_outcome(case, signing_material, "ERROR", 3)
def test_bad_signature_is_error(case: Path, signing_material: dict) -> None:
path = case / SIGNED_INPUTS["inventory"]
path.write_bytes(path.read_bytes() + b" ")
assert_outcome(case, signing_material, "ERROR", 3)
def test_signed_digest_substitution_is_error(case: Path, signing_material: dict) -> None:
path = case / SIGNED_INPUTS["receipt"]
receipt = read_json(path)
receipt["inventory_sha256"] = "0" * 64
write_json(path, receipt)
sign(path, signing_material)
assert_outcome(case, signing_material, "ERROR", 3)
@pytest.mark.parametrize("duplicate_kind", ["scenario_id", "semantic_case", "payload"])
def test_duplicate_scenario_identity_is_error(
duplicate_kind: str,
case: Path,
signing_material: dict,
) -> None:
path = case / SIGNED_INPUTS["scenarios"]
scenarios = read_jsonl(path)
duplicate = dict(scenarios[0])
if duplicate_kind == "semantic_case":
duplicate["scenario_id"] = "scenario-duplicate"
elif duplicate_kind == "payload":
duplicate["scenario_id"] = "scenario-duplicate"
duplicate["semantic_case_id"] = "unique-semantic-case"
scenarios.append(duplicate)
write_jsonl(path, scenarios)
resign_with_receipt(case, signing_material, ["scenarios"])
assert_outcome(case, signing_material, "ERROR", 3)
def test_nested_parent_symlink_is_error(
case: Path,
signing_material: dict,
) -> None:
assurance = case / "assurance"
real_assurance = case / "assurance-real"
assurance.rename(real_assurance)
assurance.symlink_to(real_assurance, target_is_directory=True)
assert_outcome(case, signing_material, "ERROR", 3)
def test_final_entry_symlink_is_error(case: Path, signing_material: dict) -> None:
inventory = case / SIGNED_INPUTS["inventory"]
real_inventory = inventory.with_name("detector-inventory.real.json")
inventory.rename(real_inventory)
inventory.symlink_to(real_inventory.name)
assert_outcome(case, signing_material, "ERROR", 3)
@pytest.mark.parametrize("runtime_failure", ["missing", "invalid_schema", "bad_signature"])
def test_runtime_profile_failure_produces_verified_error(
runtime_failure: str,
case: Path,
signing_material: dict,
) -> None:
profile = case / "assurance/runtime-profile.json"
if runtime_failure == "missing":
profile.unlink()
elif runtime_failure == "invalid_schema":
write_json(profile, {"schema_version": "wrong"})
sign(profile, signing_material)
else:
profile.write_bytes(profile.read_bytes() + b" ")
evidence = assert_outcome(case, signing_material, "ERROR", 3)
assert evidence["runtime_profile_status"] in {
"ABSENT", "PRESENT_UNVERIFIED", "SIGNATURE_VERIFIED",
}
def test_payload_object_substitution_is_error(case: Path, signing_material: dict) -> None:
(case / "assurance/scenario-payloads/attack-001.bin").write_bytes(b"easier payload")
assert_outcome(case, signing_material, "ERROR", 3)
def test_replay_observation_binding_substitution_is_error(
case: Path,
signing_material: dict,
) -> None:
object_path = case / "assurance/result-evidence/result-1.json"
observation = read_json(object_path)
observation["production_event_id"] = "unbound-event"
write_json(object_path, observation)
assert_outcome(case, signing_material, "ERROR", 3)
def test_non_detected_alert_identifier_is_error(case: Path, signing_material: dict) -> None:
results_path = case / SIGNED_INPUTS["results"]
results = read_jsonl(results_path)
results[1]["alert_event_id"] = "ambiguous-alert"
write_jsonl(results_path, results)
resign_with_receipt(case, signing_material, ["results"])
assert_outcome(case, signing_material, "ERROR", 3)Independently verify results
Write the deployment snapshot, all signed inputs, and evidence to WORM storage, read each object back, and verify digests/signatures. An identity independent of detector owners and the first runner replays the same scenarios through the same deployed paths, reconciles the complete matrix, recalculates every segment metric, and compares the canonical evidence digest. Save the included tests/test_d018_negative.py code block beside the evaluator and run its fully signed passing and negative fixtures with pytest -q tests/test_d018_negative.py. It must assert: nested object paths and final-entry symlinks are ERROR and cannot win a path race; duplicate JSON keys and NaN/Infinity are ERROR; an authentically re-signed deletion of an expected inventory entry, technical-prerequisite observation, scenario, result cell, evidence object, or source checkpoint is INSUFFICIENT_DATA with exit code 2; malformed, bad-signature, or digest-substituted inputs are ERROR; duplicate scenario IDs, semantic cases, or payload digests are ERROR; and missing or invalid runtime profiles still produce signature-verified ERROR evidence. The suite must verify every output bundle and fail if any negative fixture produces PASS. Publish a distinct evidence reference for each specialized owner so D-018 does not double-count H-026, H-036, D-008, M-008, or H-007.004 outcomes.
Action: Run this gate for every policy-approved detector release and scheduled production assurance window; block promotion or open a remediation finding unless the exact complete signed matrix produces PASS, and retain a distinct evidence reference for this D-018 claim.
Open-source tools
- garak
- PyRIT
- promptfoo
- pytest
- OpenTelemetry
- Cosign
Commercial / hosted tools
- Splunk Enterprise Security
- Microsoft Sentinel
- Google Security Operations
- Datadog Cloud SIEM
Threat mappings
MITRE ATLAS
- AML.T0015 Evade AI Model
MAESTRO
- Evasion of Detection (L5)
OWASP LLM Top 10 2026
- N/A
OWASP ML Top 10 2023
- ML01:2023 Input Manipulation Attack
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- NISTAML.022 Evasion
- NISTAML.025 Black-box Evasion
Cisco Integrated AI Security and Safety Framework
- AITech-9.2 Detection Evasion (production-path replay exposes missed attacks and detector scenario-coverage gaps)
Google Secure AI Framework 2.0 - Risks
- MEV: Model Evasion
Databricks AI Security Framework 3.0
- Model Serving - Inference response 10.1: Lack of audit and monitoring inference quality