AID-D-017: AI Detection SLO & Alert-Coverage Measurement
Establish measurable detection service-level objectives for AI and agentic systems by tracking how quickly observable anomalies become alerts, how quickly alerts are acknowledged and triaged, and how much of the alert queue is investigated across agent sessions, tool calls, memory writes, retrieval events, inference gateways, and high-risk actions.
Primary metrics:
Primary metrics:
detection_latency measures anomaly observed to alert created; acknowledgement_latency measures alert created to human acknowledged; triage_latency measures human acknowledged to triage completed; investigation_ratio measures triage-completed non-duplicate eligible alerts divided by the complete non-duplicate eligible population.Pillars: infra, app
Lifecycle: operation, validation, improvement
Scope boundary
Owns end-to-end detection service-level measurement for anomaly-to-alert latency, alert acknowledgement, triage completion, and investigation coverage over complete bounded alert cohorts. It does not own individual alert semantics, telemetry collection, or posture SLOs.
AID-D-005.010: AID-D-017 measures service-level speed and investigation coverage from observation through triage; AID-D-005.010 measures whether sensors and the telemetry pipeline remain healthy and complete. Healthy telemetry is necessary but does not prove alerts are timely, acknowledged, or investigated.
AID-D-018: AID-D-017 measures operational latency and alert-queue handling coverage; AID-D-018 measures scenario-level recall, precision, false positives, errors, abstentions, and detector coverage. Fast triage of an alert does not prove the detector finds the right attacks, and high recall does not prove timely operations.
AID-M-005.002: AID-D-017 owns detection and investigation SLOs; AID-M-005.002 owns secure-configuration posture criteria and build or release gates. Detection service quality and configuration posture are separate operational objectives.
Implementation guidance
- AID-D-017-G001: Measure detection, acknowledgement, and triage latency plus investigation coverage from a complete signed lifecycle population with an explicit observation horizon, binding the result to the exact asset, rules, sources, and policy.
- AID-D-017-G002: Compute policy-defined dwell-time, triage-time, and investigation-coverage metric series for complete, explicitly bounded AI-security alert cohorts.
- AID-D-017-G003: Package each detection-SLO result with the signed cohort, metric, policy, raw-event digests, and replay instructions needed to recompute it without querying the producing pipeline.
AID-D-017-G001
Measure detection, acknowledgement, and triage latency plus investigation coverage from a complete signed lifecycle population with an explicit observation horizon, binding the result to the exact asset, rules, sources, and policy.
When and where to run this detector
Apply this detector to every AI-security alert route declared by a signed rule and source inventory. Detection engineering owns the lifecycle schema and SLO policy; the alerting platform owner emits source-authenticated records; an assurance identity evaluates the complete fixed window after it closes. Bind each record and result to the exact AI asset/version, alert rule ID/version/configuration digest/schema digest, source checkpoints, policy, and complete observed population. A signed inventory proving no AI-security alert route is deployed yields NOT_APPLICABLE; an uninstrumented or missing route is INSUFFICIENT_DATA, never a pass.
Exact lifecycle-record schema
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://aidefend.net/schema/ai-detection-lifecycle-record-v2.json",
"type": "object",
"required": [
"schema_version", "event_id", "source_id", "source_sequence",
"asset_id", "asset_version", "alert_id", "rule_id", "rule_version",
"rule_configuration_sha256", "rule_schema_sha256",
"anomaly_observed_at", "alert_created_at", "human_acknowledged_at",
"triage_completed_at", "investigated", "disposition",
"duplicate_of_alert_id", "severity"
],
"properties": {
"schema_version": {"const": "aidefend.ai_detection_lifecycle.v2"},
"event_id": {"type": "string", "minLength": 8},
"source_id": {"type": "string", "minLength": 1},
"source_sequence": {"type": "integer", "minimum": 0},
"asset_id": {"type": "string", "minLength": 1},
"asset_version": {"type": "string", "minLength": 1},
"alert_id": {"type": "string", "minLength": 1},
"rule_id": {"type": "string", "minLength": 1},
"rule_version": {"type": "string", "minLength": 1},
"rule_configuration_sha256": {
"type": "string", "pattern": "^[0-9a-f]{64}$"
},
"rule_schema_sha256": {
"type": "string", "pattern": "^[0-9a-f]{64}$"
},
"agent_id": {"type": ["string", "null"]},
"tool_id": {"type": ["string", "null"]},
"session_id": {"type": ["string", "null"]},
"tenant_id": {"type": ["string", "null"]},
"anomaly_observed_at": {"type": "string", "format": "date-time"},
"alert_created_at": {"type": "string", "format": "date-time"},
"human_acknowledged_at": {"type": ["string", "null"], "format": "date-time"},
"triage_completed_at": {"type": ["string", "null"], "format": "date-time"},
"investigated": {"type": "boolean"},
"disposition": {
"enum": [
"true_positive", "false_positive", "benign",
"unknown", "duplicate", "not_reviewed"
]
},
"duplicate_of_alert_id": {"type": ["string", "null"]},
"severity": {
"enum": ["critical", "high", "medium", "low", "informational"]
}
},
"additionalProperties": false
}
Validate schema and lifecycle semantics
from __future__ import annotations
# File: detection/validate_lifecycle_event.py
import json
from datetime import datetime, timezone
from jsonschema import Draft202012Validator, FormatChecker
def reject_constant(value: str):
raise ValueError("non-finite JSON value is forbidden: " + value)
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError("duplicate JSON key: " + key)
result[key] = value
return result
def strict_json_object(raw: bytes, label: str) -> dict:
value = json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=reject_constant,
)
if not isinstance(value, dict):
raise ValueError(label + " must be a JSON object")
return value
def parse_time(value: str | None) -> datetime | None:
if value is None:
return None
if not isinstance(value, str):
raise ValueError("lifecycle timestamp is not text")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("lifecycle timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
class LifecycleValidator:
"""Build one validator from the exact schema bytes verified for this run."""
def __init__(self, verified_schema_raw: bytes):
schema = strict_json_object(verified_schema_raw, "lifecycle schema")
Draft202012Validator.check_schema(schema)
self._validator = Draft202012Validator(
schema, format_checker=FormatChecker()
)
def validate(self, event: dict) -> dict:
if not isinstance(event, dict):
raise ValueError("lifecycle record must be a JSON object")
errors = sorted(
self._validator.iter_errors(event),
key=lambda error: [str(value) for value in error.path],
)
if errors:
detail = "; ".join(
".".join(str(value) for value in error.path)
+ ": " + error.message
for error in errors
)
raise ValueError("invalid lifecycle record: " + detail)
observed = parse_time(event["anomaly_observed_at"])
created = parse_time(event["alert_created_at"])
acknowledged = parse_time(event["human_acknowledged_at"])
triaged = parse_time(event["triage_completed_at"])
if observed is None or created is None or created < observed:
raise ValueError("alert creation precedes anomaly observation")
if acknowledged is not None and acknowledged < created:
raise ValueError("human acknowledgement precedes alert creation")
if triaged is not None and (
acknowledged is None or triaged < acknowledged
):
raise ValueError("triage completion lacks or precedes acknowledgement")
if event["investigated"] != (triaged is not None):
raise ValueError("investigated flag disagrees with triage completion")
if not event["investigated"] and event["disposition"] != "not_reviewed":
raise ValueError("uninvestigated alert has a reviewed disposition")
if (
(event["disposition"] == "duplicate")
!= isinstance(event["duplicate_of_alert_id"], str)
):
raise ValueError("duplicate disposition and reference disagree")
return event
Reject malformed records into a separately monitored dead-letter queue and count them against collection completeness. Never clamp negative durations or synthesize timestamps.
Compute reproducible fixed-window metrics
from __future__ import annotations
# File: detection/slo_metrics.py
import math
from datetime import datetime, timezone
def parse_ts(value: str | None) -> datetime | None:
if value is None:
return None
if not isinstance(value, str):
raise ValueError("metric timestamp is not text")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("metric timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
def seconds_between(start: str | None, end: str | None) -> float | None:
start_time, end_time = parse_ts(start), parse_ts(end)
if start_time is None or end_time is None:
return None
seconds = (end_time - start_time).total_seconds()
if seconds < 0 or not math.isfinite(seconds):
raise ValueError("invalid lifecycle duration")
return seconds
def nearest_rank(values: list[float], percentile: int) -> float | None:
if not values:
return None
if not 0 < percentile <= 100:
raise ValueError("percentile is outside (0,100]")
ordered = sorted(values)
rank = max(1, (percentile * len(ordered) + 99) // 100)
return ordered[rank - 1]
def positive_seconds(value: object, label: str) -> float:
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or not math.isfinite(float(value))
or float(value) <= 0
):
raise ValueError(label + " must be finite and positive")
return float(value)
def compute_detection_slo(
events: list[dict],
*,
observation_horizon_end: datetime,
acknowledgement_deadline_seconds: float,
triage_deadline_seconds: float,
) -> dict:
if observation_horizon_end.utcoffset() is None:
raise ValueError("observation horizon lacks timezone")
horizon = observation_horizon_end.astimezone(timezone.utc)
acknowledgement_deadline = positive_seconds(
acknowledgement_deadline_seconds, "acknowledgement deadline"
)
triage_deadline = positive_seconds(
triage_deadline_seconds, "triage deadline"
)
eligible = [event for event in events if event["disposition"] != "duplicate"]
detection_seconds = []
acknowledgement_seconds = []
triage_seconds = []
acknowledgement_deadline_breaches = 0
triage_deadline_breaches = 0
unacknowledged_mature_alerts = 0
untriaged_mature_alerts = 0
acknowledgement_right_censored_alerts = 0
triage_right_censored_alerts = 0
triage_blocked_by_unacknowledged_alerts = 0
for event in eligible:
created = parse_ts(event["alert_created_at"])
acknowledged = parse_ts(event["human_acknowledged_at"])
triaged = parse_ts(event["triage_completed_at"])
detection = seconds_between(
event["anomaly_observed_at"], event["alert_created_at"]
)
if created is None or detection is None or created > horizon:
raise ValueError("alert lifecycle exceeds the observation horizon")
detection_seconds.append(detection)
if acknowledged is None:
triage_blocked_by_unacknowledged_alerts += 1
elapsed = (horizon - created).total_seconds()
if elapsed >= acknowledgement_deadline:
unacknowledged_mature_alerts += 1
else:
acknowledgement_right_censored_alerts += 1
continue
acknowledgement = seconds_between(
event["alert_created_at"], event["human_acknowledged_at"]
)
if acknowledgement is None or acknowledged > horizon:
raise ValueError("acknowledgement exceeds the observation horizon")
acknowledgement_seconds.append(acknowledgement)
if acknowledgement > acknowledgement_deadline:
acknowledgement_deadline_breaches += 1
if triaged is None:
elapsed = (horizon - acknowledged).total_seconds()
if elapsed >= triage_deadline:
untriaged_mature_alerts += 1
else:
triage_right_censored_alerts += 1
continue
triage = seconds_between(
event["human_acknowledged_at"], event["triage_completed_at"]
)
if triage is None or triaged > horizon:
raise ValueError("triage completion exceeds the observation horizon")
triage_seconds.append(triage)
if triage > triage_deadline:
triage_deadline_breaches += 1
coverage = len(triage_seconds) / len(eligible) if eligible else None
return {
"eligible_alerts": len(eligible),
"duplicate_alerts_excluded": len(events) - len(eligible),
"detection_sample_count": len(detection_seconds),
"acknowledgement_sample_count": len(acknowledgement_seconds),
"triage_sample_count": len(triage_seconds),
"investigation_ratio": round(coverage, 6) if coverage is not None else None,
"p50_detection_seconds": nearest_rank(detection_seconds, 50),
"p95_detection_seconds": nearest_rank(detection_seconds, 95),
"p50_acknowledgement_seconds": nearest_rank(acknowledgement_seconds, 50),
"p95_acknowledgement_seconds": nearest_rank(acknowledgement_seconds, 95),
"p50_triage_seconds": nearest_rank(triage_seconds, 50),
"p95_triage_seconds": nearest_rank(triage_seconds, 95),
"acknowledgement_deadline_breaches": acknowledgement_deadline_breaches,
"triage_deadline_breaches": triage_deadline_breaches,
"unacknowledged_mature_alerts": unacknowledged_mature_alerts,
"untriaged_mature_alerts": untriaged_mature_alerts,
"acknowledgement_right_censored_alerts": acknowledgement_right_censored_alerts,
"triage_right_censored_alerts": triage_right_censored_alerts,
"triage_blocked_by_unacknowledged_alerts": triage_blocked_by_unacknowledged_alerts,
"percentile_method": "nearest_rank",
}
Operational notes: Break metrics down by rule_id, severity, agent_id, and tenant_id. Keep detection, acknowledgement, and post-acknowledgement triage populations separate. A mature missing acknowledgement or triage is a deadline breach; a lifecycle that has not reached its signed deadline is right-censored and prevents PASS. Record every sample count, censor count, deadline-breach count, and the nearest_rank percentile method so an independent replay uses the same complete population.
Concept:
A detection SLO cannot be calculated from an arbitrary caller-supplied list. Use a collector-signed JSONL snapshot, an independently signed completeness receipt that binds the expected source IDs, observation horizon, counts, and checkpoints, plus separately signed schema, rule inventory, and SLO policy artifacts. The evaluator captures and verifies the exact bytes before constructing the validator or calculating metrics.
Signed input contracts
{
"schema_version": "aidefend.d017-collection-receipt.v2",
"window_start": "2026-07-12T00:00:00Z",
"window_end": "2026-07-13T00:00:00Z",
"observation_horizon_end": "2026-07-13T00:30:00Z",
"emitted_at": "2026-07-13T00:30:05Z",
"subject": {
"asset_id": "support-agent-prod",
"asset_version": "2026.07.12.1"
},
"expected_rule_ids": ["agent-tool-misuse-v4", "prompt-injection-v8"],
"expected_source_ids": ["agent-runtime", "inference-gateway", "soar"],
"event_count": 1842,
"source_counts": {
"agent-runtime": 920,
"inference-gateway": 712,
"soar": 210
},
"source_checkpoints": {
"agent-runtime": {"first": 1001, "last": 1920, "complete": true},
"inference-gateway": {"first": 5001, "last": 5712, "complete": true},
"soar": {"first": 901, "last": 1110, "complete": true}
},
"collection_complete": true,
"events_sha256": "13db55d70560f4403fd7d843d7b2c7694346941b4cea142faca52528dee48586",
"schema_sha256": "b24d08f80684f70f77600e835bc4ba8ef9d5ce3a80df1f7eebcc02fad342b162",
"rule_inventory_sha256": "fa8fa7f2f5e20d522e5c6ec6fb200dfbfb9c9da1300e2c26041b7c1800fc5adf"
}The telemetry archive signs the exact JSONL snapshot. A separately authorized completeness service signs this receipt only after comparing the export to the expected source inventory and sequence/checkpoint ledger, including the follow-up observation horizon. The schema owner, rule-inventory owner, and SLO owner sign their own artifacts with distinct root-pinned verification keys. The signed aidefend.ai-security-alert-rule-inventory.v2 artifact contains a sorted, unique rules array; every entry binds rule_id, rule_version, rule_configuration_sha256, and rule_schema_sha256, while the receipt binds the exact inventory bytes through rule_inventory_sha256. Provision /var/lib/aidefend/d017/results for the dedicated evaluator identity before running the job; callers do not supply trust keys, result paths, or evaluation time.
Capture, calculate, and publish the result
from __future__ import annotations
# File: detection/build_verified_slo_result.py
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import ctypes
import errno
import hashlib
import hmac
import json
import math
import os
from pathlib import Path
import re
import shutil
import stat
import subprocess
import tempfile
from detection.slo_metrics import compute_detection_slo
from detection.validate_lifecycle_event import (
LifecycleValidator, strict_json_object,
)
COSIGN = "/usr/local/bin/cosign"
# These fixed process-safety ceilings are not policy thresholds; size ceilings are bytes, and neither a
# signed policy nor a caller can raise them.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
HARD_EVENTS_CEILING = 256 * 1024 * 1024
HARD_METADATA_CEILING = 4 * 1024 * 1024
HARD_KEY_CEILING = 1024 * 1024
EVENTS = Path("/var/lib/aidefend/d017/input/lifecycle-events.jsonl")
EVENTS_BUNDLE = Path("/var/lib/aidefend/d017/input/lifecycle-events.sigstore.json")
COLLECTION = Path("/var/lib/aidefend/d017/input/collection-receipt.json")
COLLECTION_BUNDLE = Path("/var/lib/aidefend/d017/input/collection-receipt.sigstore.json")
POLICY = Path("/etc/aidefend/d017/policy/detection-slo-policy.json")
POLICY_BUNDLE = Path("/etc/aidefend/d017/policy/detection-slo-policy.sigstore.json")
SCHEMA = Path("/etc/aidefend/d017/schema/ai-detection-lifecycle-event.schema.json")
SCHEMA_BUNDLE = Path("/etc/aidefend/d017/schema/ai-detection-lifecycle-event.schema.sigstore.json")
RULE_INVENTORY = Path("/etc/aidefend/d017/policy/ai-security-alert-rule-inventory.json")
RULE_INVENTORY_BUNDLE = Path("/etc/aidefend/d017/policy/ai-security-alert-rule-inventory.sigstore.json")
COLLECTOR_KEY = Path("/etc/aidefend/d017/trust/lifecycle-collector.pub")
COMPLETENESS_KEY = Path("/etc/aidefend/d017/trust/completeness-service.pub")
POLICY_KEY = Path("/etc/aidefend/d017/trust/slo-policy-owner.pub")
SCHEMA_KEY = Path("/etc/aidefend/d017/trust/schema-owner.pub")
RULE_INVENTORY_KEY = Path("/etc/aidefend/d017/trust/rule-inventory-owner.pub")
RESULT_KEY = Path("/etc/aidefend/d017/trust/slo-result.pub")
RESULT_BASE = Path("/var/lib/aidefend/d017/results")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
AT_FDCWD = -100
RENAME_NOREPLACE = 1
EXIT_CODES = {
"PASS": 0, "FAIL": 1, "INSUFFICIENT_DATA": 2,
"NOT_APPLICABLE": 0, "ERROR": 3,
}
COLLECTION_FIELDS = {
"schema_version", "window_start", "window_end",
"observation_horizon_end", "emitted_at", "subject",
"expected_rule_ids", "expected_source_ids", "event_count",
"source_counts", "source_checkpoints", "collection_complete",
"events_sha256", "schema_sha256", "rule_inventory_sha256",
}
POLICY_FIELDS = {
"schema_version", "policy_version",
"max_p95_detection_seconds", "max_p95_acknowledgement_seconds",
"max_p95_triage_seconds", "acknowledgement_deadline_seconds",
"triage_deadline_seconds", "maximum_evidence_age_seconds",
"maximum_future_skew_seconds", "percentile_method",
"retention_policy_sha256",
}
RULE_BINDING_FIELDS = {
"rule_id", "rule_version",
"rule_configuration_sha256", "rule_schema_sha256",
}
@dataclass(frozen=True)
class Captured:
raw: bytes
bundle: bytes
key: bytes
def evidence(self) -> dict:
return {
"payload_sha256": hashlib.sha256(self.raw).hexdigest(),
"bundle_sha256": hashlib.sha256(self.bundle).hexdigest(),
"verification_key_sha256": hashlib.sha256(self.key).hexdigest(),
}
def canonical(value: object) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
def parse_time(value: object) -> datetime:
if not isinstance(value, str):
raise ValueError("timestamp is not text")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
def integer(value: object, label: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise ValueError(label + " must be a non-negative integer")
return value
def positive_number(value: object, label: str) -> float:
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or not math.isfinite(float(value))
or float(value) <= 0
):
raise ValueError(label + " must be finite and positive")
return float(value)
def ordered_unique_strings(value: object, label: str) -> list[str]:
if (
not isinstance(value, list)
or any(not isinstance(item, str) or not item for item in value)
or value != sorted(set(value))
):
raise ValueError(label + " must be a sorted unique string list")
return value
def read_stable(
path: Path, *, maximum_bytes: int, allow_empty: bool = False,
trust_root: bool = False,
) -> bytes:
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or before.st_size < 0
or before.st_size > maximum_bytes
or (before.st_size == 0 and not allow_empty)
):
raise ValueError("input is not a bounded regular file: " + str(path))
if trust_root and (before.st_uid != 0 or before.st_mode & 0o022):
raise PermissionError(
"verification key must be root-owned and non-writable: " + str(path)
)
chunks = []
total = 0
while True:
block = os.read(descriptor, min(1024 * 1024, maximum_bytes + 1 - total))
if not block:
break
chunks.append(block)
total += len(block)
if total > maximum_bytes:
raise ValueError("input exceeds its byte limit: " + str(path))
after = os.fstat(descriptor)
finally:
os.close(descriptor)
identity = lambda item: (
item.st_dev, item.st_ino, item.st_size,
item.st_mtime_ns, item.st_ctime_ns,
)
raw = b"".join(chunks)
if identity(before) != identity(after) or len(raw) != before.st_size:
raise RuntimeError("input changed during capture: " + str(path))
return raw
def write_exclusive(path: Path, raw: bytes, mode: int = 0o600) -> None:
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0),
mode,
)
with os.fdopen(descriptor, "wb") as stream:
stream.write(raw)
stream.flush()
os.fsync(stream.fileno())
def verify_bytes(label: str, raw: bytes, bundle: bytes, key: bytes) -> None:
with tempfile.TemporaryDirectory(prefix="aidefend-d017-verify-") as name:
root = Path(name)
os.chmod(root, 0o700)
payload = root / "payload"
signature = root / "payload.sigstore.json"
verify_key = root / "verify.pub"
for path, value in ((payload, raw), (signature, bundle), (verify_key, key)):
write_exclusive(path, value)
subprocess.run(
[
COSIGN, "verify-blob", "--key", str(verify_key),
"--bundle", str(signature), str(payload),
],
check=True, capture_output=True, text=True,
timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
for path, expected in ((payload, raw), (signature, bundle), (verify_key, key)):
actual = read_stable(
path, maximum_bytes=max(len(expected), 1),
allow_empty=(path == payload and not expected),
)
if not hmac.compare_digest(actual, expected):
raise RuntimeError(label + " private verification snapshot changed")
def capture_signed(
label: str, payload_path: Path, bundle_path: Path, key_path: Path,
*, maximum_bytes: int, allow_empty: bool = False,
) -> Captured:
raw = read_stable(
payload_path, maximum_bytes=maximum_bytes, allow_empty=allow_empty
)
bundle = read_stable(bundle_path, maximum_bytes=HARD_METADATA_CEILING)
key = read_stable(
key_path, maximum_bytes=HARD_KEY_CEILING, trust_root=True
)
verify_bytes(label, raw, bundle, key)
return Captured(raw, bundle, key)
def strict_jsonl(raw: bytes, validator: LifecycleValidator) -> list[dict]:
if not raw:
return []
if not raw.endswith(b"\n"):
raise ValueError("lifecycle JSONL must end with a newline")
lines = raw.splitlines()
if any(not line.strip() for line in lines):
raise ValueError("lifecycle JSONL contains a blank record")
return [
validator.validate(strict_json_object(line, "lifecycle event"))
for line in lines
]
def load_policy(captured: Captured) -> dict:
policy = strict_json_object(captured.raw, "detection SLO policy")
if (
set(policy) != POLICY_FIELDS
or policy.get("schema_version") != "aidefend.d017-slo-policy.v2"
or policy.get("percentile_method") != "nearest_rank"
or not isinstance(policy.get("policy_version"), str)
or not policy["policy_version"]
or not SHA256.fullmatch(str(policy.get("retention_policy_sha256")))
):
raise ValueError("detection SLO policy schema or identity differs")
for field in (
"max_p95_detection_seconds", "max_p95_acknowledgement_seconds",
"max_p95_triage_seconds", "acknowledgement_deadline_seconds",
"triage_deadline_seconds", "maximum_evidence_age_seconds",
):
positive_number(policy[field], field)
integer(policy["maximum_future_skew_seconds"], "maximum_future_skew_seconds")
return policy
def verified_inputs(evaluated_at: datetime) -> dict:
captures = {
"events": capture_signed(
"lifecycle events", EVENTS, EVENTS_BUNDLE, COLLECTOR_KEY,
maximum_bytes=HARD_EVENTS_CEILING, allow_empty=True,
),
"collection": capture_signed(
"collection receipt", COLLECTION, COLLECTION_BUNDLE,
COMPLETENESS_KEY, maximum_bytes=HARD_METADATA_CEILING,
),
"policy": capture_signed(
"SLO policy", POLICY, POLICY_BUNDLE, POLICY_KEY,
maximum_bytes=HARD_METADATA_CEILING,
),
"schema": capture_signed(
"lifecycle schema", SCHEMA, SCHEMA_BUNDLE, SCHEMA_KEY,
maximum_bytes=HARD_METADATA_CEILING,
),
"rule_inventory": capture_signed(
"rule inventory", RULE_INVENTORY, RULE_INVENTORY_BUNDLE,
RULE_INVENTORY_KEY, maximum_bytes=HARD_METADATA_CEILING,
),
}
key_digests = [
captured.evidence()["verification_key_sha256"]
for captured in captures.values()
]
if len(key_digests) != len(set(key_digests)):
raise ValueError("independent D017 input roles share a verification key")
collection = strict_json_object(
captures["collection"].raw, "collection receipt"
)
policy = load_policy(captures["policy"])
inventory = strict_json_object(
captures["rule_inventory"].raw, "rule inventory"
)
if (
set(collection) != COLLECTION_FIELDS
or collection.get("schema_version")
!= "aidefend.d017-collection-receipt.v2"
or set(collection.get("subject", {})) != {"asset_id", "asset_version"}
or any(
not isinstance(collection["subject"].get(field), str)
or not collection["subject"][field]
for field in ("asset_id", "asset_version")
)
or not isinstance(collection.get("collection_complete"), bool)
):
raise ValueError("collection receipt schema or subject differs")
expected_rules = ordered_unique_strings(
collection["expected_rule_ids"], "expected_rule_ids"
)
expected_sources = ordered_unique_strings(
collection["expected_source_ids"], "expected_source_ids"
)
integer(collection["event_count"], "event_count")
if (
not isinstance(collection["source_counts"], dict)
or set(collection["source_counts"]) != set(expected_sources)
or not isinstance(collection["source_checkpoints"], dict)
or set(collection["source_checkpoints"]) != set(expected_sources)
):
raise ValueError("signed expected source population is not exact")
for source, count in collection["source_counts"].items():
integer(count, "source count for " + source)
for field, captured_name in (
("events_sha256", "events"),
("schema_sha256", "schema"),
("rule_inventory_sha256", "rule_inventory"),
):
expected = str(collection[field])
actual = captures[captured_name].evidence()["payload_sha256"]
if not SHA256.fullmatch(expected) or not hmac.compare_digest(expected, actual):
raise ValueError(field + " does not bind the captured bytes")
inventory_fields = {
"schema_version", "asset_id", "asset_version", "rules",
}
if (
set(inventory) != inventory_fields
or inventory.get("schema_version")
!= "aidefend.ai-security-alert-rule-inventory.v2"
or inventory.get("asset_id") != collection["subject"]["asset_id"]
or inventory.get("asset_version") != collection["subject"]["asset_version"]
or not isinstance(inventory.get("rules"), list)
):
raise ValueError("signed rule inventory is invalid or misbound")
rule_bindings: dict[str, dict] = {}
for position, rule in enumerate(inventory["rules"]):
label = "rule inventory entry " + str(position)
if not isinstance(rule, dict) or set(rule) != RULE_BINDING_FIELDS:
raise ValueError(label + " has an invalid field set")
rule_id = rule.get("rule_id")
rule_version = rule.get("rule_version")
if (
not isinstance(rule_id, str) or not rule_id
or not isinstance(rule_version, str) or not rule_version
):
raise ValueError(label + " has an invalid identity or version")
for digest_field in (
"rule_configuration_sha256", "rule_schema_sha256",
):
digest = rule.get(digest_field)
if not isinstance(digest, str) or not SHA256.fullmatch(digest):
raise ValueError(label + " has an invalid " + digest_field)
if rule_id in rule_bindings:
raise ValueError("duplicate rule ID in signed inventory: " + rule_id)
rule_bindings[rule_id] = rule
if list(rule_bindings) != expected_rules:
raise ValueError("signed rule inventory order or population differs")
validator = LifecycleValidator(captures["schema"].raw)
events = strict_jsonl(captures["events"].raw, validator)
if collection["event_count"] != len(events):
raise ValueError("event count differs from the signed receipt")
if len({event["event_id"] for event in events}) != len(events):
raise ValueError("duplicate event_id in lifecycle snapshot")
if len({event["alert_id"] for event in events}) != len(events):
raise ValueError("duplicate alert_id in lifecycle snapshot")
if len({
(event["source_id"], event["source_sequence"]) for event in events
}) != len(events):
raise ValueError("duplicate source sequence in lifecycle snapshot")
expected_source_set = set(expected_sources)
for event in events:
binding = rule_bindings.get(event["rule_id"])
if (
event["asset_id"] != collection["subject"]["asset_id"]
or event["asset_version"] != collection["subject"]["asset_version"]
or event["source_id"] not in expected_source_set
):
raise ValueError("event identity is outside a signed expected population")
event_configuration_sha = event.get("rule_configuration_sha256")
event_schema_sha = event.get("rule_schema_sha256")
if (
binding is None
or event.get("rule_version") != binding["rule_version"]
or not isinstance(event_configuration_sha, str)
or not isinstance(event_schema_sha, str)
or not hmac.compare_digest(
event_configuration_sha, binding["rule_configuration_sha256"]
)
or not hmac.compare_digest(
event_schema_sha, binding["rule_schema_sha256"]
)
):
raise ValueError("event rule binding differs from the signed inventory")
events_by_alert_id = {event["alert_id"]: event for event in events}
for event in events:
if event["disposition"] != "duplicate":
continue
referenced_id = event["duplicate_of_alert_id"]
referenced = events_by_alert_id.get(referenced_id)
if (
referenced_id == event["alert_id"]
or referenced is None
or referenced["disposition"] == "duplicate"
):
raise ValueError(
"duplicate must reference one canonical non-duplicate alert"
)
observed_counts = {
source: sum(1 for event in events if event["source_id"] == source)
for source in expected_sources
}
if observed_counts != collection["source_counts"]:
raise ValueError("observed source population differs from the receipt")
checkpoint_fields = {"first", "last", "complete"}
for source in expected_sources:
checkpoint = collection["source_checkpoints"][source]
if (
not isinstance(checkpoint, dict)
or set(checkpoint) != checkpoint_fields
or not isinstance(checkpoint["complete"], bool)
):
raise ValueError("source checkpoint schema differs for " + source)
sequences = sorted(
event["source_sequence"] for event in events
if event["source_id"] == source
)
if sequences:
first = integer(checkpoint["first"], source + ".first")
last = integer(checkpoint["last"], source + ".last")
if first != sequences[0] or last != sequences[-1]:
raise ValueError("source checkpoint bounds differ for " + source)
if checkpoint["complete"] and sequences != list(range(first, last + 1)):
raise ValueError("complete source sequence has a gap for " + source)
elif checkpoint["first"] is not None or checkpoint["last"] is not None:
raise ValueError("empty source has checkpoint bounds for " + source)
start = parse_time(collection["window_start"])
end = parse_time(collection["window_end"])
horizon = parse_time(collection["observation_horizon_end"])
emitted = parse_time(collection["emitted_at"])
future = timedelta(seconds=policy["maximum_future_skew_seconds"])
if end <= start or horizon < end or emitted < horizon:
raise ValueError("measurement window or observation horizon is invalid")
if emitted > evaluated_at + future:
raise ValueError("collection receipt is future-dated beyond policy skew")
for event in events:
observed = parse_time(event["anomaly_observed_at"])
timestamps = [
parse_time(event[field]) for field in (
"alert_created_at", "human_acknowledged_at", "triage_completed_at"
) if event[field] is not None
]
if not start <= observed < end or any(value > horizon for value in timestamps):
raise ValueError("lifecycle event is outside its signed cohort or horizon")
age_seconds = max(0.0, (evaluated_at - horizon).total_seconds())
if age_seconds > policy["maximum_evidence_age_seconds"]:
readiness = "INSUFFICIENT_DATA"
readiness_reason = "signed observation horizon is stale"
elif (
collection["collection_complete"] is not True
or any(
checkpoint["complete"] is not True
for checkpoint in collection["source_checkpoints"].values()
)
):
readiness = "INSUFFICIENT_DATA"
readiness_reason = "authenticated source or checkpoint collection is incomplete"
else:
readiness = "READY"
readiness_reason = None
return {
"status": readiness,
"reason": readiness_reason,
"events": events,
"collection": collection,
"policy": policy,
"window_start": start,
"window_end": end,
"observation_horizon_end": horizon,
"evidence_age_seconds": age_seconds,
"input_evidence": {
name: captured.evidence() for name, captured in captures.items()
},
"input_key_sha256s": key_digests,
}
def classify(metrics: dict, policy: dict) -> tuple[str, str, list[str]]:
if metrics["eligible_alerts"] == 0:
return "INSUFFICIENT_DATA", "no eligible alerts in the signed cohort", []
breaches = []
for metric, threshold in (
("p95_detection_seconds", "max_p95_detection_seconds"),
("p95_acknowledgement_seconds", "max_p95_acknowledgement_seconds"),
("p95_triage_seconds", "max_p95_triage_seconds"),
):
value = metrics[metric]
if value is not None and value > policy[threshold]:
breaches.append(metric)
for metric in (
"acknowledgement_deadline_breaches", "triage_deadline_breaches",
"unacknowledged_mature_alerts", "untriaged_mature_alerts",
):
if metrics[metric] > 0:
breaches.append(metric)
if breaches:
return "FAIL", "one or more signed detection SLO targets were missed", breaches
if (
metrics["acknowledgement_right_censored_alerts"] > 0
or metrics["triage_right_censored_alerts"] > 0
):
return (
"INSUFFICIENT_DATA",
"one or more alert lifecycles have not reached a signed deadline",
[],
)
if (
metrics["detection_sample_count"] != metrics["eligible_alerts"]
or metrics["acknowledgement_sample_count"] != metrics["eligible_alerts"]
or metrics["triage_sample_count"] != metrics["eligible_alerts"]
):
raise ValueError("complete metric populations do not reconcile")
return "PASS", "all complete lifecycle populations met the signed targets", []
def evaluate(evaluated_at: datetime) -> dict:
inputs = verified_inputs(evaluated_at)
collection = inputs["collection"]
policy = inputs["policy"]
base = {
"schema_version": "aidefend.d017-slo-result.v3",
"control": "AID-D-017",
"guidance": "detection-slo-lifecycle-measurement",
"subject": collection["subject"],
"window_start": collection["window_start"],
"window_end": collection["window_end"],
"observation_horizon_end": collection["observation_horizon_end"],
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
"policy_version": policy["policy_version"],
"policy_sha256": inputs["input_evidence"]["policy"]["payload_sha256"],
"retention_policy_sha256": policy["retention_policy_sha256"],
"evidence_age_seconds": inputs["evidence_age_seconds"],
"expected_rule_ids": collection["expected_rule_ids"],
"expected_source_ids": collection["expected_source_ids"],
"event_population": len(inputs["events"]),
"input_evidence": inputs["input_evidence"],
}
if inputs["status"] != "READY":
return {
**base, "status": inputs["status"], "reason": inputs["reason"],
"breached_metrics": [], "metrics": None,
}
if not collection["expected_rule_ids"] and not inputs["events"]:
return {
**base, "status": "NOT_APPLICABLE",
"reason": "signed inventory contains no AI-security alert rule",
"breached_metrics": [], "metrics": None,
}
metrics = compute_detection_slo(
inputs["events"],
observation_horizon_end=inputs["observation_horizon_end"],
acknowledgement_deadline_seconds=policy["acknowledgement_deadline_seconds"],
triage_deadline_seconds=policy["triage_deadline_seconds"],
)
status, reason, breached = classify(metrics, policy)
return {
**base, "status": status, "reason": reason,
"breached_metrics": breached, "metrics": metrics,
}
def fsync_regular_file(path: Path) -> None:
descriptor = os.open(
path,
os.O_RDONLY
| getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_CLOEXEC", 0),
)
try:
metadata = os.fstat(descriptor)
if not stat.S_ISREG(metadata.st_mode):
raise ValueError(
"publication member is not a regular file: " + str(path)
)
os.fsync(descriptor)
finally:
os.close(descriptor)
def fsync_directory(path: Path) -> None:
descriptor = os.open(
path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
)
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def rename_noreplace(source: Path, target: Path) -> None:
libc = ctypes.CDLL(None, use_errno=True)
function = getattr(libc, "renameat2", None)
if function is None:
raise OSError(errno.ENOSYS, "renameat2 is required for no-replace publication")
function.argtypes = [
ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p,
ctypes.c_uint,
]
function.restype = ctypes.c_int
result = function(
AT_FDCWD, os.fsencode(source), AT_FDCWD, os.fsencode(target),
RENAME_NOREPLACE,
)
if result != 0:
number = ctypes.get_errno()
raise OSError(number, os.strerror(number), str(target))
def ensure_result_base() -> None:
metadata = RESULT_BASE.lstat()
if (
not stat.S_ISDIR(metadata.st_mode)
or RESULT_BASE.is_symlink()
or metadata.st_uid != os.geteuid()
or metadata.st_mode & 0o022
):
raise PermissionError(
"result base must be a pre-provisioned private evaluator directory"
)
def publish_result(value: dict, input_key_sha256s: list[str]) -> tuple[Path, dict]:
ensure_result_base()
result_key = read_stable(
RESULT_KEY, maximum_bytes=HARD_KEY_CEILING, trust_root=True
)
result_key_sha = hashlib.sha256(result_key).hexdigest()
if result_key_sha in set(input_key_sha256s):
raise ValueError("result and input verification roles share a key")
published_value = {
**value, "result_verification_key_sha256": result_key_sha,
}
raw = canonical(published_value) + b"\n"
result_sha = hashlib.sha256(raw).hexdigest()
target = RESULT_BASE / result_sha
if target.exists() or target.is_symlink():
raise FileExistsError("content-addressed result already exists: " + result_sha)
staging = Path(tempfile.mkdtemp(prefix="." + result_sha + ".", dir=RESULT_BASE))
os.chmod(staging, 0o700)
try:
result_path = staging / "result.json"
bundle_path = staging / "result.sigstore.json"
key_path = staging / "result.pub"
write_exclusive(result_path, raw)
write_exclusive(key_path, result_key)
subprocess.run(
[
COSIGN, "sign-blob", "--yes",
"--key", "env://D017_RESULT_SIGNING_KEY",
"--bundle", str(bundle_path), str(result_path),
],
check=True, timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
fsync_regular_file(bundle_path)
bundle_raw = read_stable(
bundle_path, maximum_bytes=HARD_METADATA_CEILING
)
if not hmac.compare_digest(
read_stable(result_path, maximum_bytes=HARD_METADATA_CEILING), raw
):
raise RuntimeError("private result changed before verification")
verify_bytes("D017 result", raw, bundle_raw, result_key)
completion = canonical({
"schema_version": "aidefend.signed-result-publication.v1",
"payload_sha256": result_sha,
"bundle_sha256": hashlib.sha256(bundle_raw).hexdigest(),
"verification_key_sha256": result_key_sha,
}) + b"\n"
committed_path = staging / "COMMITTED.json"
write_exclusive(committed_path, completion)
expected = {
"result.json": raw,
"result.sigstore.json": bundle_raw,
"result.pub": result_key,
"COMMITTED.json": completion,
}
if {path.name for path in staging.iterdir()} != set(expected):
raise RuntimeError("staged result directory contents differ")
for name, wanted in expected.items():
staged_path = staging / name
os.chmod(staged_path, 0o440)
fsync_regular_file(staged_path)
actual = read_stable(
staged_path, maximum_bytes=HARD_METADATA_CEILING
)
if not hmac.compare_digest(actual, wanted):
raise RuntimeError("staged result readback differs: " + name)
os.chmod(staging, 0o550)
fsync_directory(staging)
rename_noreplace(staging, target)
fsync_directory(RESULT_BASE)
staging = None
if {path.name for path in target.iterdir()} != set(expected):
raise RuntimeError("published result directory contents differ")
readback = {
name: read_stable(path, maximum_bytes=HARD_METADATA_CEILING)
for name, path in ((name, target / name) for name in expected)
}
for name, wanted in expected.items():
if not hmac.compare_digest(readback[name], wanted):
raise RuntimeError("published result readback differs: " + name)
verify_bytes(
"published D017 result", readback["result.json"],
readback["result.sigstore.json"], readback["result.pub"],
)
return target, published_value
finally:
if staging is not None and staging.exists():
os.chmod(staging, 0o700)
for path in staging.iterdir():
os.chmod(path, 0o600)
shutil.rmtree(staging)
def main() -> int:
evaluated_at = datetime.now(timezone.utc)
input_key_sha256s = []
try:
result = evaluate(evaluated_at)
input_key_sha256s = [
evidence["verification_key_sha256"]
for evidence in result.get("input_evidence", {}).values()
]
except Exception as error:
result = {
"schema_version": "aidefend.d017-slo-result.v3",
"control": "AID-D-017",
"guidance": "detection-slo-lifecycle-measurement",
"status": "ERROR",
"reason": "input verification, evaluation, or publication preparation failed",
"error_class": type(error).__name__,
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
}
target, published = publish_result(result, input_key_sha256s)
print(str(target))
return EXIT_CODES[published["status"]]
if __name__ == "__main__":
raise SystemExit(main())
Independently replay and interpret results
Replicate each committed content-addressed result directory and its exact signed inputs to append-only or WORM storage controlled by an identity separate from alert sources and the evaluator. Accept a directory only when COMMITTED.json is present, its recorded digests match readback bytes, and the v3 result verifies under the independently pinned result key. Replay must rebuild the validator from the captured signed schema bytes, reconstruct the signed source sequences and observation horizon, and recompute the detection, acknowledgement, triage, censoring, and threshold fields. Exercise a valid canonical alert with a direct duplicate, then reject rule-version, rule-configuration-digest, rule-schema-digest, missing-target, self-reference, duplicate-to-duplicate, and cyclic-duplicate fixtures. Fault-inject Cosign-bundle or staged-file fsync failure and an unexpected staging member; none may reach the atomic rename.
Action: At the detection-assurance job, reconcile the complete signed alert lifecycle population, compute each policy-defined SLO measure, persist the independently replayable signed result, and publish coverage or SLO findings to detection engineering and the control owner; do not change alert routing or promotion state here.
AID-D-017-G002
Compute policy-defined dwell-time, triage-time, and investigation-coverage metric series for complete, explicitly bounded AI-security alert cohorts.
Define one denominator before computing metrics
For each policy window, materialize an immutable eligible-alert cohort keyed by asset, rule revision, severity class, source population, and alert creation interval. Join anomaly-observed, alert-created, human-acknowledged, triage-completed, and investigation-closed events by stable alert ID and reject duplicate, reordered, impossible, or cross-tenant lifecycle links. Compute dwell and triage distributions from policy-defined timestamps and compute coverage as investigated eligible alerts divided by the complete eligible cohort; never drop unresolved alerts from the denominator.
Publish bounded series and findings
Emit count, numerator, denominator, percentile or threshold result, missing-stage count, late-arrival count, and policy window for every cohort. Backfill only under an auditable metric revision. Exercise zero-alert windows, still-open alerts, clock skew, duplicate acknowledgements, late events, rule revisions, suppression, and queue migration. A missed target is a detection-assurance finding; missing population or lifecycle evidence is INSUFFICIENT_DATA, not a favorable metric.
Executable detection-SLO computation
# File: detection_slo/compute.py
from __future__ import annotations
from datetime import datetime
def seconds(value: str) -> float:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError("timestamp lacks timezone")
return parsed.timestamp()
def compute(rows: list[dict]) -> dict:
if not rows:
raise ValueError("empty signed alert cohort")
dwell, triage, investigated = [], [], 0
for row in rows:
introduced = seconds(row["introduced_at"])
detected = seconds(row["detected_at"])
triaged = seconds(row["triaged_at"])
if not introduced <= detected <= triaged:
raise ValueError("invalid alert chronology")
dwell.append(detected - introduced)
triage.append(triaged - detected)
investigated += row["investigation_complete"] is True
return {
"population": len(rows),
"maximum_dwell_seconds": max(dwell),
"maximum_triage_seconds": max(triage),
"investigation_coverage": investigated / len(rows),
}Use incident-introduction time only when it comes from an independently reconstructed source; otherwise report the applicable observable latency metric and mark dwell time unsupported. Compare results with signed service objectives after computing the complete cohort.
AID-D-017-G003
Package each detection-SLO result with the signed cohort, metric, policy, raw-event digests, and replay instructions needed to recompute it without querying the producing pipeline.
Require signed cohort, policy, and computed result
The metric evaluator emits a signed result whose schema binds the exact cohort digest, policy digest/version, rule digest, metric-definition digest, raw-event manifest, lifecycle-population root, evaluator build, measured populations, coverage, and finding state. Packaging does not recompute the policy decision, but it must verify all three inputs and reject any aggregate that is unsigned, out of range, or bound to different evidence.
Create and verify a no-replace assurance package
set -euo pipefail
set -o noclobber
umask 077
test -n "$SLO_SIGNING_KEY_URI"
: "${SLO_CRYPTO_TIMEOUT_SECONDS:?signed crypto timeout is required}"
python - "$SLO_CRYPTO_TIMEOUT_SECONDS" <<'PY'
import math
import sys
try:
value = float(sys.argv[1])
except ValueError as error:
raise SystemExit("crypto timeout is not numeric") from error
if not math.isfinite(value) or value <= 0:
raise SystemExit("crypto timeout must be finite and positive")
PY
EVIDENCE_ROOT="$(pwd -P)/evidence"
test -d "$EVIDENCE_ROOT"
test ! -L "$EVIDENCE_ROOT"
SNAPSHOT="$(mktemp -d)"
trap 'rm -rf -- "$SNAPSHOT"' EXIT
chmod 0700 "$SNAPSHOT"
for name in cohort.json cohort.sigstore.json policy.json policy.sigstore.json result.json result.sigstore.json; do
source="evidence/input/$name"
test -f "$source"
test ! -L "$source"
install -m 0400 -- "$source" "$SNAPSHOT/$name"
done
timeout --signal=TERM "$SLO_CRYPTO_TIMEOUT_SECONDS" cosign verify-blob "$SNAPSHOT/cohort.json" \
--key trust/slo-cohort.pub \
--bundle "$SNAPSHOT/cohort.sigstore.json"
timeout --signal=TERM "$SLO_CRYPTO_TIMEOUT_SECONDS" cosign verify-blob "$SNAPSHOT/policy.json" \
--key trust/slo-policy.pub \
--bundle "$SNAPSHOT/policy.sigstore.json"
timeout --signal=TERM "$SLO_CRYPTO_TIMEOUT_SECONDS" cosign verify-blob "$SNAPSHOT/result.json" \
--key trust/slo-evaluator.pub \
--bundle "$SNAPSHOT/result.sigstore.json"
RESULT_SHA="$(python - "$SNAPSHOT/cohort.json" "$SNAPSHOT/policy.json" "$SNAPSHOT/result.json" <<'PY'
import hashlib
import json
import math
import sys
from pathlib import Path
def strict_object(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key: " + key)
value[key] = item
return value
def reject_constant(value):
raise ValueError("non-finite JSON constant: " + value)
def load(path):
raw = Path(path).read_bytes()
value = json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=strict_object,
parse_constant=reject_constant,
)
if not isinstance(value, dict):
raise ValueError("signed artifact must be a JSON object")
return raw, value
cohort_raw, cohort = load(sys.argv[1])
policy_raw, policy = load(sys.argv[2])
result_raw, result = load(sys.argv[3])
required = {
"schema_version", "cohort_sha256", "policy_sha256", "policy_version",
"rule_digest", "metric_definition_sha256", "raw_event_manifest_sha256",
"lifecycle_population_root", "evaluator_build_digest",
"eligible_population", "investigated_population", "missing_population",
"error_population", "investigation_coverage", "finding_status",
}
if set(result) != required:
raise SystemExit("SLO result schema mismatch")
if result["schema_version"] != "aidefend.detection_slo_result.v2":
raise SystemExit("SLO result schema version mismatch")
if result["cohort_sha256"] != hashlib.sha256(cohort_raw).hexdigest():
raise SystemExit("SLO result is bound to another cohort")
if result["policy_sha256"] != hashlib.sha256(policy_raw).hexdigest():
raise SystemExit("SLO result is bound to another policy")
if result["policy_version"] != policy.get("policy_version"):
raise SystemExit("SLO policy version binding mismatch")
if result["rule_digest"] != policy.get("rule_digest"):
raise SystemExit("SLO rule binding mismatch")
if result["metric_definition_sha256"] != policy.get("metric_definition_sha256"):
raise SystemExit("metric-definition binding mismatch")
if result["lifecycle_population_root"] != cohort.get("lifecycle_population_root"):
raise SystemExit("lifecycle-population binding mismatch")
if result["raw_event_manifest_sha256"] != cohort.get("raw_event_manifest_sha256"):
raise SystemExit("raw-event manifest binding mismatch")
if not isinstance(result["evaluator_build_digest"], str) or not result["evaluator_build_digest"]:
raise SystemExit("evaluator build digest is absent")
names = (
"eligible_population", "investigated_population",
"missing_population", "error_population",
)
counts = {}
for name in names:
value = result[name]
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise SystemExit(name + " is invalid")
counts[name] = value
if counts["eligible_population"] < 1:
raise SystemExit("eligible population is empty")
if (
counts["investigated_population"]
+ counts["missing_population"]
+ counts["error_population"]
!= counts["eligible_population"]
):
raise SystemExit("SLO population does not reconcile")
coverage = result["investigation_coverage"]
if (
isinstance(coverage, bool)
or not isinstance(coverage, (int, float))
or not math.isfinite(float(coverage))
or not 0.0 <= float(coverage) <= 1.0
):
raise SystemExit("investigation coverage is outside [0,1]")
expected = counts["investigated_population"] / counts["eligible_population"]
if not math.isclose(float(coverage), expected, rel_tol=0.0, abs_tol=1e-12):
raise SystemExit("investigation coverage differs from raw counts")
if result["finding_status"] not in {"FINDING", "NO_FINDING", "UNKNOWN"}:
raise SystemExit("finding status is invalid")
print(hashlib.sha256(result_raw).hexdigest())
PY
)"
PACKAGE="$EVIDENCE_ROOT/detection-slo-evidence-$RESULT_SHA.tar"
BUNDLE="$PACKAGE.sigstore.json"
test ! -e "$PACKAGE"
test ! -e "$BUNDLE"
tar --sort=name --mtime=@0 --owner=0 --group=0 --numeric-owner \
-C "$SNAPSHOT" -cf - \
cohort.json cohort.sigstore.json policy.json policy.sigstore.json \
result.json result.sigstore.json > "$PACKAGE"
timeout --signal=TERM "$SLO_CRYPTO_TIMEOUT_SECONDS" cosign sign-blob --yes --key "$SLO_SIGNING_KEY_URI" \
--bundle "$BUNDLE" "$PACKAGE"
timeout --signal=TERM "$SLO_CRYPTO_TIMEOUT_SECONDS" cosign verify-blob "$PACKAGE" \
--key trust/slo-packager.pub --bundle "$BUNDLE"Independent replay and storage
Run the script in a separately credentialed, resource-limited CI job with read-only immutable input mounts; resolve SLO_CRYPTO_TIMEOUT_SECONDS from the signed environment profile, and resolve the signing-key URI to KMS, HSM, or another approved signer. A read-only verifier fetches the stored package and bundle, re-verifies every signature, reconstructs the eligible cohort from the referenced raw lifecycle events, recomputes the result, and compares its canonical digest. Store the package through append-only, create-only storage for the duration resolved by signed organizational or legal policy. A valid metric breach is a finding; population or binding mismatch is FAIL; missing evidence is INSUFFICIENT_DATA; evaluator, verification, packaging, or storage failure is ERROR.
Open-source tools
- OpenTelemetry
- Prometheus
- Grafana
- OpenSearch
- ClickHouse
- Apache Kafka
- jsonschema
- Cosign
Commercial / hosted tools
- Splunk Enterprise Security
- Microsoft Sentinel
- Datadog
- Grafana Cloud
- ServiceNow Security Operations
Threat mappings
MITRE ATLAS
- N/A
MAESTRO
- Compromised Observability Tools (L5)
- Poisoning Observability Data (L5)
OWASP LLM Top 10 2026
- N/A
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- N/A
Google Secure AI Framework 2.0 - Risks
- N/A
Databricks AI Security Framework 3.0
- Platform 12.3: Lack of incident response