AID-H-018.006: Continuous Authorization Verification (Anti-TOCTOU)
Prevent Time-of-Check to Time-of-Use (TOCTOU) failures in agentic workflows by re-verifying authorization at execution time for each sensitive step (not only at workflow start). Bind authorization decisions to an immutable execution context (session/task/plan + delegation chain) and invalidate stale decisions on context change.
Pillars: app
Lifecycle: operation
Scope boundary
Owns fresh execution-time re-verification of authorization, approval nonce, immutable action context, delegation, target resource, policy, and live preconditions before every sensitive dispatch and after pause, retry, resume, failover, or handoff.
AID-H-018.003: AID-H-018.006 proves the approved action is still authorized and unchanged immediately before use; AID-H-018.003 performs independent validation and issues the high-impact approval receipt. A valid receipt must still be fresh, unused, correctly bound, and consistent with live resource state.
AID-H-018.008: AID-H-018.006 owns authorization freshness and anti-TOCTOU binding; AID-H-018.008 owns immutable operation identity, duplicate suppression, effect state, and uncertain-outcome reconciliation. Fresh authorization does not prevent the same effect from being dispatched twice.
AID-I-008.002: AID-H-018.006 supplies fresh execution-time authorization for the re-observed action; AID-I-008.002 owns browser request suspension and exact DOM, accessibility, method, URL, body, origin, and action re-observation. Generic authorization freshness and browser-specific observation are composed at dispatch.
AID-H-018.009: AID-H-018.006 proves authorization and live action context remain current immediately before dispatch; AID-H-018.009 binds a filesystem operation to the descriptor or handle of the object actually opened or affected. Fresh authorization and filesystem-object confinement are composed at use but produce separate evidence and outcomes.
Implementation guidance
- AID-H-018.006-G001: Re-check the canonical action and current signed authorization bundle immediately before every sensitive dispatch, atomically consume the one-time gate-decision nonce from the verified AID-H-018.003 receipt, and enforce the live resource precondition.
- AID-H-018.006-G002: Expire authorization decisions quickly and require a fresh verification after pause, resume, retry, failover, worker handoff, or execution-context change.
- AID-H-018.006-G003: Re-check session, delegation, plan, approval, target-resource, and policy invariants before each high-risk step and fail closed on any mismatch.
AID-H-018.006-G001
Re-check the canonical action and current signed authorization bundle immediately before every sensitive dispatch, atomically consume the one-time gate-decision nonce from the verified AID-H-018.003 receipt, and enforce the live resource precondition.
Complete deployment unit: execution-time verifier and dispatcher
This adapter is the last trusted hop before a state-changing provider. It replays the full authorization bundle instead of trusting a prior boolean. It accepts only the aidefend.h018.gate_decision.v3 record, whose exact field set (including effect_preview_sha256 and effect_summary_sha256) is a shared contract with AID-H-018.003-G001 and its replay verifier: when the gate recorded a preview, the bundle must contain exactly that preview artifact, its summary digest and pre-state binding must match, and every approval must be signed against the same two digests; when it recorded none, no preview or preview-bound approval may be present. Change the field set, the schema version, and this loader together, and keep a test that feeds a freshly issued decision through this loader. The only caller-controlled selector is a validated request ID. Trust roots, signed policy/identity/provider registries, keys, TLS credentials, inbox, Redis socket, and evidence outbox are root-owned fixed paths. Provider calls use mutually authenticated TLS with redirects disabled; provider responses are canonical signed artifacts from identities in the current registry.
# File: execution_gate/verify_and_dispatch.py
from __future__ import annotations
import hashlib
import json
import math
import os
import re
import secrets
import ssl
import subprocess
import tempfile
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit
import httpx
import redis
def required_positive_seconds(name: str) -> float:
value = float(os.environ[name])
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"{name} must be a finite positive duration")
return value
def required_positive_int(name: str) -> int:
value = int(os.environ[name])
if value <= 0:
raise RuntimeError(f"{name} must be a positive integer")
return value
COSIGN = "/usr/local/bin/cosign"
COSIGN_TIMEOUT_SECONDS = required_positive_seconds("VERIFIED_H018_EXECUTION_COSIGN_TIMEOUT_SECONDS")
REMOTE_TOTAL_TIMEOUT_SECONDS = required_positive_seconds("VERIFIED_H018_REMOTE_TOTAL_TIMEOUT_SECONDS")
REMOTE_CONNECT_TIMEOUT_SECONDS = required_positive_seconds("VERIFIED_H018_REMOTE_CONNECT_TIMEOUT_SECONDS")
REMOTE_RESPONSE_MAX_BYTES = required_positive_int("VERIFIED_H018_REMOTE_RESPONSE_MAX_BYTES")
CONTROL_DIR = Path("/etc/aidefend/h018")
ROOT_VERIFY_KEY = CONTROL_DIR / "trust-root.pub"
POLICY_PATH = CONTROL_DIR / "gate-policy.json"
REGISTRY_PATH = CONTROL_DIR / "identity-key-registry.json"
PROVIDER_PATH = CONTROL_DIR / "provider-registry.json"
KEY_DIR = (CONTROL_DIR / "keys").resolve()
TLS_DIR = (CONTROL_DIR / "tls").resolve()
GATE_VERIFY_KEY = CONTROL_DIR / "keys/h018-gate.pub"
GATE_REPLAY_VERIFY_KEY = CONTROL_DIR / "keys/h018-replay.pub"
EXECUTION_SIGNING_KEY = Path("/run/secrets/aidefend/h018-execution-signing.key")
EXECUTION_VERIFY_KEY = CONTROL_DIR / "keys/h018-execution.pub"
INBOX = Path("/var/lib/aidefend/h018/inbox")
OUTBOX = Path("/var/lib/aidefend/h018/evidence/execution-decisions")
REDIS_SOCKET = "/run/aidefend-h018/redis.sock"
REQUEST_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
TOKEN = re.compile(r"^[A-Za-z0-9._:-]{1,256}$")
SHA256 = re.compile(r"^[a-f0-9]{64}$")
SIGNATURE = re.compile(r"^[A-Za-z0-9+/=_-]{20,8192}$")
ACTION_FIELDS = {
"schema_version", "action_id", "action_profile", "evidence_profile",
"subject_id", "workload_id", "session_id", "task_id", "tool",
"target", "parameters", "state_binding", "requested_at",
}
GATE_FIELDS = {
"schema_version", "control", "decision_id", "request_id", "outcome",
"reason_code", "action_id", "action_sha256", "policy_version",
"policy_sha256", "registry_sha256", "evidence_sha256s",
"evidence_population_sha256", "evidence_population_count",
"approval_nonce_hashes", "effect_preview_sha256", "effect_summary_sha256",
"issued_at", "expires_at", "gate_nonce",
}
REPLAY_FIELDS = {
"schema_version", "control", "decision_sha256", "request_id",
"action_sha256", "artifact_population_sha256",
"artifact_population_count", "replay_outcome", "observed_at",
}
ENVELOPE_FIELDS = {
"schema_version", "artifact_type", "issuer_id", "key_id", "issued_at",
"expires_at", "nonce", "claims",
}
COMMON_CLAIMS = {"action_id", "action_sha256", "policy_version", "evidence_profile"}
CLAIM_FIELDS = {
"identity_context": COMMON_CLAIMS | {
"subject_id", "workload_id", "delegation_chain",
},
"validator_decision": COMMON_CLAIMS | {
"decision", "observed_population_sha256", "observed_population_count",
},
"approval": COMMON_CLAIMS | {
"approver_id", "role", "assurance", "decision",
"effect_preview_sha256", "summary_sha256",
},
"control_evidence": COMMON_CLAIMS | {
"evidence_kind", "result", "subject", "observed_population_sha256",
"observed_population_count",
},
"state_observation": COMMON_CLAIMS | {
"phase", "resource_id", "version", "state_sha256", "observed_at",
"observed_population_sha256", "observed_population_count", "effect_id",
"execution_receipt_sha256",
},
"execution_receipt": COMMON_CLAIMS | {
"gate_decision_sha256", "pre_observation_sha256",
"precondition_version", "precondition_result", "effect_status",
"effect_id", "provider_receipt_sha256",
},
}
EFFECT_PREVIEW_KIND = "effect_preview"
STATE_FIELDS = {
"provider_profile", "resource_id", "expected_pre_version",
"expected_pre_state_sha256", "expected_post_version",
"expected_post_state_sha256", "maximum_observation_age_seconds",
}
class StopExecution(Exception):
def __init__(self, outcome: str, reason_code: str):
super().__init__(reason_code)
self.outcome = outcome
self.reason_code = reason_code
def stop(outcome: str, reason_code: str) -> None:
raise StopExecution(outcome, reason_code)
def reject_floats(value: Any) -> None:
if isinstance(value, float):
stop("FAIL", "floating_point_not_allowed")
if isinstance(value, list):
for item in value:
reject_floats(item)
if isinstance(value, dict):
for key, item in value.items():
if not isinstance(key, str):
stop("FAIL", "non_string_object_key")
reject_floats(item)
def canonical(value: dict[str, Any] | list[Any]) -> bytes:
reject_floats(value)
return (json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
allow_nan=False,
) + "\n").encode("utf-8")
def sha256(raw: bytes) -> str:
return hashlib.sha256(raw).hexdigest()
def parse_time(value: Any) -> datetime:
if not isinstance(value, str):
stop("FAIL", "timestamp_type_invalid")
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
stop("FAIL", "timestamp_invalid")
if parsed.utcoffset() is None:
stop("FAIL", "timestamp_timezone_missing")
return parsed.astimezone(timezone.utc)
def signature_path(path: Path) -> Path:
return Path(str(path) + ".sig")
def verify_blob_bytes(raw: bytes, signature: Path, key: Path) -> None:
if not signature.is_file() or not key.is_file():
stop("INSUFFICIENT_DATA", "signed_input_missing")
snapshot: Path | None = None
try:
with tempfile.NamedTemporaryFile(
prefix="h018-signed-input-", suffix=".json", delete=False
) as handle:
handle.write(raw)
handle.flush()
os.fsync(handle.fileno())
snapshot = Path(handle.name)
subprocess.run(
[COSIGN, "verify-blob", "--key", str(key),
"--bundle", str(signature), str(snapshot)],
check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS,
)
except subprocess.CalledProcessError:
stop("FAIL", "signature_invalid")
finally:
if snapshot is not None:
snapshot.unlink(missing_ok=True)
def verify_blob(blob: Path, signature: Path, key: Path) -> bytes:
if not blob.is_file():
stop("INSUFFICIENT_DATA", "signed_input_missing")
raw = blob.read_bytes()
verify_blob_bytes(raw, signature, key)
return raw
def load_canonical_bytes(
raw: bytes, fields: set[str],
) -> tuple[dict[str, Any], bytes, str]:
try:
def unique_pairs(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate artifact key: {key}")
value[key] = item
return value
def reject_constant(value):
raise ValueError(f"non-finite artifact number: {value}")
document = json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_pairs, parse_constant=reject_constant,
)
except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
stop("FAIL", "artifact_json_invalid")
if not isinstance(document, dict) or set(document) != fields:
stop("FAIL", "artifact_schema_mismatch")
if canonical(document) != raw:
stop("FAIL", "artifact_not_canonical")
return document, raw, sha256(raw)
def load_canonical(
path: Path, fields: set[str],
) -> tuple[dict[str, Any], bytes, str]:
if not path.is_file():
stop("INSUFFICIENT_DATA", "required_artifact_missing")
return load_canonical_bytes(path.read_bytes(), fields)
def load_root_signed(path: Path, fields: set[str]) -> tuple[dict[str, Any], str]:
raw = verify_blob(path, signature_path(path), ROOT_VERIFY_KEY)
document, _, digest = load_canonical_bytes(raw, fields)
return document, digest
def active(record: dict[str, Any], now: datetime) -> bool:
if record.get("status") != "active" or parse_time(record.get("valid_until")) <= now:
return False
revoked_at = record.get("revoked_at")
return revoked_at is None or now < parse_time(revoked_at)
def require_identity(
registry: dict[str, Any], identity_id: str, now: datetime, role: str | None = None,
) -> dict[str, Any]:
record = registry["identities"].get(identity_id)
fields = {"status", "roles", "can_issue", "valid_until", "revoked_at"}
if not isinstance(record, dict) or set(record) != fields or not active(record, now):
stop("FAIL", "identity_inactive_or_revoked")
if role is not None and role not in record["roles"]:
stop("FAIL", "approver_role_not_granted")
return record
def key_for(registry: dict[str, Any], key_id: str, issuer_id: str, artifact_type: str, now: datetime) -> Path:
record = registry["keys"].get(key_id)
fields = {"owner_id", "status", "file_name", "artifact_types", "valid_until", "revoked_at"}
if not isinstance(record, dict) or set(record) != fields or not active(record, now):
stop("FAIL", "signing_key_inactive_or_revoked")
if record["owner_id"] != issuer_id or artifact_type not in record["artifact_types"]:
stop("FAIL", "signing_key_not_authorized")
file_name = record["file_name"]
if not isinstance(file_name, str) or Path(file_name).name != file_name:
stop("FAIL", "registry_key_path_invalid")
path = (KEY_DIR / file_name).resolve()
if path.parent != KEY_DIR or not path.is_file():
stop("INSUFFICIENT_DATA", "trusted_key_unavailable")
return path
def load_envelope(path: Path, registry: dict[str, Any], now: datetime) -> dict[str, Any]:
envelope, raw, digest = load_canonical(path, ENVELOPE_FIELDS)
if envelope["schema_version"] != "aidefend.h018.signed_artifact.v2":
stop("FAIL", "artifact_schema_version_unsupported")
artifact_type = envelope["artifact_type"]
if artifact_type not in CLAIM_FIELDS:
stop("FAIL", "artifact_type_unsupported")
issuer = require_identity(registry, envelope["issuer_id"], now)
if artifact_type not in issuer["can_issue"]:
stop("FAIL", "issuer_not_authorized")
key = key_for(registry, envelope["key_id"], envelope["issuer_id"], artifact_type, now)
verify_blob_bytes(raw, signature_path(path), key)
issued_at, expires_at = parse_time(envelope["issued_at"]), parse_time(envelope["expires_at"])
if not issued_at <= now < expires_at:
stop("FAIL", "artifact_not_current")
if not isinstance(envelope["nonce"], str) or not TOKEN.fullmatch(envelope["nonce"]):
stop("FAIL", "artifact_nonce_invalid")
claims = envelope["claims"]
if not isinstance(claims, dict) or set(claims) != CLAIM_FIELDS[artifact_type]:
stop("FAIL", "typed_claim_schema_mismatch")
return {"document": envelope, "raw": raw, "sha256": digest, "expires_at": expires_at}
def logical_type(envelope: dict[str, Any]) -> str:
if envelope["artifact_type"] != "control_evidence":
return envelope["artifact_type"]
kind = envelope["claims"]["evidence_kind"]
if not isinstance(kind, str) or not TOKEN.fullmatch(kind):
stop("FAIL", "evidence_kind_invalid")
return "control_evidence:" + kind
def load_action(path: Path) -> tuple[dict[str, Any], bytes, str]:
action, raw, digest = load_canonical(path, ACTION_FIELDS)
if action["schema_version"] != "aidefend.h018.action_plan.v2":
stop("FAIL", "action_schema_version_unsupported")
for field in (
"action_id", "action_profile", "evidence_profile", "subject_id",
"workload_id", "session_id", "task_id", "tool", "target",
):
if not isinstance(action[field], str) or not TOKEN.fullmatch(action[field]):
stop("FAIL", "action_identifier_invalid")
if not isinstance(action["parameters"], dict):
stop("FAIL", "action_parameters_invalid")
state = action["state_binding"]
if not isinstance(state, dict) or set(state) != STATE_FIELDS:
stop("FAIL", "state_binding_schema_invalid")
for field in (
"provider_profile", "resource_id", "expected_pre_version", "expected_post_version",
):
if not isinstance(state[field], str) or not TOKEN.fullmatch(state[field]):
stop("FAIL", "state_binding_identifier_invalid")
for field in ("expected_pre_state_sha256", "expected_post_state_sha256"):
if not isinstance(state[field], str) or not SHA256.fullmatch(state[field]):
stop("FAIL", "state_binding_digest_invalid")
age = state["maximum_observation_age_seconds"]
if isinstance(age, bool) or not isinstance(age, int) or age < 1:
stop("FAIL", "observation_freshness_policy_invalid")
return action, raw, digest
def profile_for(policy: dict[str, Any], action: dict[str, Any]) -> dict[str, Any]:
profile = policy["profiles"].get(action["action_profile"])
fields = {
"applicability", "required_approval_count", "required_distinct_roles",
"required_assurance", "maximum_receipt_seconds",
"allowed_evidence_profiles", "evidence_requirements",
}
if not isinstance(profile, dict) or set(profile) != fields:
stop("FAIL", "action_profile_not_registered")
if profile["applicability"] == "not_applicable":
stop("NOT_APPLICABLE", "policy_marks_control_not_applicable")
if profile["applicability"] != "required":
stop("FAIL", "profile_applicability_invalid")
count = profile["required_approval_count"]
roles = profile["required_distinct_roles"]
ttl = profile["maximum_receipt_seconds"]
if (
isinstance(count, bool) or not isinstance(count, int) or count < 1
or not isinstance(roles, list) or len(roles) != count or len(set(roles)) != len(roles)
or not all(isinstance(x, str) and TOKEN.fullmatch(x) for x in roles)
or not isinstance(profile["required_assurance"], str) or not profile["required_assurance"]
or isinstance(ttl, bool) or not isinstance(ttl, int) or ttl < 1
or not isinstance(profile["allowed_evidence_profiles"], list)
or len(profile["allowed_evidence_profiles"]) != len(set(profile["allowed_evidence_profiles"]))
or not isinstance(profile["evidence_requirements"], dict)
or action["evidence_profile"] not in profile["allowed_evidence_profiles"]
):
stop("FAIL", "profile_approval_or_evidence_policy_invalid")
required = profile["evidence_requirements"].get(action["evidence_profile"])
if (
not isinstance(required, list) or len(required) != len(set(required))
or not all(isinstance(x, str) and x for x in required)
):
stop("FAIL", "evidence_profile_policy_invalid")
return profile
def verify_authorization_bundle(
request_dir: Path, action: dict[str, Any], action_sha: str,
policy: dict[str, Any], registry: dict[str, Any], gate: dict[str, Any], now: datetime,
) -> list[dict[str, Any]]:
artifact_dir = request_dir / "artifacts"
paths = sorted(artifact_dir.glob("*.json")) if artifact_dir.is_dir() else []
if not paths:
stop("INSUFFICIENT_DATA", "evidence_population_missing")
if len(paths) > 256 or any(path.is_symlink() or path.resolve().parent != artifact_dir.resolve() for path in paths):
stop("FAIL", "evidence_population_or_path_invalid")
artifacts = [load_envelope(path, registry, now) for path in paths]
digests = sorted(item["sha256"] for item in artifacts)
if digests != gate["evidence_sha256s"] or len(digests) != len(set(digests)):
stop("FAIL", "gate_evidence_population_mismatch")
if sha256(canonical(digests)) != gate["evidence_population_sha256"]:
stop("FAIL", "gate_evidence_population_digest_mismatch")
binding = (action["action_id"], action_sha, policy["policy_version"], action["evidence_profile"])
for item in artifacts:
claims = item["document"]["claims"]
if (
claims["action_id"], claims["action_sha256"],
claims["policy_version"], claims["evidence_profile"],
) != binding:
stop("FAIL", "authorization_artifact_binding_mismatch")
profile = profile_for(policy, action)
approvals = [x for x in artifacts if x["document"]["artifact_type"] == "approval"]
evidence = [x for x in artifacts if x["document"]["artifact_type"] != "approval"]
if sorted(logical_type(x["document"]) for x in evidence) != sorted(
profile["evidence_requirements"][action["evidence_profile"]]
):
stop("INSUFFICIENT_DATA", "evidence_population_not_exact")
if len(approvals) != profile["required_approval_count"]:
stop("INSUFFICIENT_DATA", "approval_count_not_exact")
identities = [x for x in evidence if x["document"]["artifact_type"] == "identity_context"]
validators = [x for x in evidence if x["document"]["artifact_type"] == "validator_decision"]
if len(identities) != 1 or len(validators) != 1:
stop("INSUFFICIENT_DATA", "identity_or_validator_evidence_not_exact")
identity = identities[0]["document"]["claims"]
if (
identity["subject_id"] != action["subject_id"]
or identity["workload_id"] != action["workload_id"]
or not isinstance(identity["delegation_chain"], list)
or len(identity["delegation_chain"]) != len(set(identity["delegation_chain"]))
or not all(isinstance(x, str) and TOKEN.fullmatch(x) for x in identity["delegation_chain"])
):
stop("FAIL", "identity_context_binding_mismatch")
require_identity(registry, action["subject_id"], now)
require_identity(registry, action["workload_id"], now)
for delegated_identity in identity["delegation_chain"]:
require_identity(registry, delegated_identity, now)
validator = validators[0]["document"]["claims"]
if (
validator["decision"] != "allow"
or not SHA256.fullmatch(validator["observed_population_sha256"])
or isinstance(validator["observed_population_count"], bool)
or not isinstance(validator["observed_population_count"], int)
or validator["observed_population_count"] < 1
):
stop("FAIL", "validator_evidence_invalid")
approver_ids: list[str] = []
roles: list[str] = []
nonce_hashes: list[str] = []
for item in approvals:
claims = item["document"]["claims"]
if claims["decision"] != "approved" or claims["assurance"] != profile["required_assurance"]:
stop("FAIL", "approval_decision_or_assurance_invalid")
require_identity(registry, claims["approver_id"], now, claims["role"])
approver_ids.append(claims["approver_id"])
roles.append(claims["role"])
nonce_hashes.append(sha256(item["document"]["nonce"].encode("utf-8")))
if (
len(set(approver_ids)) != len(approver_ids)
or len(set(roles)) != len(roles)
or len(set(nonce_hashes)) != len(nonce_hashes)
or sorted(roles) != sorted(profile["required_distinct_roles"])
or sorted(nonce_hashes) != gate["approval_nonce_hashes"]
):
stop("FAIL", "approval_identity_role_or_nonce_mismatch")
previews = [
x for x in evidence
if x["document"]["artifact_type"] == "control_evidence"
and x["document"]["claims"]["evidence_kind"] == EFFECT_PREVIEW_KIND
]
if gate["effect_preview_sha256"] is None:
# No preview was recorded: none may be present, and no approval may
# claim to be bound to one.
if previews or gate["effect_summary_sha256"] is not None or any(
x["document"]["claims"]["effect_preview_sha256"] is not None
or x["document"]["claims"]["summary_sha256"] is not None
for x in approvals
):
stop("FAIL", "effect_preview_binding_mismatch")
else:
if (
len(previews) != 1 or previews[0]["sha256"] != gate["effect_preview_sha256"]
or not isinstance(gate["effect_summary_sha256"], str)
or not SHA256.fullmatch(gate["effect_summary_sha256"])
):
stop("FAIL", "effect_preview_binding_mismatch")
subject = previews[0]["document"]["claims"]["subject"]
if (
not isinstance(subject, dict)
or subject.get("summary_sha256") != gate["effect_summary_sha256"]
or subject.get("state_generation_sha256") != sha256(canonical(action["state_binding"]))
):
stop("FAIL", "effect_preview_stale_or_summary_mismatch")
for item in approvals:
claims = item["document"]["claims"]
if (
claims["effect_preview_sha256"] != gate["effect_preview_sha256"]
or claims["summary_sha256"] != gate["effect_summary_sha256"]
):
stop("FAIL", "approval_not_bound_to_effect_preview")
for item in evidence:
if item["document"]["artifact_type"] == "control_evidence":
claims = item["document"]["claims"]
if (
claims["result"] != "PASS"
or not SHA256.fullmatch(claims["observed_population_sha256"])
or isinstance(claims["observed_population_count"], bool)
or not isinstance(claims["observed_population_count"], int)
or claims["observed_population_count"] < 0
):
stop("FAIL", "control_evidence_invalid")
return artifacts
def safe_tls_file(file_name: Any) -> Path:
if not isinstance(file_name, str) or Path(file_name).name != file_name:
stop("FAIL", "provider_tls_path_invalid")
path = (TLS_DIR / file_name).resolve()
if path.parent != TLS_DIR or not path.is_file():
stop("INSUFFICIENT_DATA", "provider_tls_material_missing")
return path
def https_url(value: Any) -> str:
if not isinstance(value, str):
stop("FAIL", "provider_url_invalid")
parsed = urlsplit(value)
if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment:
stop("FAIL", "provider_url_not_fixed_https")
return value
def provider_for(provider_registry: dict[str, Any], name: str) -> dict[str, Any]:
source = provider_registry["providers"].get(name)
fields = {
"applicability", "observer_url", "executor_url", "ca_file",
"client_cert_file", "client_key_file", "observer_issuer_id",
"executor_issuer_id", "observer_key_id", "executor_key_id",
"supports_atomic_precondition",
}
if not isinstance(source, dict) or set(source) != fields:
stop("FAIL", "provider_profile_not_registered")
provider = dict(source)
if provider["applicability"] == "not_applicable":
stop("NOT_APPLICABLE", "provider_marks_control_not_applicable")
if provider["applicability"] != "required" or provider["supports_atomic_precondition"] is not True:
stop("FAIL", "provider_cannot_enforce_atomic_precondition")
if provider["observer_issuer_id"] == provider["executor_issuer_id"]:
stop("FAIL", "observer_not_independent_from_executor")
provider["observer_url"] = https_url(provider["observer_url"])
provider["executor_url"] = https_url(provider["executor_url"])
for field in ("ca_file", "client_cert_file", "client_key_file"):
provider[field] = str(safe_tls_file(provider[field]))
return provider
def authenticated_client(provider: dict[str, Any]) -> httpx.Client:
context = ssl.create_default_context(cafile=provider["ca_file"])
context.minimum_version = ssl.TLSVersion.TLSv1_3
context.load_cert_chain(provider["client_cert_file"], provider["client_key_file"])
return httpx.Client(
verify=context, http2=True, follow_redirects=False,
timeout=httpx.Timeout(REMOTE_TOTAL_TIMEOUT_SECONDS, connect=REMOTE_CONNECT_TIMEOUT_SECONDS),
)
def send_streamed(client: httpx.Client, url: str, **kwargs) -> httpx.Response:
request = client.build_request("POST", url, **kwargs)
return client.send(request, stream=True)
def verify_remote_artifact(
response: httpx.Response, registry: dict[str, Any], expected_type: str,
expected_issuer: str, expected_key: str, now: datetime,
) -> dict[str, Any]:
raw = bytearray()
try:
if response.status_code in {409, 412}:
stop("FAIL", "provider_atomic_precondition_rejected")
try:
response.raise_for_status()
except httpx.HTTPStatusError:
stop("ERROR", "provider_response_error")
signature = response.headers.get("Aidefend-Signature", "")
if not SIGNATURE.fullmatch(signature):
stop("FAIL", "provider_signature_missing_or_invalid")
for chunk in response.iter_bytes():
raw.extend(chunk)
if len(raw) > REMOTE_RESPONSE_MAX_BYTES:
stop("ERROR", "provider_response_exceeds_signed_cap")
finally:
response.close()
with tempfile.TemporaryDirectory(prefix="h018-verify-") as directory:
artifact_path = Path(directory) / "artifact.json"
artifact_path.write_bytes(bytes(raw))
signature_path(artifact_path).write_text(signature + "\n", encoding="ascii")
artifact = load_envelope(artifact_path, registry, now)
document = artifact["document"]
if (
document["artifact_type"] != expected_type
or document["issuer_id"] != expected_issuer
or document["key_id"] != expected_key
):
stop("FAIL", "provider_artifact_identity_mismatch")
return artifact
def check_observation(
artifact: dict[str, Any], action: dict[str, Any], action_sha: str,
policy_version: str, phase: str, expected_version: str,
expected_state_sha: str, expected_effect_id: str | None,
expected_execution_sha: str | None, now: datetime,
) -> None:
claims = artifact["document"]["claims"]
state = action["state_binding"]
if (
claims["action_id"] != action["action_id"]
or claims["action_sha256"] != action_sha
or claims["policy_version"] != policy_version
or claims["evidence_profile"] != action["evidence_profile"]
):
stop("FAIL", "observation_action_binding_mismatch")
observed_at = parse_time(claims["observed_at"])
age = (now - observed_at).total_seconds()
if age < 0 or age > state["maximum_observation_age_seconds"]:
stop("FAIL", "observation_not_fresh")
if (
claims["phase"] != phase or claims["resource_id"] != state["resource_id"]
or claims["version"] != expected_version or claims["state_sha256"] != expected_state_sha
or claims["effect_id"] != expected_effect_id
or claims["execution_receipt_sha256"] != expected_execution_sha
or not isinstance(claims["observed_population_sha256"], str)
or not SHA256.fullmatch(claims["observed_population_sha256"])
or isinstance(claims["observed_population_count"], bool)
or not isinstance(claims["observed_population_count"], int)
or claims["observed_population_count"] < 1
):
stop("FAIL", "mutable_state_drift_or_population_invalid")
def consume_gate_once(gate: dict[str, Any], gate_sha: str, now: datetime) -> str:
expires_at = parse_time(gate["expires_at"])
ttl = int((expires_at - now).total_seconds())
if ttl < 1:
stop("FAIL", "gate_expired_before_dispatch")
nonce_hash = sha256(gate["gate_nonce"].encode("utf-8"))
client = redis.Redis(unix_socket_path=REDIS_SOCKET, decode_responses=True)
consumed = client.set("h018:gate-consumed:" + nonce_hash, gate_sha, nx=True, ex=ttl)
if consumed is not True:
stop("FAIL", "gate_nonce_replayed")
return nonce_hash
def execute(request_id: str, context: dict[str, Any]) -> None:
now = datetime.now(timezone.utc)
request_dir = INBOX / request_id
action, action_raw, action_sha = load_action(request_dir / "action-plan.json")
context.update(action_id=action["action_id"], action_sha256=action_sha)
policy, policy_sha = load_root_signed(
POLICY_PATH, {"schema_version", "policy_version", "valid_from", "expires_at", "profiles"},
)
registry, registry_sha = load_root_signed(
REGISTRY_PATH,
{"schema_version", "registry_version", "issued_at", "expires_at", "keys", "identities"},
)
provider_registry, provider_sha = load_root_signed(
PROVIDER_PATH,
{"schema_version", "registry_version", "issued_at", "expires_at", "providers"},
)
if (
policy["schema_version"] != "aidefend.h018.gate_policy.v2"
or registry["schema_version"] != "aidefend.h018.identity_key_registry.v2"
or provider_registry["schema_version"] != "aidefend.h018.provider_registry.v2"
):
stop("FAIL", "control_document_schema_unsupported")
for start, end in (
(policy["valid_from"], policy["expires_at"]),
(registry["issued_at"], registry["expires_at"]),
(provider_registry["issued_at"], provider_registry["expires_at"]),
):
if not parse_time(start) <= now < parse_time(end):
stop("FAIL", "control_document_not_current")
context.update(
policy_version=policy["policy_version"], policy_sha256=policy_sha,
registry_sha256=registry_sha, provider_registry_sha256=provider_sha,
)
authorization_dir = request_dir / "authorization"
gate_path = authorization_dir / "gate-decision.json"
replay_path = authorization_dir / "gate-replay.json"
gate_raw = verify_blob(gate_path, signature_path(gate_path), GATE_VERIFY_KEY)
gate, _, gate_sha = load_canonical_bytes(gate_raw, GATE_FIELDS)
replay_raw = verify_blob(
replay_path, signature_path(replay_path), GATE_REPLAY_VERIFY_KEY
)
replay, _, replay_sha = load_canonical_bytes(replay_raw, REPLAY_FIELDS)
context.update(gate_decision_sha256=gate_sha, gate_replay_sha256=replay_sha)
if (
gate["schema_version"] != "aidefend.h018.gate_decision.v3"
or gate["control"] != "AID-H-018.003" or gate["outcome"] != "PASS"
or gate["request_id"] != request_id or gate["action_id"] != action["action_id"]
or gate["action_sha256"] != action_sha or gate["policy_version"] != policy["policy_version"]
or gate["policy_sha256"] != policy_sha or gate["registry_sha256"] != registry_sha
or not isinstance(gate["gate_nonce"], str) or not TOKEN.fullmatch(gate["gate_nonce"])
or not parse_time(gate["issued_at"]) <= now < parse_time(gate["expires_at"])
):
stop("FAIL", "gate_decision_stale_or_binding_invalid")
if (
replay["schema_version"] != "aidefend.h018.gate_replay.v2"
or replay["control"] != "AID-H-018.003" or replay["replay_outcome"] != "PASS"
or replay["decision_sha256"] != gate_sha or replay["request_id"] != request_id
or replay["action_sha256"] != action_sha
or replay["artifact_population_sha256"] != gate["evidence_population_sha256"]
or replay["artifact_population_count"] != gate["evidence_population_count"]
or not parse_time(gate["issued_at"]) <= parse_time(replay["observed_at"]) <= now
):
stop("FAIL", "independent_gate_replay_invalid")
artifacts = verify_authorization_bundle(request_dir, action, action_sha, policy, registry, gate, now)
context["artifact_population_sha256"] = sha256(canonical(sorted(x["sha256"] for x in artifacts)))
context["artifact_population_count"] = len(artifacts)
provider = provider_for(provider_registry, action["state_binding"]["provider_profile"])
common_request = {
"action_id": action["action_id"], "action_sha256": action_sha,
"policy_version": policy["policy_version"],
"evidence_profile": action["evidence_profile"],
"resource_id": action["state_binding"]["resource_id"],
}
with authenticated_client(provider) as client:
response = send_streamed(client,
provider["observer_url"],
content=canonical(common_request | {
"phase": "pre", "effect_id": None, "execution_receipt_sha256": None,
}),
headers={"Content-Type": "application/aidefend-observation-request+json"},
)
pre_now = datetime.now(timezone.utc)
pre = verify_remote_artifact(
response, registry, "state_observation",
provider["observer_issuer_id"], provider["observer_key_id"], pre_now,
)
check_observation(
pre, action, action_sha, policy["policy_version"], "pre",
action["state_binding"]["expected_pre_version"],
action["state_binding"]["expected_pre_state_sha256"], None, None, pre_now,
)
context["pre_observation_sha256"] = pre["sha256"]
nonce_hash = consume_gate_once(gate, gate_sha, datetime.now(timezone.utc))
context["gate_nonce_sha256"] = nonce_hash
context["gate_consumed_at"] = datetime.now(timezone.utc).isoformat()
response = send_streamed(client,
provider["executor_url"], content=action_raw,
headers={
"Content-Type": "application/aidefend-action-plan+json",
"If-Match": '"' + action["state_binding"]["expected_pre_version"] + '"',
"Idempotency-Key": action["action_id"],
"Aidefend-Gate-Decision-SHA256": gate_sha,
"Aidefend-Pre-Observation-SHA256": pre["sha256"],
"Aidefend-Expected-State-SHA256": action["state_binding"]["expected_pre_state_sha256"],
},
)
execution = verify_remote_artifact(
response, registry, "execution_receipt",
provider["executor_issuer_id"], provider["executor_key_id"],
datetime.now(timezone.utc),
)
claims = execution["document"]["claims"]
if (
claims["action_id"] != action["action_id"] or claims["action_sha256"] != action_sha
or claims["policy_version"] != policy["policy_version"]
or claims["evidence_profile"] != action["evidence_profile"]
or claims["gate_decision_sha256"] != gate_sha
or claims["pre_observation_sha256"] != pre["sha256"]
or claims["precondition_version"] != action["state_binding"]["expected_pre_version"]
or claims["precondition_result"] != "matched"
or claims["effect_status"] not in {"applied", "no_change"}
or not isinstance(claims["effect_id"], str) or not TOKEN.fullmatch(claims["effect_id"])
or not isinstance(claims["provider_receipt_sha256"], str)
or not SHA256.fullmatch(claims["provider_receipt_sha256"])
):
stop("FAIL", "signed_execution_receipt_invalid")
context["execution_receipt_sha256"] = execution["sha256"]
response = send_streamed(client,
provider["observer_url"],
content=canonical(common_request | {
"phase": "post", "effect_id": claims["effect_id"],
"execution_receipt_sha256": execution["sha256"],
}),
headers={"Content-Type": "application/aidefend-observation-request+json"},
)
post_now = datetime.now(timezone.utc)
post = verify_remote_artifact(
response, registry, "state_observation",
provider["observer_issuer_id"], provider["observer_key_id"], post_now,
)
check_observation(
post, action, action_sha, policy["policy_version"], "post",
action["state_binding"]["expected_post_version"],
action["state_binding"]["expected_post_state_sha256"],
claims["effect_id"], execution["sha256"], post_now,
)
context["post_readback_sha256"] = post["sha256"]
def publish_signed(record: dict[str, Any]) -> tuple[Path, str]:
OUTBOX.mkdir(parents=True, exist_ok=True, mode=0o750)
name = record["request_id"] + "." + record["execution_decision_id"]
temporary = OUTBOX / ("." + name + ".tmp")
temporary_signature = signature_path(temporary)
final = OUTBOX / (name + ".json")
final_signature = signature_path(final)
payload = canonical(record)
with temporary.open("xb") as handle:
os.fchmod(handle.fileno(), 0o640)
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
subprocess.run(
[
COSIGN, "sign-blob", "--yes", "--key", str(EXECUTION_SIGNING_KEY),
"--bundle", str(temporary_signature), str(temporary),
],
check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS,
)
verify_blob(temporary, temporary_signature, EXECUTION_VERIFY_KEY)
os.replace(temporary, final)
os.replace(temporary_signature, final_signature)
if final.read_bytes() != payload:
raise OSError("execution decision readback mismatch")
verify_blob(final, final_signature, EXECUTION_VERIFY_KEY)
return final, sha256(payload)
def main() -> int:
if len(sys.argv) != 2 or not REQUEST_ID.fullmatch(sys.argv[1]):
raise SystemExit("usage: verify_and_dispatch.py REQUEST_ID")
request_id = sys.argv[1]
context: dict[str, Any] = {
"action_id": None, "action_sha256": None, "policy_version": None,
"policy_sha256": None, "registry_sha256": None,
"provider_registry_sha256": None, "gate_decision_sha256": None,
"gate_replay_sha256": None, "artifact_population_sha256": None,
"artifact_population_count": 0, "pre_observation_sha256": None,
"execution_receipt_sha256": None, "post_readback_sha256": None,
"gate_nonce_sha256": None, "gate_consumed_at": None,
}
outcome, reason = "PASS", "execution_and_readback_verified"
try:
execute(request_id, context)
except StopExecution as exc:
outcome, reason = exc.outcome, exc.reason_code
except (redis.RedisError, httpx.HTTPError, subprocess.TimeoutExpired, OSError, ssl.SSLError):
outcome, reason = "ERROR", "trusted_dependency_or_transport_error"
except Exception:
outcome, reason = "ERROR", "unhandled_execution_gate_error"
record = {
"schema_version": "aidefend.h018.execution_decision.v2",
"control": "AID-H-018.006",
"execution_decision_id": secrets.token_urlsafe(24),
"request_id": request_id,
"outcome": outcome,
"reason_code": reason,
"action_id": context["action_id"],
"action_sha256": context["action_sha256"],
"policy_version": context["policy_version"],
"policy_sha256": context["policy_sha256"],
"registry_sha256": context["registry_sha256"],
"provider_registry_sha256": context["provider_registry_sha256"],
"gate_decision_sha256": context["gate_decision_sha256"],
"gate_replay_sha256": context["gate_replay_sha256"],
"artifact_population_sha256": context["artifact_population_sha256"],
"artifact_population_count": context["artifact_population_count"],
"pre_observation_sha256": context["pre_observation_sha256"],
"execution_receipt_sha256": context["execution_receipt_sha256"],
"post_readback_sha256": context["post_readback_sha256"],
"gate_nonce_sha256": context["gate_nonce_sha256"],
"gate_consumed_at": context["gate_consumed_at"],
"issued_at": datetime.now(timezone.utc).isoformat(),
}
path, digest = publish_signed(record)
print(json.dumps({"outcome": outcome, "decision": str(path), "sha256": digest}))
return 0 if outcome in {"PASS", "NOT_APPLICABLE"} else 2
if __name__ == "__main__":
raise SystemExit(main())Execution and evidence contract
Computer-use boundary: AID-I-008.002 continues to own browser-specific suspension and exact control/DOM re-observation. It hands the unchanged browser action and H-018.003 receipt to this control; H-018.006 owns the generic execution-time authorization replay, atomic consumption of the one-time gate-decision nonce from the verified AID-H-018.003 receipt, and state-bound dispatch.
Action: Put this adapter in the only network path to each sensitive executor. Do not accept unsigned PDP booleans, caller-selected verification keys, shared-secret HMAC leases, unauthenticated HTTP, or placeholder state checks.
Keep the action safely bounded
Reconcile the complete population of every dispatch attempt, gate nonce, state precondition, provider execution, and required readback; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.
AID-H-018.006-G002
Expire authorization decisions quickly and require a fresh verification after pause, resume, retry, failover, worker handoff, or execution-context change.
Bind freshness to the execution transition
The signed authorization bundle contains issue, not-before, expiry, policy revision, token and delegation revisions, plan digest, action digest, task attempt, worker generation, and a one-time nonce. The dispatcher obtains current time from its trusted clock, rejects expired or not-yet-valid bundles, and requires reauthorization whenever a workflow is resumed, retried, migrated, restored from a checkpoint, or reassigned to another worker. Time windows come only from the signed organizational policy.
Test stale-decision denial
Exercise expiry boundaries, delayed queues, retry after partial failure, worker restart, region failover, changed delegation, revoked token, and modified plan cases. Require every stale decision to be denied before the provider call and prove from provider and audit readback that no side effect occurred. Preserve the old decision, fresh replacement, transition event, nonce result, dispatcher identity, and negative provider evidence.
AID-H-018.006-G003
Re-check session, delegation, plan, approval, target-resource, and policy invariants before each high-risk step and fail closed on any mismatch.
Compare immutable and mutable invariants
Immediately before dispatch, recompute the canonical action digest and compare the authenticated subject, tenant, session and task IDs, delegation-chain digest, effective scope, tool and operation, plan step, target resource identity and version, approval receipt, replay state, policy digest, and expected provider route with the signed authorization context. Re-read mutable target state with a conditional version or ETag; never accept caller-supplied current state.
Prove the exact effect
Dispatch only with an atomic provider precondition and one-time nonce. Store the intent before the call, authenticate the provider receipt, and have a separate read-only identity verify the exact post-action resource state. A mismatch, ambiguous provider result, missing readback, reused nonce, or changed invariant is a denial or failed outcome, never a best-effort continuation.
Open-source tools
- Cosign
- Valkey
- Redis 8+ (AGPL-3.0)
- HTTPX
- OPAL (Open Policy Administration Layer)
- Open Policy Agent (OPA)
- SpiceDB
Source-available / open-weight tools
- HashiCorp Vault Community (BUSL-1.1; source-available)
Commercial / hosted tools
- Permit.io
- AWS Key Management Service (KMS)
- HashiCorp Vault Enterprise
- HCP Vault Dedicated
Threat mappings
MITRE ATLAS
- AML.T0053 AI Agent Tool Invocation
MAESTRO
- Agent Tool Misuse (L7)
- Privilege Escalation (Cross-Layer)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection (fresh pre-dispatch authorization and bound action context limit injected requests from exploiting stale approvals)
- LLM03:2026 Excessive Agency
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI03:2026 Identity and Privilege Abuse
- ASI02:2026 Tool Misuse and Exploitation (TOCTOU prevention ensures authorization is current at execution time)
NIST Adversarial Machine Learning 2025
- NISTAML.039 Compromising connected resources (re-verification prevents stale access to resources)
Cisco Integrated AI Security and Safety Framework
- AISubtech-12.1.1 Parameter Manipulation
- AISubtech-14.1.2 Insufficient Access Controls
- AISubtech-14.2.1 Permission Escalation via Delegation
- AISubtech-2.1.1 Execution Approval Bypass
- AISubtech-2.1.2 Capability and Permission Overreach
- AISubtech-2.2.3 Authorization and Constraint Erosion
- AISubtech-4.3.4 Replay Exploitation
- AITech-12.1 Tool Exploitation
- AITech-14.2 Abuse of Delegated Authority
- AITech-2.1 Excessive Agency
- AITech-2.2 Goal Drift
Google Secure AI Framework 2.0 - Risks
- RA: Rogue Actions (continuous re-authorization catches rogue actions at execution time)
Databricks AI Security Framework 3.0
- Agents - Core 13.3: Privilege Compromise (continuous re-authorization prevents TOCTOU privilege escalation)
- Agents - Core 13.2: Tool Misuse (execution-time authorization prevents stale-credential tool misuse)
- Agents - Tools MCP Server 13.17: Confused Deputy (continuous re-authorization prevents confused deputy via stale delegation)