AID-D-005.011: Incident-Time AI Volatile-State Capture & Evidence Preservation
Capture a bounded, pre-containment snapshot of security-relevant volatile AI-system state—such as active agent/session identifiers, pending tool-call state, inference-gateway correlation IDs, retrieval/index revisions, runtime/network facts, queue offsets, and in-flight trace references—then preserve the sanitized artifacts and their digest manifest without delaying containment beyond a signed deadline.
Pillars: app, model, infra, data
Lifecycle: operation, response
Scope boundary
Owns incident-triggered, deadline-bounded capture and preservation of otherwise ephemeral AI runtime state before containment, using an allowlisted plan, sanitized artifacts, signed digest manifests, immutable storage, and readback verification.
AID-D-005.001: AID-D-005.011 captures a bounded volatile-state snapshot when an incident is declared; AID-D-005.001 continuously generates and collects routine structured logs. Volatile capture preserves in-flight facts that ordinary event logging may never materialize.
AID-D-005.010: AID-D-005.011 preserves incident-time state and its evidence manifest; AID-D-005.010 determines whether the ongoing telemetry path and sensors are complete and healthy. A snapshot can document current state despite sensor loss, but it does not repair or certify the telemetry pipeline.
Implementation guidance
- AID-D-005.011-G001: Use a signed, allowlisted capture plan to collect a deadline-bounded set of volatile AI-runtime facts before containment; preserve only sanitized artifacts and signed manifests in immutable storage, then read every stored object back and recompute its digest.
AID-D-005.011-G001
Use a signed, allowlisted capture plan to collect a deadline-bounded set of volatile AI-runtime facts before containment; preserve only sanitized artifacts and signed manifests in immutable storage, then read every stored object back and recompute its digest.
Capture contract
An authenticated incident coordinator verifies separately signed incident, policy, and AI asset-inventory artifacts. The policy owns the exact adapter population and HTTPS origins, allowed artifact types, per-adapter timeout, total capture budget, containment safety margin, maximum bytes, retention, and forbidden fields. Adapters are pre-deployed components (for example, an OpenTelemetry tail-buffer exporter, Velociraptor client, Kubernetes state collector, or agent-runtime snapshot endpoint); this coordinator never executes caller-supplied commands. Raw prompts, hidden chain-of-thought, credentials, tokens, and raw tool secrets are default-denied. Any exception requires a separately signed incident-specific H-037.001 break-glass authorization naming the exact field and retention.
Deadline-bounded coordinator
from __future__ import annotations
import math as _aidefend_math
import os as _aidefend_os
# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
value = _aidefend_os.environ.get(name)
if value is None or not value.strip():
raise RuntimeError(f"required runtime-profile field is absent: {name}")
return value.strip()
def _aidefend_positive_float(name):
try:
value = float(_aidefend_required(name))
except ValueError as error:
raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
if not _aidefend_math.isfinite(value) or value <= 0:
raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
return value
def _aidefend_positive_int(name):
try:
value = int(_aidefend_required(name))
except ValueError as error:
raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
if value < 1:
raise RuntimeError(f"runtime-profile field must be positive: {name}")
return value
def _aidefend_nonnegative_int(name):
try:
value = int(_aidefend_required(name))
except ValueError as error:
raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
if value < 0:
raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
return value
def _aidefend_fraction(name):
value = _aidefend_positive_float(name)
if value > 1:
raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
return value
RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
CAPTURE_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D005011_CAPTURE_COMMAND_TIMEOUT_SECONDS")
# file: incident/capture_ai_volatile_state.py
import hashlib, json, os, subprocess, time
from datetime import datetime, timezone
from pathlib import Path
import hmac
import re
import tempfile
from urllib.parse import urlparse
import requests
INCIDENT = Path("incident/trigger.json")
POLICY = Path("policy/volatile-capture-policy.json")
INVENTORY = Path("inventory/volatile-state-adapters.json")
MANIFEST = Path("incident/AID-D-005.011-manifest.json")
FORBIDDEN = {"raw_prompt", "hidden_chain_of_thought", "credential",
"access_token", "refresh_token", "raw_tool_secret"}
def digest(path: Path) -> str:
captured = _VERIFIED_INPUTS.get(Path(path))
if captured is not None:
return hashlib.sha256(captured).hexdigest()
value = hashlib.sha256()
with Path(path).open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
value.update(chunk)
return value.hexdigest()
_VERIFIED_INPUTS: dict[Path, bytes] = {}
def _write_private_snapshot(path: Path, payload: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.chmod(path, 0o400)
def _strict_json(raw: bytes) -> object:
def unique_object(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key: " + key)
value[key] = item
return value
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + value)
),
)
def verify(path: Path, key_env: str) -> None:
payload_path = Path(path)
signature_path = Path(str(payload_path) + ".sig")
payload_raw = payload_path.read_bytes()
signature_raw = signature_path.read_bytes()
key_reference = os.environ[key_env]
remote_key = bool(re.match(
r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
)) or key_reference.startswith("pkcs11:")
key_raw = None if remote_key else (
key_reference.encode("utf-8")
if key_reference.startswith("-----BEGIN ")
else Path(key_reference).read_bytes()
)
with tempfile.TemporaryDirectory(prefix="aidefend-d005011-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
signature_snapshot = root / "signature"
_write_private_snapshot(payload_snapshot, payload_raw)
_write_private_snapshot(signature_snapshot, signature_raw)
captured = [(payload_snapshot, payload_raw), (signature_snapshot, signature_raw)]
key_argument = key_reference
if key_raw is not None:
key_snapshot = root / "verify-key.pub"
_write_private_snapshot(key_snapshot, key_raw)
captured.append((key_snapshot, key_raw))
key_argument = str(key_snapshot)
try:
subprocess.run(
[
"cosign", "verify-blob", "--key", key_argument,
"--bundle", str(signature_snapshot), str(payload_snapshot),
],
check=True, capture_output=True, text=True,
timeout=CAPTURE_COMMAND_TIMEOUT_SECONDS,
)
if any(
not hmac.compare_digest(candidate.read_bytes(), expected)
for candidate, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for candidate, _ in captured:
os.chmod(candidate, 0o600)
_VERIFIED_INPUTS[payload_path] = payload_raw
def now() -> str:
return datetime.now(timezone.utc).isoformat()
def load() -> tuple[dict, dict, dict]:
verify(INCIDENT, "D005011_INCIDENT_VERIFY_KEY")
verify(POLICY, "D005011_POLICY_VERIFY_KEY")
verify(INVENTORY, "D005011_INVENTORY_VERIFY_KEY")
incident = _strict_json(_VERIFIED_INPUTS[INCIDENT])
policy = _strict_json(_VERIFIED_INPUTS[POLICY])
inventory = _strict_json(_VERIFIED_INPUTS[INVENTORY])
incident_fields = {"schema_version", "incident_id", "triggered_at",
"subject_assets", "containment_deadline"}
if set(incident) != incident_fields or incident["schema_version"] != "aidefend.incident-trigger.v1":
raise ValueError("incident schema differs")
policy_fields = {"schema_version", "policy_version", "required_adapter_ids",
"allowed_artifact_types", "forbidden_fields", "adapter_timeout_seconds",
"capture_budget_seconds", "containment_safety_margin_seconds",
"maximum_total_bytes", "maximum_response_bytes", "response_chunk_bytes",
"retention_seconds", "worm_bucket"}
if set(policy) != policy_fields or policy["schema_version"] != "aidefend.volatile-capture-policy.v1":
raise ValueError("policy schema differs")
for field in ("adapter_timeout_seconds", "capture_budget_seconds",
"containment_safety_margin_seconds", "maximum_total_bytes",
"maximum_response_bytes", "response_chunk_bytes", "retention_seconds"):
if isinstance(policy[field], bool) or not isinstance(policy[field], int) or policy[field] < 1:
raise ValueError(f"{field} is invalid")
if (policy["maximum_response_bytes"] < 2
or policy["response_chunk_bytes"] > policy["maximum_response_bytes"]):
raise ValueError("adapter response bounds are invalid")
if not FORBIDDEN.issubset(set(policy["forbidden_fields"])):
raise ValueError("policy removed a default-denied field")
inventory_fields = {"schema_version", "inventory_version", "subject_assets", "adapters"}
if set(inventory) != inventory_fields or inventory["schema_version"] != "aidefend.volatile-adapter-inventory.v1":
raise ValueError("inventory schema differs")
if sorted(inventory["subject_assets"]) != sorted(incident["subject_assets"]):
raise ValueError("incident subject differs from inventory")
adapter_fields = {"adapter_id", "component_id", "component_version", "endpoint",
"mtls_cert_sha256", "supported_artifact_types"}
if any(not isinstance(item, dict) or set(item) != adapter_fields
for item in inventory["adapters"]):
raise ValueError("adapter entry differs")
ids = [item["adapter_id"] for item in inventory["adapters"]]
if len(ids) != len(set(ids)) or set(ids) != set(policy["required_adapter_ids"]):
raise ValueError("required adapter population differs")
for item in inventory["adapters"]:
uri = urlparse(item["endpoint"])
if uri.scheme != "https" or uri.username or uri.password or uri.query or uri.fragment:
raise ValueError("adapter endpoint is not an allowlisted HTTPS origin")
if not set(item["supported_artifact_types"]).issubset(set(policy["allowed_artifact_types"])):
raise ValueError("adapter artifact type is unauthorized")
return incident, policy, inventory
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate adapter JSON key: {key}")
value[key] = item
return value
def read_adapter_json(response, policy):
try:
chunks, observed = [], 0
for chunk in response.iter_content(chunk_size=policy["response_chunk_bytes"]):
observed += len(chunk)
if observed > policy["maximum_response_bytes"]:
raise ValueError("adapter response exceeds signed bound")
chunks.append(chunk)
return json.loads(b"".join(chunks).decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)))
finally:
response.close()
def capture(adapter: dict, incident: dict, policy: dict,
deadline: float, remaining_bytes: int) -> dict:
remaining = deadline - time.monotonic()
if remaining <= 0 or remaining_bytes <= 0:
return {"adapter_id": adapter["adapter_id"], "component_id": adapter["component_id"],
"status": "INSUFFICIENT_DATA",
"reason": "containment reserve or byte budget reached",
"capture_started_at": now(), "capture_ended_at": now()}
response = requests.post(
adapter["endpoint"].rstrip("/") + "/v1/security/volatile-capture",
json={"schema_version": "aidefend.capture-request.v1",
"incident_id": incident["incident_id"],
"subject_assets": incident["subject_assets"],
"allowed_artifact_types": adapter["supported_artifact_types"],
"forbidden_fields": policy["forbidden_fields"],
"maximum_bytes": remaining_bytes,
"deadline": incident["containment_deadline"]},
cert=(os.environ["D005011_CLIENT_CERT"], os.environ["D005011_CLIENT_KEY"]),
verify=os.environ["D005011_CA_BUNDLE"],
timeout=min(float(policy["adapter_timeout_seconds"]), remaining),
headers={"Idempotency-Key": incident["incident_id"] + ":" + adapter["adapter_id"], "Accept-Encoding": "identity"},
allow_redirects=False,
stream=True,
)
response.raise_for_status()
result = read_adapter_json(response, policy)
fields = {"schema_version", "adapter_id", "component_id", "component_version",
"status", "capture_started_at", "capture_ended_at", "clock_offset_ms",
"consistency_epoch", "source_revision", "queue_offsets", "artifacts"}
if set(result) != fields or result["schema_version"] != "aidefend.capture-result.v1":
raise ValueError("adapter result schema differs")
if result["adapter_id"] != adapter["adapter_id"] or result["component_id"] != adapter["component_id"]:
raise ValueError("adapter identity differs")
if result["component_version"] != adapter["component_version"]:
raise ValueError("adapter component version differs")
if result["status"] not in {"CAPTURED", "INSUFFICIENT_DATA", "ERROR"}:
raise ValueError("adapter status differs")
artifact_fields = {"artifact_type", "immutable_uri", "sha256", "size_bytes",
"retention_until", "sanitization_policy_version", "readback_sha256"}
if any(not isinstance(item, dict) or set(item) != artifact_fields
or item["artifact_type"] not in adapter["supported_artifact_types"]
or item["sha256"] != item["readback_sha256"]
or isinstance(item["size_bytes"], bool) or not isinstance(item["size_bytes"], int)
or item["size_bytes"] < 1 for item in result["artifacts"]):
raise ValueError("artifact manifest/readback differs")
if result["status"] == "CAPTURED":
if not result["artifacts"]:
raise ValueError("adapter claimed CAPTURED with an empty artifact population")
for field in ("capture_started_at", "capture_ended_at",
"consistency_epoch", "source_revision"):
if not isinstance(result[field], str) or not result[field]:
raise ValueError(f"captured component lacks {field}")
if (isinstance(result["clock_offset_ms"], bool)
or not isinstance(result["clock_offset_ms"], (int, float))
or not isinstance(result["queue_offsets"], dict)):
raise ValueError("captured consistency facts are invalid")
if sum(item["size_bytes"] for item in result["artifacts"]) > remaining_bytes:
raise ValueError("adapter exceeded remaining signed byte budget")
return result
def main() -> None:
try:
incident, policy, inventory = load()
if not inventory["adapters"]:
status, results = "NOT_APPLICABLE", []
else:
containment = datetime.fromisoformat(
incident["containment_deadline"].replace("Z", "+00:00")
).astimezone(timezone.utc)
usable = min(float(policy["capture_budget_seconds"]),
(containment - datetime.now(timezone.utc)).total_seconds()
- float(policy["containment_safety_margin_seconds"]))
deadline, results = time.monotonic() + max(0.0, usable), []
used_bytes = 0
for adapter in sorted(inventory["adapters"], key=lambda item: item["adapter_id"]):
try:
result = capture(
adapter, incident, policy, deadline,
policy["maximum_total_bytes"] - used_bytes,
)
results.append(result)
used_bytes += sum(
artifact["size_bytes"] for artifact in result.get("artifacts", [])
)
except Exception as error:
results.append({"adapter_id": adapter["adapter_id"],
"component_id": adapter["component_id"], "status": "ERROR",
"reason": f"{type(error).__name__}: {error}",
"capture_started_at": now(), "capture_ended_at": now()})
if time.monotonic() >= deadline:
break
seen = {item["adapter_id"] for item in results}
for adapter in inventory["adapters"]:
if adapter["adapter_id"] not in seen:
results.append({"adapter_id": adapter["adapter_id"],
"component_id": adapter["component_id"],
"status": "INSUFFICIENT_DATA",
"reason": "not started before containment reserve",
"capture_started_at": now(), "capture_ended_at": now()})
states = {item["status"] for item in results}
status = "ERROR" if "ERROR" in states else (
"INSUFFICIENT_DATA" if states != {"CAPTURED"} else "PASS"
)
manifest = {"schema_version": "aidefend.d005011-manifest.v1",
"control": "AID-D-005.011", "incident_id": incident["incident_id"],
"status": status,
"consistency_claim": "COMPONENT_FACTS_ONLY_NOT_DISTRIBUTED_ATOMIC",
"incident_sha256": digest(INCIDENT), "policy_sha256": digest(POLICY),
"policy_version": policy["policy_version"],
"inventory_sha256": digest(INVENTORY),
"inventory_version": inventory["inventory_version"],
"component_results": results,
"containment_deadline": incident["containment_deadline"],
"worm_bucket": policy["worm_bucket"],
"retention_seconds": policy["retention_seconds"],
"capture_completed_at": now()}
except Exception as error:
manifest = {"schema_version": "aidefend.d005011-manifest.v1",
"control": "AID-D-005.011", "status": "ERROR",
"reason": f"{type(error).__name__}: {error}",
"capture_completed_at": now()}
MANIFEST.parent.mkdir(parents=True, exist_ok=True)
MANIFEST.write_text(json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n",
encoding="utf-8")
subprocess.run(["cosign", "sign-blob", "--yes", "--key",
"env://D005011_MANIFEST_SIGNING_KEY", "--bundle",
str(MANIFEST) + ".sig", str(MANIFEST)], check=True, timeout=CAPTURE_COMMAND_TIMEOUT_SECONDS)
verify(MANIFEST, "D005011_MANIFEST_VERIFY_KEY")
if __name__ == "__main__":
main()
Preserve partial capture without delaying containment
The storage writer copies the signed manifest to the policy-named object-lock/WORM bucket, reads it back, and verifies its digest, signature, retention mode, and retention-until value. Every adapter result must already contain the artifact URI, byte count, digest, readback digest, retention, sanitizer-policy version, component version, capture start/end, clock offset, consistency epoch, source revision, and queue offsets. An independent incident-forensics identity repeats all readbacks and verifies the complete adapter population.
Accept the capture only when every required adapter is collected and independently read back within the deadline. Record partial, deadline-limited, invalid, or failed capture in the manifest, and require signed inventory to justify any excluded unsupported state. At the reserved deadline, stop capture attempts and emit the partial or failed-capture manifest so Isolate/Evict proceeds. Never delay containment, treat empty capture as clean, or claim distributed atomic consistency from component-local facts.
Verify complete coverage
The population is every target, capture step, artifact, skip, timeout, error, deadline, sanitization result, manifest entry, evidence write and independent readback declared by the signed incident capture plan.
Open-source tools
- OpenTelemetry Collector
- Velociraptor
- GRR Rapid Response (endpoint volatile-state and artifact acquisition; AI-specific collectors and immutable preservation remain integration-owned)
- Kubernetes
- Ceph Object Gateway
- Cosign
Commercial / hosted tools
- Amazon S3 Object Lock
- Azure Immutable Blob Storage
- Google Cloud Bucket Lock
- Splunk Enterprise Security
Threat mappings
MITRE ATLAS
- N/A
MAESTRO
- Repudiation (L7)
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
- Agents - Core 13.8: Repudiation & Untraceability