AID-D-015: High-Risk Approval Bypass & HITL Activity Detection
Detect missing, invalid, expired, mismatched, replayed, or anomalous human-approval records around high-risk agent actions, and monitor HITL request, decision, escalation, release, and execution activity for unattended or abusive patterns. Correlate the signed approval, authenticated operator, exact action digest, policy version, and canonical event chain against the execution record; this control never authorizes or executes the action.
Pillars: app
Lifecycle: operation
Scope boundary
Owns telemetry correlation, signed approval-record validation, HITL activity analytics, and findings for missing, invalid, expired, mismatched, replayed, unattended, or abusive high-risk approval lifecycles. It never authorizes, unlocks, or executes the action.
AID-H-018.003: AID-D-015 independently detects bypass, replay, mismatch, expiry, and anomalous approval activity; AID-H-018.003 validates the exact high-impact action and issues the canonical runtime approval or denial receipt. The detector must remain independent of the gate whose failures it is intended to find.
AID-M-006.003: AID-D-015 detects violations and anomalies in the observed HITL event chain; AID-M-006.003 defines the canonical escalation topology, event fields, producer obligations, and reconciliation contract. The Model control specifies expected evidence, while Detect owns production findings when that evidence is missing or inconsistent.
Implementation guidance
- AID-D-015-G001: Correlate high-risk execution with a signed, short-lived approval bound to trusted issuer, audience, authenticated approver, exact action digest, nonce, and one-time token ID; alert on missing, invalid, expired, mismatched, or replayed evidence.
- AID-D-015-G002: Evaluate the complete canonical HITL v3 lifecycle population against signature-verified gate-event records, the signed analytic policy, and authenticated collection completeness to detect approval bypass, invalid terminal semantics, operator bursts, and unattended escalation.
AID-D-015-G001
Correlate high-risk execution with a signed, short-lived approval bound to trusted issuer, audience, authenticated approver, exact action digest, nonce, and one-time token ID; alert on missing, invalid, expired, mismatched, or replayed evidence.
Concept and scope boundary
The authorization gate remains in AID-H-018.003. This detector independently re-verifies the signed approval attached to each observed execution. It compares the exact expected challenge and nonce, resolves the approver's authentication event from the identity provider over mTLS, and keeps an idempotency ledger. Issuer, audience, signing keys, evidence-reference schemes, authentication assurance, token lifetime, clock skew, authentication-event age, and request timeout come from one exact-schema policy verified under the detector image's pinned policy key. Event strings, caller-supplied limits, or approved=true fields are not evidence.
Verify approval, authentication, context, and reuse
from __future__ import annotations
# File: monitoring/high_risk_confirmation_monitor.py
import hashlib
import hmac
import json
import math
import os
import subprocess
import time
import tempfile
from pathlib import Path
from typing import Any
from urllib.parse import quote, urlparse
import jwt
import requests
from redis import Redis
from redis.exceptions import RedisError
POLICY_PATH = Path("/etc/aidefend/d015/approval-monitor-policy.json")
POLICY_SIGNATURE_PATH = Path("/etc/aidefend/d015/approval-monitor-policy.sig")
POLICY_PUBLIC_KEY_PATH = Path("/opt/aidefend/trust/d015-policy-authority.pub")
IDP_CLIENT_CERTIFICATE_PATH = "/var/run/secrets/d015/idp-client.crt"
IDP_CLIENT_KEY_PATH = "/var/run/secrets/d015/idp-client.key"
IDP_CA_BUNDLE_PATH = "/opt/aidefend/trust/idp-ca.pem"
SUPPORTED_EVIDENCE_SCHEMES = {"https", "s3", "gs"}
POLICY_VERIFY_TIMEOUT_SECONDS = float(os.environ["D015_POLICY_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(POLICY_VERIFY_TIMEOUT_SECONDS) or POLICY_VERIFY_TIMEOUT_SECONDS <= 0:
raise RuntimeError("signed runtime-profile verification timeout is invalid")
POLICY_FIELDS = {
"schema_version", "policy_version", "approval_issuer",
"approval_audience", "approval_keys", "idp_audit_issuer",
"idp_audit_base_url", "allowed_auth_assurance_levels",
"allowed_evidence_schemes", "max_approval_lifetime_seconds",
"max_clock_skew_seconds", "max_auth_event_age_seconds",
"max_evidence_refs", "idp_request_timeout_seconds",
"idp_response_max_bytes", "idp_response_chunk_bytes",
}
REQUIRED = [
"iss", "aud", "sub", "iat", "nbf", "exp", "jti", "nonce",
"challenge_id", "action_sha256", "executor_principal_id",
"approver_authentication_event_id", "evidence_refs",
]
class PolicyLoadError(RuntimeError):
pass
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise PolicyLoadError(f"duplicate policy key: {key}")
value[key] = item
return value
def load_signed_policy() -> tuple[dict[str, Any], str]:
raw = POLICY_PATH.read_bytes()
signature_raw = POLICY_SIGNATURE_PATH.read_bytes()
with tempfile.TemporaryDirectory(prefix="aidefend-d015-") as directory:
snapshot = Path(directory) / "policy.json"
signature = Path(directory) / "policy.sig"
snapshot.write_bytes(raw)
signature.write_bytes(signature_raw)
subprocess.run(
["cosign", "verify-blob", "--key", str(POLICY_PUBLIC_KEY_PATH),
"--bundle", str(signature), str(snapshot)],
check=True, capture_output=True, text=True,
timeout=POLICY_VERIFY_TIMEOUT_SECONDS,
)
policy = json.loads(raw.decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)))
if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
raise PolicyLoadError("approval-monitor policy schema mismatch")
if policy["schema_version"] != "aidefend.d015002-policy.v1":
raise PolicyLoadError("unsupported approval-monitor policy schema")
for field in (
"policy_version", "approval_issuer", "approval_audience",
"idp_audit_issuer", "idp_audit_base_url",
):
if not isinstance(policy[field], str) or not policy[field].strip():
raise PolicyLoadError(f"policy field {field} is empty")
base = urlparse(policy["idp_audit_base_url"])
if (
base.scheme != "https" or not base.netloc or base.username
or base.password or base.query or base.fragment
):
raise PolicyLoadError("IDP audit base URL must be a fixed HTTPS origin/path")
keys = policy["approval_keys"]
if not isinstance(keys, dict) or not keys:
raise PolicyLoadError("signed approval key registry is empty")
if any(
not isinstance(kid, str) or not kid or "/" in kid
or not isinstance(pem, str) or "BEGIN PUBLIC KEY" not in pem
for kid, pem in keys.items()
):
raise PolicyLoadError("signed approval key registry is malformed")
for field in (
"max_approval_lifetime_seconds", "max_clock_skew_seconds",
"max_auth_event_age_seconds", "max_evidence_refs",
"idp_request_timeout_seconds", "idp_response_max_bytes",
"idp_response_chunk_bytes",
):
value = policy[field]
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise PolicyLoadError(f"policy field {field} must be a positive integer")
if (policy["idp_response_max_bytes"] < 2
or policy["idp_response_chunk_bytes"] > policy["idp_response_max_bytes"]):
raise PolicyLoadError("IDP response bounds are invalid")
schemes = policy["allowed_evidence_schemes"]
if (
not isinstance(schemes, list) or not schemes
or any(not isinstance(item, str) or not item for item in schemes)
or len(schemes) != len(set(schemes))
or not set(schemes).issubset(SUPPORTED_EVIDENCE_SCHEMES)
):
raise PolicyLoadError("allowed evidence schemes are invalid")
assurance = policy["allowed_auth_assurance_levels"]
if (
not isinstance(assurance, list) or not assurance
or any(not isinstance(item, str) or not item for item in assurance)
or len(assurance) != len(set(assurance))
):
raise PolicyLoadError("authentication assurance policy is invalid")
return policy, hashlib.sha256(raw).hexdigest()
def canonical_json(value: Any) -> str:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
)
def require_nonempty_string(value: Any, field: str) -> str:
if not isinstance(value, str) or not value.strip() or len(value) > 2048:
raise ConfirmationViolation(f"invalid {field}")
return value
def validate_evidence_refs(value: Any, policy: dict[str, Any]) -> list[str]:
if (
not isinstance(value, list)
or not 1 <= len(value) <= policy["max_evidence_refs"]
):
raise ConfirmationViolation("evidence_refs must be a bounded non-empty list")
refs = []
for raw in value:
ref = require_nonempty_string(raw, "evidence_ref")
parsed = urlparse(ref)
if parsed.scheme not in set(policy["allowed_evidence_schemes"]):
raise ConfirmationViolation("unsupported evidence reference scheme")
if not parsed.netloc or parsed.username or parsed.password or parsed.fragment:
raise ConfirmationViolation("malformed evidence reference")
if parsed.scheme in {"s3", "gs"} and parsed.path in {"", "/"}:
raise ConfirmationViolation("object-store evidence reference needs an object path")
refs.append(ref)
return refs
def canonical_action_digest(action: dict[str, Any]) -> str:
required = {"action_type", "target", "parameters", "tenant_id", "policy_version"}
if not isinstance(action, dict) or set(action) != required:
raise ValueError(f"action must contain exactly {sorted(required)}")
try:
canonical = canonical_json(action)
except (TypeError, ValueError) as exc:
raise ValueError("action is not canonically JSON serializable") from exc
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
def validate_time_window(
claims: dict,
now: int,
policy: dict[str, Any],
) -> tuple[int, int, int]:
timestamps = []
for field in ("iat", "nbf", "exp"):
value = claims.get(field)
if isinstance(value, bool) or not isinstance(value, int):
raise ConfirmationViolation(f"{field} must be an integer timestamp")
timestamps.append(value)
issued_at, not_before, expires_at = timestamps
clock_skew = policy["max_clock_skew_seconds"]
if issued_at > now + clock_skew:
raise ConfirmationViolation("approval is future-issued")
if not_before > now + clock_skew:
raise ConfirmationViolation("approval is not yet valid")
if not_before < issued_at - clock_skew:
raise ConfirmationViolation("approval nbf precedes issuance beyond clock skew")
if not_before > expires_at:
raise ConfirmationViolation("approval expires before its not-before time")
lifetime = expires_at - issued_at
if lifetime <= 0 or lifetime > policy["max_approval_lifetime_seconds"]:
raise ConfirmationViolation("approval lifetime exceeds policy")
if now > expires_at + clock_skew:
raise ConfirmationViolation("approval is expired")
return issued_at, not_before, expires_at
def read_idp_json(response, policy):
try:
chunks, observed = [], 0
for chunk in response.iter_content(chunk_size=policy["idp_response_chunk_bytes"]):
observed += len(chunk)
if observed > policy["idp_response_max_bytes"]:
raise ConfirmationViolation("identity-provider 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 resolve_approver_authentication_event(
*,
event_id: str,
access_token: str,
policy: dict[str, Any],
) -> tuple[dict, str]:
event_id = require_nonempty_string(event_id, "approver_authentication_event_id")
access_token = require_nonempty_string(access_token, "idp_access_token")
url = policy["idp_audit_base_url"].rstrip("/") + "/" + quote(event_id, safe="")
headers = {
"Authorization": "Bearer " + access_token,
"Accept": "application/json",
"Accept-Encoding": "identity",
}
with requests.Session() as session:
session.trust_env = False
with session.get(
url,
headers=headers,
cert=(IDP_CLIENT_CERTIFICATE_PATH, IDP_CLIENT_KEY_PATH),
verify=IDP_CA_BUNDLE_PATH,
timeout=policy["idp_request_timeout_seconds"],
allow_redirects=False,
stream=True,
) as response:
response.raise_for_status()
event = read_idp_json(response, policy)
if not isinstance(event, dict):
raise ConfirmationViolation("identity provider returned a malformed event")
return event, url
def verify_authentication_event(
*,
event: dict,
expected_event_id: str,
expected_subject: str,
approval_issued_at: int,
now: int,
policy: dict[str, Any],
) -> None:
if event.get("issuer") != policy["idp_audit_issuer"]:
raise ConfirmationViolation("untrusted authentication-event issuer")
if not hmac.compare_digest(str(event.get("event_id", "")), expected_event_id):
raise ConfirmationViolation("authentication-event ID mismatch")
if not hmac.compare_digest(str(event.get("subject", "")), expected_subject):
raise ConfirmationViolation("authentication-event subject mismatch")
if event.get("status") != "success":
raise ConfirmationViolation("approver authentication was not successful")
if event.get("assurance_level") not in set(
policy["allowed_auth_assurance_levels"]
):
raise ConfirmationViolation("approver authentication assurance is insufficient")
authenticated_at = event.get("authenticated_at_unix")
expires_at = event.get("expires_at_unix")
if (
isinstance(authenticated_at, bool)
or isinstance(expires_at, bool)
or not isinstance(authenticated_at, int)
or not isinstance(expires_at, int)
):
raise ConfirmationViolation("authentication-event timestamps are malformed")
clock_skew = policy["max_clock_skew_seconds"]
if authenticated_at > approval_issued_at + clock_skew:
raise ConfirmationViolation("approver authentication occurred after approval issuance")
if approval_issued_at - authenticated_at > policy["max_auth_event_age_seconds"]:
raise ConfirmationViolation("approver authentication is too old")
if expires_at <= authenticated_at:
raise ConfirmationViolation("authentication-event lifetime is invalid")
if authenticated_at > now + clock_skew or now > expires_at:
raise ConfirmationViolation("approver authentication event is not currently valid")
def verify_observed_execution(
*,
execution_id: str,
authenticated_executor_principal: str,
action: dict[str, Any],
approval_token: str,
expected_challenge_id: str,
expected_nonce: str,
replay_store: Redis,
idp_access_token: str,
) -> dict[str, Any]:
policy, policy_sha256 = load_signed_policy()
execution_id = require_nonempty_string(execution_id, "execution_id")
authenticated_executor_principal = require_nonempty_string(
authenticated_executor_principal,
"authenticated_executor_principal",
)
expected_challenge_id = require_nonempty_string(
expected_challenge_id,
"expected_challenge_id",
)
expected_nonce = require_nonempty_string(expected_nonce, "expected_nonce")
header = jwt.get_unverified_header(approval_token)
key_id = require_nonempty_string(header.get("kid"), "kid")
public_key = policy["approval_keys"].get(key_id)
if public_key is None:
raise ConfirmationViolation("untrusted approval signing key")
unverified = jwt.decode(approval_token, options={"verify_signature": False})
if unverified.get("iss") != policy["approval_issuer"]:
raise ConfirmationViolation("untrusted approval issuer")
claims = jwt.decode(
approval_token,
public_key,
algorithms=["EdDSA"],
issuer=policy["approval_issuer"],
audience=policy["approval_audience"],
options={"require": REQUIRED},
leeway=policy["max_clock_skew_seconds"],
)
now = int(time.time())
issued_at, _, expires_at = validate_time_window(claims, now, policy)
executor = require_nonempty_string(
claims["executor_principal_id"],
"executor_principal_id",
)
if not hmac.compare_digest(executor, authenticated_executor_principal):
raise ConfirmationViolation("approval is bound to another executor")
challenge_id = require_nonempty_string(claims["challenge_id"], "challenge_id")
nonce = require_nonempty_string(claims["nonce"], "nonce")
if not hmac.compare_digest(challenge_id, expected_challenge_id):
raise ConfirmationViolation("approval challenge mismatch")
if not hmac.compare_digest(nonce, expected_nonce):
raise ConfirmationViolation("approval nonce mismatch")
observed_digest = canonical_action_digest(action)
signed_digest = require_nonempty_string(claims["action_sha256"], "action_sha256")
if not hmac.compare_digest(signed_digest, observed_digest):
raise ConfirmationViolation("executed action differs from approved action")
approver = require_nonempty_string(claims["sub"], "approver subject")
auth_event_id = require_nonempty_string(
claims["approver_authentication_event_id"],
"approver_authentication_event_id",
)
auth_event, auth_event_ref = resolve_approver_authentication_event(
event_id=auth_event_id,
access_token=idp_access_token,
policy=policy,
)
verify_authentication_event(
event=auth_event,
expected_event_id=auth_event_id,
expected_subject=approver,
approval_issued_at=issued_at,
now=now,
policy=policy,
)
evidence_refs = validate_evidence_refs(claims["evidence_refs"], policy)
jti = require_nonempty_string(claims["jti"], "jti")
context = {
"execution_id": execution_id,
"action_sha256": observed_digest,
"challenge_id": challenge_id,
"nonce": nonce,
"approver_principal_id": approver,
"executor_principal_id": executor,
}
encoded_context = canonical_json(context)
ttl = max(expires_at - now + policy["max_clock_skew_seconds"], 1)
replay_key = f"aidefend:approval-monitor:jti:{jti}"
first_delivery = replay_store.set(
replay_key,
encoded_context,
nx=True,
ex=ttl,
)
idempotent_redelivery = False
if not first_delivery:
existing = replay_store.get(replay_key)
if isinstance(existing, bytes):
existing = existing.decode("utf-8")
if existing != encoded_context:
raise ConfirmationViolation(
"approval jti reused across a different execution, action, or challenge context"
)
idempotent_redelivery = True
return {
"schema_version": "aidefend.confirmation_verification.v3",
"status": "pass",
"execution_id": execution_id,
"executor_principal_id": executor,
"approver_principal_id": approver,
"approver_authentication_event_id": auth_event_id,
"approver_authentication_event_ref": auth_event_ref,
"approval_issuer": claims["iss"],
"approval_audience": policy["approval_audience"],
"policy_version": policy["policy_version"],
"policy_sha256": policy_sha256,
"approval_jti": jti,
"challenge_id": challenge_id,
"nonce": nonce,
"action_sha256": observed_digest,
"issued_at": issued_at,
"expires_at": expires_at,
"evidence_refs": evidence_refs,
"signing_key_id": key_id,
"idempotent_redelivery": idempotent_redelivery,
}
class ConfirmationViolation(ValueError):
"""An observed approval binding or authorization violation, not an I/O error."""
def inspect_or_find(**kwargs) -> dict[str, Any]:
try:
return verify_observed_execution(**kwargs)
except (
OSError,
PolicyLoadError,
RedisError,
requests.RequestException,
subprocess.CalledProcessError,
json.JSONDecodeError,
) as exc:
return {
"schema_version": "aidefend.confirmation_verification.v3",
"status": "ERROR",
"execution_id": kwargs.get("execution_id"),
"reason": "policy, identity-provider, or replay-ledger dependency failed",
"error_class": type(exc).__name__,
}
except Exception as exc:
return {
"schema_version": "aidefend.confirmation_verification.v3",
"status": "ANOMALY_FOUND",
"severity": "critical",
"finding_type": "high_risk_confirmation_bypass_or_replay",
"execution_id": kwargs.get("execution_id"),
"reason": str(exc),
}
Complete-population observation requirements
The AI execution-gateway owner emits workload-authenticated high-risk execution records; the identity owner exposes independently authenticated step-up events; detection engineering observes the append-only execution stream after side-effect execution and never authorizes the action. A separately signed collection receipt must bind the exact application/agent asset and version, route and action-class policy versions, fixed window, expected high-risk execution IDs, event count, source checkpoints, completeness flag, and event-set digest. Invoke inspect_or_find for every expected execution and independently re-verify its approval JWT, action digest, challenge, nonce, approver authentication event, evidence references, and replay-ledger context. Re-verify the authorization record; a gateway decision is input evidence, not a detector result.
AID-D-015-G002
Evaluate the complete canonical HITL v3 lifecycle population against signature-verified gate-event records, the signed analytic policy, and authenticated collection completeness to detect approval bypass, invalid terminal semantics, operator bursts, and unattended escalation.
Canonical Model-to-Detect event schema
Read the exact aidefend.hitl_audit_event.v3 records retained by AID-M-006.003 and independently verified aidefend.h018_gate_receipt.v1 artifacts produced by AID-H-018.003. This detector never treats an audit event as authorization. Exact-schema or signature failures are ERROR; authenticated collection gaps are INSUFFICIENT_DATA.
Verify receipts and replay lifecycle analytics
from __future__ import annotations
# File: monitoring/hitl_v3_activity_analytics.py
import hashlib
import json
import re
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
EVENT_FIELDS = {
"schema_version", "event_type", "event_id", "checkpoint_instance_id",
"checkpoint_id", "action_id", "action_digest", "policy_version",
"occurred_at", "lifecycle_sequence", "actor_id", "outcome", "causation_event_id",
"gate_receipt_sha256",
}
EVENT_TYPES = {
"hitl_activation", "hitl_escalation", "hitl_decision",
"hitl_timeout", "gate_receipt_observed", "final_action_observed",
}
OUTCOMES_BY_EVENT = {
"hitl_activation": {"activated"},
"hitl_escalation": {"escalated"},
"hitl_decision": {"approved", "rejected"},
"hitl_timeout": {"timeout"},
"gate_receipt_observed": {"allow", "deny"},
"final_action_observed": {"executed", "cancelled", "failed"},
}
RECEIPT_FIELDS = {
"schema_version", "control", "decision", "action_id", "action_digest",
"identity_context_sha256", "policy_version", "evidence_sha256s",
"issued_at", "expires_at", "nonce",
}
SHA256 = re.compile(r"^[0-9a-f]{64}$")
def parse_time(value: str) -> datetime:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
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 parse_gate_receipt(raw: bytes) -> dict:
receipt = json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=reject_constant,
)
if not isinstance(receipt, dict) or set(receipt) != RECEIPT_FIELDS:
raise ValueError("H-018.003 receipt schema mismatch")
if (
receipt["schema_version"] != "aidefend.h018_gate_receipt.v1"
or receipt["control"] != "AID-H-018.003"
):
raise ValueError("unsupported gate receipt")
if (
receipt["decision"] not in {"allow", "deny"}
or any(
not isinstance(receipt[field], str) or not receipt[field]
for field in (
"action_id", "policy_version", "issued_at",
"expires_at", "nonce",
)
)
or not isinstance(receipt["action_digest"], str)
or not SHA256.fullmatch(receipt["action_digest"])
or not isinstance(receipt["identity_context_sha256"], str)
or not SHA256.fullmatch(receipt["identity_context_sha256"])
or not isinstance(receipt["evidence_sha256s"], list)
or any(
not isinstance(value, str) or not SHA256.fullmatch(value)
for value in receipt["evidence_sha256s"]
)
or len(receipt["evidence_sha256s"])
!= len(set(receipt["evidence_sha256s"]))
):
raise ValueError("gate receipt binding or evidence fields are invalid")
issued_at = parse_time(receipt["issued_at"])
expires_at = parse_time(receipt["expires_at"])
if issued_at >= expires_at:
raise ValueError("gate receipt validity interval is invalid")
return receipt
def analyze(
events: list[dict],
verified_receipts: dict[str, dict],
*,
window_start: datetime,
window_end: datetime,
decision_sla: timedelta,
operator_burst_limit: int,
collection_complete: bool,
expected_policy_version: str,
expected_checkpoint_instance_ids: list[str],
timeout_default: str,
rule_version: str,
) -> dict:
if not collection_complete:
return {"status": "INSUFFICIENT_DATA", "reason": "event collection incomplete"}
if (
window_start.tzinfo is None or window_end.tzinfo is None
or window_end <= window_start or decision_sla <= timedelta(0)
or isinstance(operator_burst_limit, bool)
or not isinstance(operator_burst_limit, int) or operator_burst_limit < 1
or timeout_default not in {"deny", "cancel"}
or not isinstance(expected_checkpoint_instance_ids, list)
or any(
not isinstance(value, str) or not value
for value in expected_checkpoint_instance_ids
)
or len(expected_checkpoint_instance_ids)
!= len(set(expected_checkpoint_instance_ids))
):
raise ValueError("signed analytic inputs are invalid")
expected_instances = set(expected_checkpoint_instance_ids)
expected_set_sha256 = hashlib.sha256(
json.dumps(
sorted(expected_instances), separators=(",", ":"), ensure_ascii=False,
).encode("utf-8")
).hexdigest()
grouped: dict[str, list[dict]] = defaultdict(list)
seen_ids: set[str] = set()
for index, event in enumerate(events):
if not isinstance(event, dict) or set(event) != EVENT_FIELDS:
raise ValueError(f"event {index} schema mismatch")
if (
event["schema_version"] != "aidefend.hitl_audit_event.v3"
or event["event_type"] not in EVENT_TYPES
or event["event_id"] in seen_ids
or event["policy_version"] != expected_policy_version
or event["outcome"] not in OUTCOMES_BY_EVENT.get(event["event_type"], set())
or isinstance(event["lifecycle_sequence"], bool)
or not isinstance(event["lifecycle_sequence"], int)
or event["lifecycle_sequence"] < 0
):
raise ValueError("event version, identity, type, or policy mismatch")
for field in (
"event_id", "checkpoint_instance_id", "checkpoint_id",
"action_id", "policy_version",
):
if not isinstance(event[field], str) or not event[field]:
raise ValueError(f"event {field} is absent")
if (
not isinstance(event["action_digest"], str)
or not SHA256.fullmatch(event["action_digest"])
):
raise ValueError("event action digest is invalid")
if (
event["causation_event_id"] is not None
and (
not isinstance(event["causation_event_id"], str)
or not event["causation_event_id"]
)
):
raise ValueError("event causation identity is invalid")
if (
event["event_type"] == "gate_receipt_observed"
and (
not isinstance(event["gate_receipt_sha256"], str)
or not SHA256.fullmatch(event["gate_receipt_sha256"])
)
):
raise ValueError("gate event receipt digest is invalid")
if (
event["event_type"] != "gate_receipt_observed"
and event["gate_receipt_sha256"] is not None
):
raise ValueError("non-gate event carries a gate receipt digest")
if (
event["event_type"] == "hitl_decision"
and (
not isinstance(event["actor_id"], str)
or not event["actor_id"]
)
):
raise ValueError("HITL decision lacks an authenticated actor")
if (
event["actor_id"] is not None
and (
not isinstance(event["actor_id"], str)
or not event["actor_id"]
)
):
raise ValueError("event actor identity is invalid")
occurred_at = parse_time(event["occurred_at"])
if not window_start <= occurred_at < window_end:
raise ValueError("event timestamp falls outside the signed window")
seen_ids.add(event["event_id"])
grouped[event["checkpoint_instance_id"]].append(event)
if set(grouped) != expected_instances:
raise ValueError(
"authenticated checkpoint-instance population differs from denominator"
)
findings: list[dict] = []
decisions_by_actor: Counter[str] = Counter()
activation_count = 0
mature_checkpoint_count = 0
right_censored_checkpoint_count = 0
for instance_id, lifecycle in sorted(grouped.items()):
ordered = sorted(
lifecycle,
key=lambda item: (
item["lifecycle_sequence"],
parse_time(item["occurred_at"]),
),
)
if [item["lifecycle_sequence"] for item in ordered] != list(range(len(ordered))):
findings.append({
"checkpoint_instance_id": instance_id,
"type": "invalid_lifecycle_sequence",
})
continue
occurred_times = [parse_time(item["occurred_at"]) for item in ordered]
if any(
current < previous
for previous, current in zip(occurred_times, occurred_times[1:])
):
raise ValueError("event occurred_at regresses across lifecycle_sequence")
bindings = {
(
item["checkpoint_id"], item["action_id"],
item["action_digest"], item["policy_version"],
)
for item in ordered
}
if len(bindings) != 1:
findings.append({"checkpoint_instance_id": instance_id, "type": "binding_mismatch"})
continue
state = "START"
invalid_transition = None
for event in ordered:
event_type = event["event_type"]
if state == "START" and event_type == "hitl_activation":
state = "ACTIVE"
elif state == "ACTIVE" and event_type == "hitl_escalation":
state = "ACTIVE" # Explicitly permit repeated escalation while active.
elif state == "ACTIVE" and event_type in {"hitl_decision", "hitl_timeout"}:
state = "TERMINAL"
elif state == "TERMINAL" and event_type == "gate_receipt_observed":
state = "GATED"
elif state == "GATED" and event_type == "final_action_observed":
state = "FINAL"
else:
invalid_transition = {"state": state, "event_type": event_type}
break
if invalid_transition is not None:
findings.append({
"checkpoint_instance_id": instance_id,
"type": "invalid_lifecycle_transition",
**invalid_transition,
})
continue
by_type: dict[str, list[dict]] = defaultdict(list)
previous: str | None = None
for event in ordered:
by_type[event["event_type"]].append(event)
if previous is None:
if event["event_type"] != "hitl_activation" or event["causation_event_id"] is not None:
findings.append({"checkpoint_instance_id": instance_id, "type": "invalid_chain_root"})
elif event["causation_event_id"] != previous:
findings.append({"checkpoint_instance_id": instance_id, "type": "broken_causation_chain"})
previous = event["event_id"]
terminals = by_type["hitl_decision"] + by_type["hitl_timeout"]
gates = by_type["gate_receipt_observed"]
finals = by_type["final_action_observed"]
activations = by_type["hitl_activation"]
if len(activations) != 1:
findings.append({"checkpoint_instance_id": instance_id, "type": "missing_or_duplicate_activation"})
continue
activation_count += 1
if ordered[0]["event_type"] != "hitl_activation":
findings.append({"checkpoint_instance_id": instance_id, "type": "activation_not_first"})
continue
if len(terminals) > 1:
findings.append({"checkpoint_instance_id": instance_id, "type": "duplicate_terminal"})
continue
if not terminals:
if gates or finals:
findings.append({
"checkpoint_instance_id": instance_id,
"type": "gate_or_final_before_terminal",
})
if window_end - parse_time(activations[0]["occurred_at"]) >= decision_sla:
mature_checkpoint_count += 1
findings.append({"checkpoint_instance_id": instance_id, "type": "unattended_escalation"})
else:
right_censored_checkpoint_count += 1
continue
mature_checkpoint_count += 1
terminal = terminals[0]
if (
parse_time(terminal["occurred_at"])
- parse_time(activations[0]["occurred_at"])
> decision_sla
):
findings.append({
"checkpoint_instance_id": instance_id,
"type": "decision_sla_exceeded",
})
if terminal["event_type"] == "hitl_decision" and terminal["actor_id"]:
decisions_by_actor[terminal["actor_id"]] += 1
if len(gates) != 1 or len(finals) != 1:
findings.append({"checkpoint_instance_id": instance_id, "type": "missing_or_duplicate_gate_or_final"})
continue
gate, final = gates[0], finals[0]
positions = {
event["event_id"]: position
for position, event in enumerate(ordered)
}
if not (
positions[terminal["event_id"]]
< positions[gate["event_id"]]
< positions[final["event_id"]]
):
findings.append({
"checkpoint_instance_id": instance_id,
"type": "terminal_gate_final_order_invalid",
})
digest = gate["gate_receipt_sha256"]
receipt = verified_receipts.get(digest)
if receipt is None:
findings.append({"checkpoint_instance_id": instance_id, "type": "missing_verified_gate_receipt"})
continue
if (
receipt["action_id"] != gate["action_id"]
or receipt["action_digest"] != gate["action_digest"]
or receipt["policy_version"] != gate["policy_version"]
or receipt["decision"] != gate["outcome"]
):
findings.append({"checkpoint_instance_id": instance_id, "type": "gate_receipt_binding_mismatch"})
gate_time = parse_time(gate["occurred_at"])
if not (
parse_time(receipt["issued_at"])
<= gate_time
< parse_time(receipt["expires_at"])
):
findings.append({
"checkpoint_instance_id": instance_id,
"type": "gate_receipt_outside_validity",
})
if terminal["event_type"] == "hitl_timeout":
allowed_finals = (
{"cancelled", "failed"}
if timeout_default == "deny" else {"cancelled"}
)
if gate["outcome"] != "deny" or final["outcome"] not in allowed_finals:
findings.append({
"checkpoint_instance_id": instance_id,
"type": "timeout_default_not_enforced",
"timeout_default": timeout_default,
})
else:
if terminal["outcome"] == "rejected" and gate["outcome"] != "deny":
findings.append({
"checkpoint_instance_id": instance_id,
"type": "negative_terminal_was_allowed",
})
if (
gate["outcome"] == "deny"
and final["outcome"] not in {"cancelled", "failed"}
):
findings.append({
"checkpoint_instance_id": instance_id,
"type": "denied_action_executed",
})
if final["outcome"] == "executed" and (
terminal["outcome"] != "approved" or gate["outcome"] != "allow"
):
findings.append({"checkpoint_instance_id": instance_id, "type": "execution_without_approved_gate"})
receipt_use = Counter(
event["gate_receipt_sha256"]
for event in events
if event["event_type"] == "gate_receipt_observed"
)
for digest, count in sorted(receipt_use.items()):
if count > 1:
findings.append({
"gate_receipt_sha256": digest,
"type": "gate_receipt_reused",
"use_count": count,
})
for actor_id, count in decisions_by_actor.items():
if count > operator_burst_limit:
findings.append({
"actor_id": actor_id, "type": "operator_decision_burst",
"decision_count": count,
})
if findings:
status = "ANOMALY_FOUND"
reason = "one or more HITL lifecycle findings exist"
elif activation_count == 0:
status = "INSUFFICIENT_DATA"
reason = "no applicable HITL activation occurred in the authenticated window"
elif right_censored_checkpoint_count:
status = "INSUFFICIENT_DATA"
reason = "one or more HITL activations have not reached the signed decision horizon"
else:
status = "NO_THRESHOLD_BREACH"
reason = "every mature HITL activation completed without a lifecycle finding"
return {
"schema_version": "aidefend.hitl_activity_findings.v3",
"status": status,
"reason": reason,
"rule_version": rule_version,
"timeout_default": timeout_default,
"event_count": len(events),
"checkpoint_instance_count": len(grouped),
"checkpoint_instance_denominator": len(expected_instances),
"expected_checkpoint_instance_set_sha256": expected_set_sha256,
"activation_count": activation_count,
"mature_checkpoint_count": mature_checkpoint_count,
"right_censored_checkpoint_count": right_censored_checkpoint_count,
"findings": findings,
}
Action: Run the evaluator as a separate detection identity. It captures policy and receipt bundles against fixed root-owned keys, binds the receipt to the exact JSONL bytes and complete gate-receipt manifest, and takes live time only from local UTC. A missing mature terminal is a finding; a not-yet-mature activation or empty applicable window is INSUFFICIENT_DATA, never a clean result. The result, v3 bundle, fixed verification key, and completion manifest are committed together under a content-addressed create-once directory. A separate retention identity reads back those exact bytes, verifies the bundle, and may replay the recorded time for audit only; replay cannot emit or replace a live PASS. Do not change approval, execution, isolation, or eviction state from this detector.
Mandatory production envelope
The replay function above is a pure analytic component. Start the evaluator only through a root-owned launcher that has already verified the exact signed runtime profile and written immutable profile, bundle, and admission-receipt snapshots under the fixed admission directory. The child re-verifies both artifacts under pinned keys; it accepts no profile path, limit, timeout, or signer from arguments or environment and has no fallback profile. The analytic policy and ingestion receipt bind the admitted profile digest. Missing, mutable, malformed, stale, future, or mismatched admission evidence is ERROR. Signed limits govern every subsequent read, Cosign deadline, gate-receipt population, and checkpoint denominator. An authenticated receipt that explicitly reports rejected or dead-letter records remains INSUFFICIENT_DATA.
from __future__ import annotations
# File: monitoring/verified_hitl_analytics_inputs.py
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import hashlib
import hmac
import json
import math
import os
from pathlib import Path
import re
import stat
import subprocess
import tempfile
COSIGN = "/usr/local/bin/cosign"
# Structural bootstrap cap for launcher-admitted metadata; policy cannot raise it.
BOOTSTRAP_METADATA_CAPTURE_CEILING = 4 * 1024 * 1024
ADMISSION_ROOT = Path("/run/aidefend/d015/admission")
RUNTIME_PROFILE_PATH = ADMISSION_ROOT / "runtime-profile.json"
RUNTIME_PROFILE_BUNDLE_PATH = ADMISSION_ROOT / "runtime-profile.sigstore.json"
ADMISSION_RECEIPT_PATH = ADMISSION_ROOT / "admission-receipt.json"
ADMISSION_RECEIPT_BUNDLE_PATH = ADMISSION_ROOT / "admission-receipt.sigstore.json"
RUNTIME_PROFILE_KEY = Path("/etc/aidefend/d015/trust/runtime-profile.pub")
ADMISSION_RECEIPT_KEY = Path("/etc/aidefend/d015/trust/runtime-admission-receipt.pub")
POLICY_KEY = Path("/etc/aidefend/d015/trust/hitl-analytics-policy.pub")
RECEIPT_KEY = Path("/etc/aidefend/d015/trust/hitl-ingestion-receipt.pub")
GATE_RECEIPT_KEY = Path("/etc/aidefend/d015/trust/h018-gate-receipt.pub")
RESULT_KEY = Path("/etc/aidefend/d015/trust/hitl-analytics-result.pub")
RESULT_SIGNING_KEY = Path("/run/secrets/aidefend/d015/hitl-analytics-result.key")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
RUNTIME_PROFILE_FIELDS = {
"schema_version", "profile_version", "workload_id",
"cosign_timeout_seconds", "maximum_input_bytes", "input_chunk_bytes",
"maximum_gate_receipts", "maximum_checkpoint_instances",
"issued_at", "expires_at",
}
ADMISSION_RECEIPT_FIELDS = {
"schema_version", "status", "admission_id", "launcher_id", "workload_id",
"profile_version", "profile_sha256", "profile_bundle_sha256",
"profile_verification_key_sha256", "issued_at", "expires_at",
}
POLICY_FIELDS = {
"schema_version", "policy_version", "applicable", "not_applicable_reason",
"asset_id", "asset_version", "expected_stream_id",
"analysis_window_seconds", "decision_sla_seconds",
"operator_burst_limit", "retention_policy_sha256",
"hitl_policy_version", "timeout_default", "runtime_profile_sha256",
"maximum_receipt_age_seconds", "maximum_future_skew_seconds",
}
RECEIPT_FIELDS = {
"schema_version", "asset_id", "asset_version", "stream_id",
"window_start", "window_end", "complete", "event_count",
"events_sha256", "emitted_at", "event_schema_version",
"gate_receipt_sha256s", "expected_checkpoint_instance_ids",
"runtime_profile_sha256", "stream_checkpoint",
"rejected_event_count", "dead_letter_count",
}
@dataclass(frozen=True)
class Captured:
raw: bytes
bundle: bytes
key: bytes
def evidence(self):
return {
"payload_sha256": hashlib.sha256(self.raw).hexdigest(),
"bundle_sha256": hashlib.sha256(self.bundle).hexdigest(),
"verification_key_sha256": hashlib.sha256(self.key).hexdigest(),
}
@dataclass(frozen=True)
class RuntimeContext:
profile_version: str
cosign_timeout_seconds: float
maximum_input_bytes: int
input_chunk_bytes: int
maximum_gate_receipts: int
maximum_checkpoint_instances: int
profile: Captured
admission_receipt: Captured
@property
def profile_sha256(self):
return self.profile.evidence()["payload_sha256"]
def evidence(self):
return {
"runtime_profile": self.profile.evidence(),
"runtime_admission_receipt": self.admission_receipt.evidence(),
}
def reject_constant(value):
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(raw, label):
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 strict_jsonl(raw):
if not raw:
return []
lines = raw.splitlines()
if any(not line.strip() for line in lines):
raise ValueError("event JSONL contains a blank record")
return [strict_json(line, "HITL event") for line in lines]
def parse_time(value):
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, label, *, positive=False):
if isinstance(value, bool) or not isinstance(value, int):
raise ValueError(label + " must be an integer")
if value < (1 if positive else 0):
raise ValueError(label + " is outside its allowed range")
return value
def _read_stable(
path, *, maximum_bytes=None, chunk_bytes=64 * 1024,
allow_empty=False, require_root_owned=False,
):
path = Path(path)
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or (before.st_size == 0 and not allow_empty)
or before.st_size < 0
or (maximum_bytes is not None and before.st_size > maximum_bytes)
):
raise ValueError("input is not a bounded regular file: " + str(path))
if require_root_owned and (before.st_uid != 0 or before.st_mode & 0o022):
raise PermissionError("trusted input is not root-owned and non-writable")
chunks = []
while True:
block = os.read(descriptor, chunk_bytes)
if not block:
break
chunks.append(block)
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 _assert_admission_root():
details = os.stat(ADMISSION_ROOT, follow_symlinks=False)
if (
not stat.S_ISDIR(details.st_mode)
or details.st_uid != 0
or details.st_mode & 0o022
):
raise PermissionError("launcher admission directory is not root-owned and immutable")
def read_admitted(path):
_assert_admission_root()
path = Path(path)
allowed = {
RUNTIME_PROFILE_PATH, RUNTIME_PROFILE_BUNDLE_PATH,
ADMISSION_RECEIPT_PATH, ADMISSION_RECEIPT_BUNDLE_PATH,
}
if path not in allowed or path.parent != ADMISSION_ROOT:
raise PermissionError("runtime bootstrap path is not launcher-controlled")
return _read_stable(
path, maximum_bytes=BOOTSTRAP_METADATA_CAPTURE_CEILING,
require_root_owned=True,
)
def read_trust_root(path):
return _read_stable(
path, maximum_bytes=BOOTSTRAP_METADATA_CAPTURE_CEILING,
require_root_owned=True,
)
def read_stable(path, runtime, *, allow_empty=False, trust=False):
return _read_stable(
path,
maximum_bytes=runtime.maximum_input_bytes,
chunk_bytes=runtime.input_chunk_bytes,
allow_empty=allow_empty,
require_root_owned=trust,
)
def assert_trusted_regular(path):
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
try:
details = os.fstat(descriptor)
if (
not stat.S_ISREG(details.st_mode)
or details.st_uid != 0
or details.st_mode & 0o022
):
raise PermissionError("signing key is not a trusted regular file")
finally:
os.close(descriptor)
def write_private(path, raw):
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as stream:
stream.write(raw)
stream.flush()
os.fsync(stream.fileno())
def _verify_captured(label, captured, timeout_seconds):
with tempfile.TemporaryDirectory(prefix="aidefend-d015-") as name:
root = Path(name)
os.chmod(root, 0o700)
payload = root / "payload"
bundle = root / "payload.sigstore.json"
verify_key = root / "verify.pub"
snapshots = (
(payload, captured.raw),
(bundle, captured.bundle),
(verify_key, captured.key),
)
for path, value in snapshots:
write_private(path, value)
subprocess.run(
[
COSIGN, "verify-blob", "--key", str(verify_key),
"--bundle", str(bundle), str(payload),
],
check=True, capture_output=True, text=True,
timeout=timeout_seconds,
)
for path, value in snapshots:
if not hmac.compare_digest(
_read_stable(path, maximum_bytes=len(value) or 1), value
):
raise RuntimeError(label + " private verification snapshot changed")
def _finite_positive(value, label):
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(label + " must be numeric")
number = float(value)
if not math.isfinite(number) or number <= 0:
raise ValueError(label + " must be finite and positive")
return number
def load_runtime(evaluated_at):
profile_raw = read_admitted(RUNTIME_PROFILE_PATH)
profile_bundle = read_admitted(RUNTIME_PROFILE_BUNDLE_PATH)
profile_key = read_trust_root(RUNTIME_PROFILE_KEY)
profile = strict_json(profile_raw, "runtime profile")
if (
set(profile) != RUNTIME_PROFILE_FIELDS
or profile.get("schema_version") != "aidefend.d015-runtime-profile.v1"
or profile.get("workload_id") != "AID-D-015-G002"
or not isinstance(profile.get("profile_version"), str)
or not profile["profile_version"]
):
raise ValueError("launcher-admitted runtime profile schema differs")
timeout = _finite_positive(
profile["cosign_timeout_seconds"], "cosign_timeout_seconds"
)
for field in (
"maximum_input_bytes", "input_chunk_bytes", "maximum_gate_receipts",
"maximum_checkpoint_instances",
):
integer(profile[field], field, positive=True)
if profile["input_chunk_bytes"] > profile["maximum_input_bytes"]:
raise ValueError("runtime input chunk exceeds maximum input bytes")
issued = parse_time(profile["issued_at"])
expires = parse_time(profile["expires_at"])
if issued > evaluated_at or expires <= issued or evaluated_at >= expires:
raise ValueError("launcher-admitted runtime profile is future, stale, or invalid")
profile_capture = Captured(profile_raw, profile_bundle, profile_key)
if any(
len(value) > profile["maximum_input_bytes"]
for value in (profile_raw, profile_bundle, profile_key)
):
raise ValueError("admitted runtime-profile material exceeds its signed limit")
_verify_captured("runtime profile", profile_capture, timeout)
receipt_raw = read_admitted(ADMISSION_RECEIPT_PATH)
receipt_bundle = read_admitted(ADMISSION_RECEIPT_BUNDLE_PATH)
receipt_key = read_trust_root(ADMISSION_RECEIPT_KEY)
if any(
len(value) > profile["maximum_input_bytes"]
for value in (receipt_raw, receipt_bundle, receipt_key)
):
raise ValueError("runtime-admission material exceeds signed input limit")
receipt_capture = Captured(receipt_raw, receipt_bundle, receipt_key)
_verify_captured("runtime admission receipt", receipt_capture, timeout)
receipt = strict_json(receipt_raw, "runtime admission receipt")
profile_evidence = profile_capture.evidence()
if (
set(receipt) != ADMISSION_RECEIPT_FIELDS
or receipt.get("schema_version")
!= "aidefend.d015-runtime-admission-receipt.v1"
or receipt.get("status") != "ADMITTED"
or receipt.get("workload_id") != "AID-D-015-G002"
or receipt.get("profile_version") != profile["profile_version"]
or not isinstance(receipt.get("admission_id"), str)
or not receipt["admission_id"]
or not isinstance(receipt.get("launcher_id"), str)
or not receipt["launcher_id"]
or not isinstance(receipt.get("profile_sha256"), str)
or not SHA256.fullmatch(receipt["profile_sha256"])
or not hmac.compare_digest(
receipt["profile_sha256"],
profile_evidence["payload_sha256"],
)
or not isinstance(receipt.get("profile_bundle_sha256"), str)
or not SHA256.fullmatch(receipt["profile_bundle_sha256"])
or not hmac.compare_digest(
receipt["profile_bundle_sha256"],
profile_evidence["bundle_sha256"],
)
or not isinstance(receipt.get("profile_verification_key_sha256"), str)
or not SHA256.fullmatch(receipt["profile_verification_key_sha256"])
or not hmac.compare_digest(
receipt["profile_verification_key_sha256"],
profile_evidence["verification_key_sha256"],
)
):
raise ValueError("runtime admission receipt is malformed or misbound")
receipt_issued = parse_time(receipt["issued_at"])
receipt_expires = parse_time(receipt["expires_at"])
if (
receipt_issued < issued or receipt_issued > evaluated_at
or receipt_expires <= receipt_issued or receipt_expires > expires
or evaluated_at >= receipt_expires
):
raise ValueError("runtime admission receipt is future, stale, or invalid")
return RuntimeContext(
profile_version=profile["profile_version"],
cosign_timeout_seconds=timeout,
maximum_input_bytes=profile["maximum_input_bytes"],
input_chunk_bytes=profile["input_chunk_bytes"],
maximum_gate_receipts=profile["maximum_gate_receipts"],
maximum_checkpoint_instances=profile["maximum_checkpoint_instances"],
profile=profile_capture,
admission_receipt=receipt_capture,
)
def capture_signed(
label, payload_path, bundle_path, key_path, runtime, *, expected_sha=None,
):
raw = read_stable(payload_path, runtime, allow_empty=False)
bundle = read_stable(bundle_path, runtime)
key = read_stable(key_path, runtime, trust=True)
if expected_sha is not None and not hmac.compare_digest(
hashlib.sha256(raw).hexdigest(), expected_sha
):
raise ValueError(label + " digest differs from signed manifest")
captured = Captured(raw, bundle, key)
_verify_captured(label, captured, runtime.cosign_timeout_seconds)
return captured
def load_policy(captured, runtime):
policy = strict_json(captured.raw, "HITL analytic policy")
if (
set(policy) != POLICY_FIELDS
or policy.get("schema_version") != "aidefend.hitl-analytics-policy.v5"
or not isinstance(policy.get("applicable"), bool)
):
raise ValueError("HITL analytic policy schema differs")
runtime_profile_sha256 = policy.get("runtime_profile_sha256")
if (
not isinstance(runtime_profile_sha256, str)
or not SHA256.fullmatch(runtime_profile_sha256)
or not hmac.compare_digest(runtime_profile_sha256, runtime.profile_sha256)
):
raise ValueError("analytic policy is bound to another runtime profile")
for field in ("policy_version", "asset_id", "asset_version"):
if not isinstance(policy[field], str) or not policy[field]:
raise ValueError("HITL policy identity is absent")
reason = policy["not_applicable_reason"]
if policy["applicable"] and reason is not None:
raise ValueError("applicable policy contains a not-applicable reason")
if not policy["applicable"] and (
not isinstance(reason, str) or not reason.strip()
):
raise ValueError("not-applicable policy lacks a reason")
if policy["applicable"]:
for field in ("expected_stream_id", "hitl_policy_version"):
if not isinstance(policy[field], str) or not policy[field]:
raise ValueError("applicable HITL stream identity is absent")
elif policy["expected_stream_id"] is not None:
raise ValueError("not-applicable policy names a stream")
if (
policy["timeout_default"] not in {"deny", "cancel"}
or not isinstance(policy["retention_policy_sha256"], str)
or not SHA256.fullmatch(policy["retention_policy_sha256"])
):
raise ValueError("HITL policy binding is invalid")
for field in (
"analysis_window_seconds", "decision_sla_seconds",
"operator_burst_limit", "maximum_receipt_age_seconds",
):
integer(policy[field], field, positive=True)
integer(policy["maximum_future_skew_seconds"], "maximum_future_skew_seconds")
if policy["decision_sla_seconds"] > policy["analysis_window_seconds"]:
raise ValueError("decision SLA exceeds the analysis window")
return policy
def verified_inputs(
*, policy_path, policy_bundle, events_path, receipt_path, receipt_bundle,
evaluated_at, runtime,
):
policy_capture = capture_signed(
"analytic policy", policy_path, policy_bundle, POLICY_KEY, runtime
)
policy = load_policy(policy_capture, runtime)
policy_sha = policy_capture.evidence()["payload_sha256"]
if not policy["applicable"]:
return {
"status": "INSUFFICIENT_DATA",
"reason": (
"detector policy cannot establish product applicability: "
+ policy["not_applicable_reason"]
),
"asset_id": policy["asset_id"],
"asset_version": policy["asset_version"],
"policy_sha256": policy_sha,
"policy_version": policy["policy_version"],
"input_evidence": {"policy": policy_capture.evidence()},
}
receipt_capture = capture_signed(
"ingestion receipt", receipt_path, receipt_bundle, RECEIPT_KEY, runtime
)
receipt = strict_json(receipt_capture.raw, "HITL ingestion receipt")
events_raw = read_stable(events_path, runtime, allow_empty=True)
events = strict_jsonl(events_raw)
if (
set(receipt) != RECEIPT_FIELDS
or receipt.get("schema_version") != "aidefend.hitl-ingestion-receipt.v5"
or receipt.get("event_schema_version") != "aidefend.hitl_audit_event.v3"
or receipt.get("asset_id") != policy["asset_id"]
or receipt.get("asset_version") != policy["asset_version"]
or receipt.get("stream_id") != policy["expected_stream_id"]
or not isinstance(receipt.get("runtime_profile_sha256"), str)
or not SHA256.fullmatch(receipt["runtime_profile_sha256"])
or not hmac.compare_digest(
receipt["runtime_profile_sha256"], runtime.profile_sha256
)
or not isinstance(receipt.get("complete"), bool)
or not isinstance(receipt.get("stream_checkpoint"), str)
or not receipt["stream_checkpoint"]
):
raise ValueError("HITL ingestion receipt is invalid or misbound")
for field in ("event_count", "rejected_event_count", "dead_letter_count"):
integer(receipt[field], "receipt." + field)
events_sha = hashlib.sha256(events_raw).hexdigest()
if (
receipt["event_count"] != len(events)
or not isinstance(receipt["events_sha256"], str)
or not SHA256.fullmatch(receipt["events_sha256"])
or not hmac.compare_digest(receipt["events_sha256"], events_sha)
):
raise ValueError("receipt does not bind the exact event bytes and count")
digests = receipt["gate_receipt_sha256s"]
if (
not isinstance(digests, list)
or any(not isinstance(value, str) or not SHA256.fullmatch(value) for value in digests)
or len(digests) != len(set(digests))
or len(digests) > runtime.maximum_gate_receipts
):
raise ValueError("gate receipt manifest is invalid")
expected_instances = receipt["expected_checkpoint_instance_ids"]
if (
not isinstance(expected_instances, list)
or any(
not isinstance(value, str) or not value
for value in expected_instances
)
or len(expected_instances) != len(set(expected_instances))
or len(expected_instances) > runtime.maximum_checkpoint_instances
):
raise ValueError("checkpoint-instance denominator is invalid")
expected_set_sha256 = hashlib.sha256(
json.dumps(
sorted(expected_instances), separators=(",", ":"), ensure_ascii=False,
).encode("utf-8")
).hexdigest()
start = parse_time(receipt["window_start"])
end = parse_time(receipt["window_end"])
emitted = parse_time(receipt["emitted_at"])
future = timedelta(seconds=policy["maximum_future_skew_seconds"])
if (
end <= start
or int((end - start).total_seconds()) != policy["analysis_window_seconds"]
or emitted < end
or emitted > evaluated_at + future
):
raise ValueError("receipt window or issuance time is invalid")
if (evaluated_at - emitted).total_seconds() > policy["maximum_receipt_age_seconds"]:
status, reason = "INSUFFICIENT_DATA", "ingestion receipt is stale"
elif (
receipt["complete"] is not True
or receipt["rejected_event_count"]
or receipt["dead_letter_count"]
):
status, reason = "INSUFFICIENT_DATA", "authenticated HITL collection is incomplete"
else:
status, reason = "READY", None
return {
"status": status, "reason": reason,
"asset_id": policy["asset_id"], "asset_version": policy["asset_version"],
"events": events, "events_raw": events_raw,
"events_sha256": events_sha, "gate_receipt_sha256s": digests,
"window_start": start, "window_end": end,
"decision_sla": timedelta(seconds=policy["decision_sla_seconds"]),
"operator_burst_limit": policy["operator_burst_limit"],
"timeout_default": policy["timeout_default"],
"expected_checkpoint_instance_ids": expected_instances,
"expected_checkpoint_instance_count": len(expected_instances),
"expected_checkpoint_instance_set_sha256": expected_set_sha256,
"rule_version": policy["policy_version"],
"expected_hitl_policy_version": policy["hitl_policy_version"],
"policy_sha256": policy_sha,
"receipt_sha256": receipt_capture.evidence()["payload_sha256"],
"retention_policy_sha256": policy["retention_policy_sha256"],
"stream_checkpoint": receipt["stream_checkpoint"],
"input_evidence": {
"policy": policy_capture.evidence(),
"receipt": receipt_capture.evidence(),
"events": {"payload_sha256": events_sha, "event_count": len(events)},
},
}
Run the verified workflow
from __future__ import annotations
# File: monitoring/evaluate_verified_hitl_window.py
import argparse
import ctypes
import errno
from datetime import datetime, timezone
import hashlib
import hmac
import json
import os
from pathlib import Path
import stat
import subprocess
import tempfile
import sys
from monitoring.hitl_v3_activity_analytics import analyze, parse_gate_receipt
from monitoring.verified_hitl_analytics_inputs import (
COSIGN, GATE_RECEIPT_KEY, RESULT_KEY, RESULT_SIGNING_KEY,
assert_trusted_regular, capture_signed, load_runtime, read_stable,
verified_inputs, write_private,
)
RESULT_BASE = Path("/var/lib/aidefend/evidence/hitl-analytics-results")
EXIT_CODES = {
"PASS": 0, "FAIL": 1, "INSUFFICIENT_DATA": 2,
"NOT_APPLICABLE": 0, "ERROR": 3,
}
AT_FDCWD = -100
RENAME_NOREPLACE = 1
def canonical(value):
return json.dumps(
value, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
def fsync_directory(path):
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def fsync_file(path):
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
try:
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
raise ValueError("staged member is not a regular file")
os.fsync(descriptor)
finally:
os.close(descriptor)
def rename_noreplace(source, target):
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 verify_private_result(result_path, bundle_path, key_path, runtime):
subprocess.run(
[
COSIGN, "verify-blob", "--key", str(key_path),
"--bundle", str(bundle_path), str(result_path),
],
check=True, capture_output=True, text=True,
timeout=runtime.cosign_timeout_seconds,
)
def publish_result(value, runtime):
raw = canonical(value) + b"\n"
result_sha = hashlib.sha256(raw).hexdigest()
base_details = os.stat(RESULT_BASE, follow_symlinks=False)
if (
not stat.S_ISDIR(base_details.st_mode)
or base_details.st_uid != os.geteuid()
or base_details.st_mode & 0o022
):
raise PermissionError(
"result base must be a pre-provisioned, owner-controlled directory"
)
staging = Path(tempfile.mkdtemp(prefix="." + result_sha + ".", dir=RESULT_BASE))
os.chmod(staging, 0o700)
target = RESULT_BASE / result_sha
try:
result_path = staging / "result.json"
bundle_path = staging / "result.sigstore.json"
key_path = staging / "result.pub"
write_private(result_path, raw)
key_raw = read_stable(RESULT_KEY, runtime, trust=True)
write_private(key_path, key_raw)
assert_trusted_regular(RESULT_SIGNING_KEY)
subprocess.run(
[
COSIGN, "sign-blob", "--yes",
"--key", str(RESULT_SIGNING_KEY),
"--bundle", str(bundle_path), str(result_path),
],
check=True, timeout=runtime.cosign_timeout_seconds,
)
fsync_file(bundle_path)
os.chmod(bundle_path, 0o400)
os.chmod(result_path, 0o400)
os.chmod(key_path, 0o400)
verify_private_result(result_path, bundle_path, key_path, runtime)
bundle_raw = read_stable(bundle_path, runtime)
if not hmac.compare_digest(read_stable(result_path, runtime), raw):
raise RuntimeError("private result readback differs")
completion = {
"schema_version": "aidefend.signed-result-publication.v1",
"payload_sha256": result_sha,
"bundle_sha256": hashlib.sha256(bundle_raw).hexdigest(),
"verification_key_sha256": hashlib.sha256(key_raw).hexdigest(),
}
completion_raw = canonical(completion) + b"\n"
complete_path = staging / "complete.json"
write_private(complete_path, completion_raw)
os.chmod(complete_path, 0o400)
for member in (result_path, bundle_path, key_path, complete_path):
fsync_file(member)
expected = {
"result.json": raw,
"result.sigstore.json": bundle_raw,
"result.pub": key_raw,
"complete.json": completion_raw,
}
if {path.name for path in staging.iterdir()} != set(expected):
raise RuntimeError("staging directory contains an unexpected member")
os.chmod(staging, 0o750)
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")
for name, expected_raw in expected.items():
if not hmac.compare_digest(read_stable(target / name, runtime), expected_raw):
raise RuntimeError("published result readback differs: " + name)
verify_private_result(
target / "result.json",
target / "result.sigstore.json",
target / "result.pub",
runtime,
)
return target
finally:
if staging is not None and staging.exists():
for path in staging.iterdir():
path.unlink()
staging.rmdir()
def evaluate(arguments, evaluated_at, runtime):
inputs = verified_inputs(
policy_path=arguments.policy,
policy_bundle=arguments.policy_bundle,
events_path=arguments.events,
receipt_path=arguments.receipt,
receipt_bundle=arguments.receipt_bundle,
evaluated_at=evaluated_at, runtime=runtime,
)
base = {
"schema_version": "aidefend.d015-hitl-evidence.v3",
"control": "AID-D-015",
"guidance": "canonical-hitl-lifecycle-analytics",
"asset_id": inputs["asset_id"],
"asset_version": inputs["asset_version"],
"policy_sha256": inputs["policy_sha256"],
"input_evidence": {**runtime.evidence(), **inputs["input_evidence"]},
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
}
if inputs["status"] != "READY":
return {**base, "status": inputs["status"], "reason": inputs["reason"]}
verified_receipts = {}
gate_receipt_evidence = {}
for digest in inputs["gate_receipt_sha256s"]:
artifact = arguments.gate_receipt_dir / (digest + ".json")
bundle = arguments.gate_receipt_dir / (digest + ".sigstore.json")
captured = capture_signed(
"gate receipt", artifact, bundle, GATE_RECEIPT_KEY, runtime,
expected_sha=digest,
)
receipt = parse_gate_receipt(captured.raw)
verified_receipts[digest] = receipt
gate_receipt_evidence[digest] = captured.evidence()
observed_digests = {
event["gate_receipt_sha256"]
for event in inputs["events"]
if event.get("event_type") == "gate_receipt_observed"
}
if observed_digests != set(verified_receipts):
raise ValueError("verified gate-receipt population differs from events")
analytic = analyze(
inputs["events"],
verified_receipts,
window_start=inputs["window_start"],
window_end=inputs["window_end"],
decision_sla=inputs["decision_sla"],
operator_burst_limit=inputs["operator_burst_limit"],
collection_complete=True,
expected_policy_version=inputs["expected_hitl_policy_version"],
expected_checkpoint_instance_ids=(
inputs["expected_checkpoint_instance_ids"]
),
timeout_default=inputs["timeout_default"],
rule_version=inputs["rule_version"],
)
if analytic["status"] == "ANOMALY_FOUND":
status, reason = "FAIL", "HITL lifecycle finding exists"
elif analytic["status"] == "INSUFFICIENT_DATA":
status, reason = "INSUFFICIENT_DATA", analytic["reason"]
elif analytic["status"] == "NO_THRESHOLD_BREACH":
status, reason = "PASS", analytic["reason"]
else:
raise ValueError("analytic returned an unsupported state")
receipt_set_sha = hashlib.sha256(
("\n".join(sorted(verified_receipts)) + "\n").encode("utf-8")
).hexdigest()
return {
**base,
"status": status,
"reason": reason,
"ingestion_receipt_sha256": inputs["receipt_sha256"],
"events_sha256": inputs["events_sha256"],
"retention_policy_sha256": inputs["retention_policy_sha256"],
"stream_checkpoint": inputs["stream_checkpoint"],
"event_population": len(inputs["events"]),
"gate_receipt_population": len(verified_receipts),
"gate_receipt_set_sha256": receipt_set_sha,
"checkpoint_instance_denominator": (
inputs["expected_checkpoint_instance_count"]
),
"expected_checkpoint_instance_set_sha256": (
inputs["expected_checkpoint_instance_set_sha256"]
),
"input_evidence": {
**base["input_evidence"],
"gate_receipts": gate_receipt_evidence,
},
"analytic": analytic,
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--policy", type=Path, required=True)
parser.add_argument("--policy-bundle", type=Path, required=True)
parser.add_argument("--events", type=Path, required=True)
parser.add_argument("--receipt", type=Path, required=True)
parser.add_argument("--receipt-bundle", type=Path, required=True)
parser.add_argument("--gate-receipt-dir", type=Path, required=True)
arguments = parser.parse_args()
evaluated_at = datetime.now(timezone.utc)
try:
runtime = load_runtime(evaluated_at)
except Exception as error:
result = {
"schema_version": "aidefend.d015-hitl-evidence.v3",
"control": "AID-D-015",
"guidance": "canonical-hitl-lifecycle-analytics",
"status": "ERROR",
"reason": "launcher runtime admission failed",
"error_class": type(error).__name__,
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
}
print(canonical(result).decode("utf-8"), file=sys.stderr)
return EXIT_CODES["ERROR"]
try:
result = evaluate(arguments, evaluated_at, runtime)
except Exception as error:
result = {
"schema_version": "aidefend.d015-hitl-evidence.v3",
"control": "AID-D-015",
"guidance": "canonical-hitl-lifecycle-analytics",
"status": "ERROR",
"reason": "input verification or evaluation failed",
"error_class": type(error).__name__,
"input_evidence": runtime.evidence(),
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
}
try:
target = publish_result(result, runtime)
except Exception as error:
failure = {
"schema_version": "aidefend.d015-hitl-evidence.v3",
"control": "AID-D-015",
"guidance": "canonical-hitl-lifecycle-analytics",
"status": "ERROR",
"reason": "signed result publication failed",
"error_class": type(error).__name__,
"runtime_profile_sha256": runtime.profile_sha256,
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
}
print(canonical(failure).decode("utf-8"), file=sys.stderr)
return EXIT_CODES["ERROR"]
print(target)
return EXIT_CODES[result["status"]]
if __name__ == "__main__":
raise SystemExit(main())
Open-source tools
- Keycloak
- OpenTelemetry
- Apache Kafka
- Yubico python-fido2
- PyJWT
- Requests (approval-service and issuer readback transport)
- Valkey
- Redis 8+ AGPLv3 source build
- Cosign
Commercial / hosted tools
- Okta Workforce Identity
- Cisco Duo
- Microsoft Entra ID
- Splunk Enterprise Security
Threat mappings
MITRE ATLAS
- AML.T0053 AI Agent Tool Invocation (confirmation telemetry surfaces unapproved high-risk tool execution)
- AML.T0086 Exfiltration via AI Agent Tool Invocation (approval-bypass detection surfaces unconfirmed export attempts)
MAESTRO
- Agent Tool Misuse (L7)
- Data Exfiltration (L2) (confirmation telemetry surfaces coerced or bypassed approvals around export actions)
OWASP LLM Top 10 2026
- LLM03:2026 Excessive Agency
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI09:2026 Human-Agent Trust Exploitation
- ASI02:2026 Tool Misuse and Exploitation
- ASI03:2026 Identity and Privilege Abuse
NIST Adversarial Machine Learning 2025
- NISTAML.039 Compromising connected resources
Cisco Integrated AI Security and Safety Framework
- AISubtech-14.2.1 Permission Escalation via Delegation (delegated high-risk actions lacking matching human-approval evidence)
- AISubtech-2.1.1 Execution Approval Bypass
- AISubtech-4.3.4 Replay Exploitation
- AITech-14.2 Abuse of Delegated Authority (delegated high-risk actions lacking matching human-approval evidence)
- AITech-2.1 Excessive Agency
Google Secure AI Framework 2.0 - Risks
- RA: Rogue Actions (confirmation telemetry surfaces rogue high-risk execution without the expected approval trail)
Databricks AI Security Framework 3.0
- Agents - Core 13.10: Overwhelming Human in the Loop
- Model Serving - Inference requests 9.13: Excessive agency
- Agents - Core 13.3: Privilege Compromise (missing-confirmation telemetry exposes privilege abuse)