AID-H-031.004: Evidence-Bound Promotion & High-Risk Human Approval
Require the final merge, build, or deployment decision for an AI-generated development artifact to be bound to the complete evidence chain that justified it, and require higher-friction human approval when the signed route marks privileged, destructive, or production-critical paths. The final gate verifies
AID-H-031.001 provenance and routing, the signed AID-H-031.002 static-stage decision, the signed AID-H-031.003 dynamic-stage decision when routed, policy version, approver identity, and the exact change or artifact digest before issuing a one-time promotion authorization.Pillars: infra, app
Lifecycle: validation, operation
Scope boundary
Owns the final one-time merge, build, release, or deployment authorization for an AI-generated development artifact, bound to exact provenance, route, static and required dynamic evidence, policy, artifact digest, and any mandated human approval.
AID-H-003.002: AID-H-031.004 authorizes promotion of AI-generated source, infrastructure, workflow, deployment, build, or repository artifacts; AID-H-003.002 gates model artifacts and deployment-bound model packages. The promoted artifact class determines the owning gate.
AID-H-031.002: AID-H-031.004 alone authorizes promotion; AID-H-031.002 supplies the required static-stage result. Static analysis remains evidence even when it passes.
AID-H-031.003: AID-H-031.004 alone authorizes promotion; AID-H-031.003 supplies the dynamic-stage decision when policy routes the change to behavioral validation. Dynamic validation cannot self-promote the artifact it tested.
Implementation guidance
- AID-H-031.004-G001: Bind AI-generated promotion decisions to an immutable evidence record that captures artifact digests, policy version, and validator outputs.
- AID-H-031.004-G002: Require named or dual approval for high-risk AI-generated changes before promotion.
AID-H-031.004-G001
Bind AI-generated promotion decisions to an immutable evidence record that captures artifact digests, policy version, and validator outputs.
Boundary and signed promotion policy
This is the only stage in AID-H-031 that may issue a promotion authorization. It verifies independently signed provenance, routing, static-stage and subject evidence; requires the signed dynamic-stage decision and signed human-approval decision when the selected tier says they are mandatory; rehashes the exact artifact; and records every non-PASS result. The policy below is an example organizational profile, not a framework default.
{
"schema_version": "aidefend.h031.promotion-policy.v1",
"policy_version": "2026.07.21.1",
"routing_policy_version": "2026.07.17.1",
"valid_until": "2026-10-01T00:00:00Z",
"authorization_ttl_seconds": 300,
"tiers": {
"generated-standard": {
"require_dynamic_stage": false,
"approval_profile_id": "none",
"allowed_targets": ["staging"]
},
"generated-privileged": {
"require_dynamic_stage": true,
"approval_profile_id": "high-risk-dual",
"allowed_targets": ["staging", "production"]
}
}
}Verify the complete evidence chain and sign every decision
# File: ci/issue_h031_promotion.py
from __future__ import annotations
import hashlib
import json
import math
import os
import re
import secrets
import stat
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
import redis
CONTROL = Path("/etc/aidefend/h031")
INPUT = Path("/var/lib/aidefend/h031/input")
OUTPUT = Path("/var/lib/aidefend/h031/promotion-decisions")
ARTIFACT = INPUT / "promotion-subject.bin"
FILES = {
"policy": CONTROL / "promotion-policy.json",
"provenance": INPUT / "generated-change.json",
"routing": INPUT / "routing.json",
"static": INPUT / "static-stage.json",
"subject": INPUT / "dynamic-subject.json",
"dynamic": INPUT / "dynamic-stage.json",
"approval": INPUT / "approval-decision.json",
}
KEYS = {
"policy": CONTROL / "keys/promotion-policy.pub",
"provenance": CONTROL / "keys/provenance.pub",
"routing": CONTROL / "keys/routing.pub",
"static": CONTROL / "keys/static-stage.pub",
"subject": CONTROL / "keys/build-subject.pub",
"dynamic": CONTROL / "keys/dynamic-stage.pub",
"approval": CONTROL / "keys/approval-broker.pub",
}
SIGNING_KEY = Path("/run/secrets/aidefend/h031-promotion.key")
VERIFY_KEY = CONTROL / "keys/promotion.pub"
COSIGN = "/usr/local/bin/cosign"
REDIS_SOCKET = "/run/aidefend-h031/redis.sock"
SHA256 = re.compile(r"^[a-f0-9]{64}$")
TARGET = os.environ["AIDEFEND_PROMOTION_TARGET"]
COMMAND_TIMEOUT = float(os.environ["VERIFIED_H031_COMMAND_TIMEOUT_SECONDS"])
if not math.isfinite(COMMAND_TIMEOUT) or not 0 < COMMAND_TIMEOUT <= 300:
raise RuntimeError("verified command timeout is invalid")
FIELDS = {
"policy": {"schema_version", "policy_version", "routing_policy_version",
"valid_until", "authorization_ttl_seconds", "tiers"},
"provenance": {"schema_version", "generated_by", "generator_id", "model_id",
"generation_event_sha256", "source_repo", "base_commit", "head_commit",
"ticket_id", "file_manifest", "change_set_sha256",
"touched_risk_paths", "ci_workflow_identity", "ci_run_id"},
"routing": {"schema_version", "provenance_sha256", "policy_version", "selected_tier",
"required_checks", "required_review_roles", "base_commit", "head_commit",
"change_set_sha256"},
"static": {"schema_version", "outcome", "provenance_sha256", "routing_sha256",
"change_set_sha256", "check_report_sha256", "check_outcomes",
"derived_risk_paths"},
"subject": {"schema_version", "change_set_sha256", "routing_sha256",
"artifact_sha256", "artifact_media_type", "build_provenance_sha256",
"release_id", "issued_at"},
"dynamic": {"schema_version", "control", "outcome", "reason_code", "policy_sha256",
"routing_sha256", "static_stage_sha256", "subject_sha256",
"behavioral_evidence_sha256", "issued_at"},
"approval": {"schema_version", "control", "outcome", "reason_code", "request_id",
"requester_id", "ticket_id", "approval_profile_id", "change_set_sha256",
"artifact_sha256", "routing_sha256", "subject_sha256", "policy_sha256",
"identity_registry_sha256", "collection_sha256", "approver_ids",
"approval_ids", "roles", "issued_at", "expires_at"},
}
class GateStop(Exception):
def __init__(self, outcome: str, reason: str):
super().__init__(reason)
self.outcome = outcome
self.reason = reason
def canonical(value: Any) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False).encode("utf-8")
def sha256(raw: bytes) -> str:
return hashlib.sha256(raw).hexdigest()
def read_regular_bytes(path: Path) -> bytes:
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags)
with os.fdopen(descriptor, "rb") as handle:
before = os.fstat(handle.fileno())
if not stat.S_ISREG(before.st_mode) or before.st_size == 0:
raise GateStop("INSUFFICIENT_DATA", "required_input_not_regular")
raw = handle.read()
after = os.fstat(handle.fileno())
identity = lambda value: (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns)
if identity(before) != identity(after):
raise GateStop("INSUFFICIENT_DATA", "required_input_changed_during_read")
return raw
def strict_object(raw: bytes, label: str) -> dict[str, Any]:
def unique(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate {label} key: {key}")
result[key] = value
return result
value = json.loads(raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique,
parse_constant=lambda token: (_ for _ in ()).throw(
ValueError("non-finite JSON token: " + token)))
if not isinstance(value, dict):
raise ValueError(label + " must be an object")
return value
def verify_signed(name: str) -> tuple[dict[str, Any], str]:
path = FILES[name]
bundle = path.with_name(path.name + ".sigstore.json")
try:
raw = read_regular_bytes(path)
bundle_raw = read_regular_bytes(bundle)
except (FileNotFoundError, PermissionError, OSError) as error:
raise GateStop("INSUFFICIENT_DATA", name + "_signed_input_missing") from error
try:
with tempfile.TemporaryDirectory(prefix="h031-promotion-input-") as directory:
root = Path(directory)
payload_snapshot = root / "payload.json"
bundle_snapshot = root / "bundle.json"
payload_snapshot.write_bytes(raw)
bundle_snapshot.write_bytes(bundle_raw)
os.chmod(payload_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run([COSIGN, "verify-blob", "--key", str(KEYS[name]),
"--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True,
timeout=COMMAND_TIMEOUT)
if payload_snapshot.read_bytes() != raw or bundle_snapshot.read_bytes() != bundle_raw:
raise GateStop("ERROR", name + "_snapshot_changed")
except subprocess.CalledProcessError as error:
raise GateStop("FAIL", name + "_signature_invalid") from error
document = strict_object(raw, name)
if set(document) != FIELDS[name] or raw not in {canonical(document), canonical(document) + b"\n"}:
raise GateStop("FAIL", name + "_schema_or_canonical_form_invalid")
return document, sha256(canonical(document))
def parse_time(value: Any, reason: str) -> datetime:
if not isinstance(value, str):
raise GateStop("FAIL", reason)
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as error:
raise GateStop("FAIL", reason) from error
if parsed.utcoffset() is None:
raise GateStop("FAIL", reason)
return parsed.astimezone(timezone.utc)
def require_sha256(value: Any, reason: str) -> str:
if not isinstance(value, str) or SHA256.fullmatch(value) is None:
raise GateStop("FAIL", reason)
return value
def artifact_sha256(path: Path) -> str:
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags)
digest = hashlib.sha256()
with os.fdopen(descriptor, "rb") as handle:
before = os.fstat(handle.fileno())
if not stat.S_ISREG(before.st_mode) or before.st_size == 0:
raise GateStop("INSUFFICIENT_DATA", "artifact_not_regular")
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
after = os.fstat(handle.fileno())
identity = lambda value: (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns)
if identity(before) != identity(after):
raise GateStop("INSUFFICIENT_DATA", "artifact_changed_during_hash")
return digest.hexdigest()
def optional_input_present(path: Path) -> bool:
return path.exists() or path.is_symlink()
def evaluate(context: dict[str, Any]) -> tuple[int, datetime]:
policy, policy_sha = verify_signed("policy")
provenance, provenance_sha = verify_signed("provenance")
routing, routing_sha = verify_signed("routing")
static, static_sha = verify_signed("static")
subject, subject_sha = verify_signed("subject")
context.update(policy_sha256=policy_sha, provenance_sha256=provenance_sha,
routing_sha256=routing_sha, static_stage_sha256=static_sha,
subject_sha256=subject_sha)
now = datetime.now(timezone.utc)
if policy["schema_version"] != "aidefend.h031.promotion-policy.v1":
raise GateStop("FAIL", "promotion_policy_version_unsupported")
if parse_time(policy["valid_until"], "promotion_policy_expiry_invalid") <= now:
raise GateStop("FAIL", "promotion_policy_expired")
ttl = policy["authorization_ttl_seconds"]
if isinstance(ttl, bool) or not isinstance(ttl, int) or not 1 <= ttl <= 3600:
raise GateStop("FAIL", "authorization_ttl_invalid")
if provenance["schema_version"] != "aidefend.generated-change-provenance.v2":
raise GateStop("FAIL", "provenance_version_unsupported")
manifest = provenance["file_manifest"]
if not isinstance(manifest, list) or not manifest:
raise GateStop("INSUFFICIENT_DATA", "provenance_population_empty")
change_sha = require_sha256(provenance["change_set_sha256"], "change_digest_invalid")
if sha256(canonical(manifest)) != change_sha:
raise GateStop("FAIL", "provenance_manifest_digest_mismatch")
if (routing["schema_version"] != "aidefend.generated-change-routing.v2"
or routing["policy_version"] != policy["routing_policy_version"]):
raise GateStop("FAIL", "routing_policy_binding_mismatch")
if (routing["provenance_sha256"] != provenance_sha
or routing["change_set_sha256"] != change_sha
or routing["base_commit"] != provenance["base_commit"]
or routing["head_commit"] != provenance["head_commit"]):
raise GateStop("FAIL", "routing_provenance_binding_mismatch")
if static["schema_version"] != "aidefend.generated-static-admission.v2":
raise GateStop("FAIL", "static_stage_version_unsupported")
if (static["outcome"] != "pass"
or static["provenance_sha256"] != provenance_sha
or static["routing_sha256"] != routing_sha
or static["change_set_sha256"] != change_sha):
raise GateStop("FAIL", "static_stage_did_not_pass_exact_binding")
if subject["schema_version"] != "aidefend.h031.dynamic-subject.v1":
raise GateStop("FAIL", "subject_version_unsupported")
observed_artifact_sha = artifact_sha256(ARTIFACT)
if (subject["routing_sha256"] != routing_sha
or subject["change_set_sha256"] != change_sha
or require_sha256(subject["artifact_sha256"], "subject_artifact_digest_invalid")
!= observed_artifact_sha):
raise GateStop("FAIL", "subject_artifact_binding_mismatch")
tiers = policy["tiers"]
tier = tiers.get(routing["selected_tier"]) if isinstance(tiers, dict) else None
if not isinstance(tier, dict) or set(tier) != {
"require_dynamic_stage", "approval_profile_id", "allowed_targets"
}:
raise GateStop("FAIL", "selected_promotion_tier_missing")
if tier["require_dynamic_stage"] not in {True, False}:
raise GateStop("FAIL", "dynamic_stage_requirement_invalid")
if (not isinstance(tier["approval_profile_id"], str)
or not tier["approval_profile_id"]):
raise GateStop("FAIL", "approval_profile_invalid")
allowed_targets = tier["allowed_targets"]
if (not isinstance(allowed_targets, list) or not allowed_targets
or len(allowed_targets) != len(set(allowed_targets))
or TARGET not in allowed_targets):
raise GateStop("FAIL", "promotion_target_not_allowed")
if tier["require_dynamic_stage"]:
dynamic, dynamic_sha = verify_signed("dynamic")
if (dynamic["schema_version"] != "aidefend.h031.dynamic-stage-decision.v1"
or dynamic["control"] != "AID-H-031.003"
or dynamic["outcome"] != "PASS"
or dynamic["routing_sha256"] != routing_sha
or dynamic["static_stage_sha256"] != static_sha
or dynamic["subject_sha256"] != subject_sha):
raise GateStop("FAIL", "required_dynamic_stage_did_not_pass")
context["dynamic_stage_sha256"] = dynamic_sha
elif optional_input_present(FILES["dynamic"]):
raise GateStop("FAIL", "unexpected_dynamic_stage_input")
profile = tier["approval_profile_id"]
if profile != "none":
approval, approval_sha = verify_signed("approval")
if (approval["schema_version"] != "aidefend.h031.approval-decision.v2"
or approval["control"] != "AID-H-031.004"
or approval["outcome"] != "PASS"
or approval["approval_profile_id"] != profile
or approval["ticket_id"] != provenance["ticket_id"]
or approval["change_set_sha256"] != change_sha
or approval["artifact_sha256"] != observed_artifact_sha
or approval["routing_sha256"] != routing_sha
or approval["subject_sha256"] != subject_sha
or not isinstance(approval["approver_ids"], list)
or not approval["approver_ids"]
or len(approval["approver_ids"]) != len(set(approval["approver_ids"]))
or not isinstance(approval["approval_ids"], list)
or len(approval["approval_ids"]) != len(approval["approver_ids"])
or not isinstance(approval["roles"], list)
or not approval["roles"]
or approval["requester_id"] in approval["approver_ids"]
or parse_time(approval["expires_at"], "approval_expiry_invalid") <= now):
raise GateStop("FAIL", "required_human_approval_did_not_pass")
context["approval_decision_sha256"] = approval_sha
elif optional_input_present(FILES["approval"]):
raise GateStop("FAIL", "unexpected_approval_input")
context.update(artifact_sha256=observed_artifact_sha,
change_set_sha256=change_sha,
release_id=subject["release_id"],
selected_tier=routing["selected_tier"],
promotion_target=TARGET)
return ttl, now + timedelta(seconds=ttl)
def verify_output(path: Path) -> None:
subprocess.run([COSIGN, "verify-blob", "--key", str(VERIFY_KEY),
"--bundle", str(path) + ".sigstore.json", str(path)],
check=True, capture_output=True, text=True,
timeout=COMMAND_TIMEOUT)
def publish(record: dict[str, Any]) -> tuple[Path, str]:
OUTPUT.mkdir(parents=True, exist_ok=True, mode=0o750)
stem = "promotion-" + secrets.token_urlsafe(18)
temporary = OUTPUT / ("." + stem + ".json.tmp")
final = OUTPUT / (stem + ".json")
temporary_bundle = Path(str(temporary) + ".sigstore.json")
final_bundle = Path(str(final) + ".sigstore.json")
payload = canonical(record) + b"\n"
with temporary.open("xb") as handle:
os.fchmod(handle.fileno(), 0o440)
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
subprocess.run([COSIGN, "sign-blob", "--yes", "--key", str(SIGNING_KEY),
"--bundle", str(temporary_bundle), str(temporary)],
check=True, capture_output=True, text=True,
timeout=COMMAND_TIMEOUT)
os.replace(temporary, final)
os.replace(temporary_bundle, final_bundle)
if final.read_bytes() != payload:
raise OSError("promotion decision readback mismatch")
verify_output(final)
return final, sha256(payload)
def main() -> int:
context = {name: None for name in (
"policy_sha256", "provenance_sha256", "routing_sha256",
"static_stage_sha256", "subject_sha256", "dynamic_stage_sha256",
"approval_decision_sha256", "artifact_sha256", "change_set_sha256",
"release_id", "selected_tier", "promotion_target")}
outcome, reason = "PASS", "complete_evidence_chain_verified"
ttl, expires_at = 0, datetime.now(timezone.utc)
try:
ttl, expires_at = evaluate(context)
except GateStop as error:
outcome, reason = error.outcome, error.reason
except (OSError, subprocess.SubprocessError, redis.RedisError):
outcome, reason = "ERROR", "trusted_gate_dependency_error"
except Exception:
outcome, reason = "ERROR", "unexpected_promotion_gate_error"
issued_at = datetime.now(timezone.utc)
record = {
"schema_version": "aidefend.h031.promotion-authorization.v2",
"control": "AID-H-031.004",
"authorization_id": secrets.token_urlsafe(24),
"outcome": outcome,
"reason_code": reason,
**context,
"authorization_nonce": secrets.token_urlsafe(32) if outcome == "PASS" else None,
"issued_at": issued_at.isoformat(),
"expires_at": expires_at.isoformat() if outcome == "PASS" else issued_at.isoformat(),
}
payload_sha = sha256(canonical(record) + b"\n")
if outcome == "PASS":
store = redis.Redis(unix_socket_path=REDIS_SOCKET, decode_responses=True)
reserved = store.set("aidefend:h031:authorization:" + record["authorization_id"],
payload_sha, nx=True, ex=ttl)
if not reserved:
raise RuntimeError("one-time authorization reservation failed")
path, digest = publish(record)
print(json.dumps({"outcome": outcome, "decision": str(path), "sha256": digest}))
return 0 if outcome == "PASS" else 2
if __name__ == "__main__":
raise SystemExit(main())
Consume once and deploy the exact open artifact
# File: deploy/consume_h031_authorization.py
from __future__ import annotations
import hashlib
import json
import os
import stat
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import BinaryIO, Callable
import redis
COSIGN = "/usr/local/bin/cosign"
VERIFY_KEY = Path("/etc/aidefend/h031/keys/promotion.pub")
COMMAND_TIMEOUT = float(os.environ["VERIFIED_H031_COMMAND_TIMEOUT_SECONDS"])
REDIS_SOCKET = "/run/aidefend-h031/redis.sock"
AUTH_FIELDS = {
"schema_version", "control", "authorization_id", "outcome", "reason_code",
"policy_sha256", "provenance_sha256", "routing_sha256", "static_stage_sha256",
"subject_sha256", "dynamic_stage_sha256", "approval_decision_sha256",
"artifact_sha256", "change_set_sha256", "release_id", "selected_tier",
"promotion_target", "authorization_nonce", "issued_at", "expires_at",
}
def canonical(value: dict) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False).encode("utf-8")
def verified_authorization(path: Path, bundle: Path) -> tuple[dict, str]:
raw = path.read_bytes()
bundle_raw = bundle.read_bytes()
with tempfile.TemporaryDirectory(prefix="h031-deploy-auth-") as directory:
root = Path(directory)
payload_snapshot = root / "authorization.json"
bundle_snapshot = root / "bundle.json"
payload_snapshot.write_bytes(raw)
bundle_snapshot.write_bytes(bundle_raw)
subprocess.run([COSIGN, "verify-blob", "--key", str(VERIFY_KEY),
"--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True,
timeout=COMMAND_TIMEOUT)
value = json.loads(raw.decode("utf-8", errors="strict"))
if (not isinstance(value, dict) or set(value) != AUTH_FIELDS
or raw not in {canonical(value), canonical(value) + b"\n"}):
raise PermissionError("promotion authorization schema or canonical form differs")
return value, hashlib.sha256(canonical(value) + b"\n").hexdigest()
def consume_and_deploy(
authorization_path: Path,
authorization_bundle: Path,
artifact_path: Path,
expected_target: str,
deploy_from_open_artifact: Callable[[BinaryIO, dict], None],
) -> None:
authorization, authorization_sha = verified_authorization(
authorization_path, authorization_bundle
)
expires = datetime.fromisoformat(
authorization["expires_at"].replace("Z", "+00:00")
).astimezone(timezone.utc)
if (authorization["schema_version"] != "aidefend.h031.promotion-authorization.v2"
or authorization["control"] != "AID-H-031.004"
or authorization["outcome"] != "PASS"
or authorization["promotion_target"] != expected_target
or expires <= datetime.now(timezone.utc)):
raise PermissionError("promotion authorization is not usable at this sink")
descriptor = os.open(artifact_path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
with os.fdopen(descriptor, "rb") as artifact:
before = os.fstat(artifact.fileno())
if not stat.S_ISREG(before.st_mode) or before.st_size == 0:
raise PermissionError("promotion artifact is not one non-empty regular file")
digest = hashlib.sha256()
for block in iter(lambda: artifact.read(1024 * 1024), b""):
digest.update(block)
after = os.fstat(artifact.fileno())
identity = lambda value: (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns)
if (identity(before) != identity(after)
or digest.hexdigest() != authorization["artifact_sha256"]):
raise PermissionError("artifact changed or differs from authorization")
artifact.seek(0)
store = redis.Redis(unix_socket_path=REDIS_SOCKET, decode_responses=True)
consumed = store.eval(
"local v=redis.call('GET',KEYS[1]); "
"if (not v) or v~=ARGV[1] then return 0 end; "
"redis.call('DEL',KEYS[1]); return 1",
1,
"aidefend:h031:authorization:" + authorization["authorization_id"],
authorization_sha,
)
if consumed != 1:
raise PermissionError("promotion authorization is missing, changed, expired, or replayed")
deploy_from_open_artifact(artifact, authorization)
Action: The final signer and Redis ledger are unavailable to the AI agent, build runner, scanners, sandbox and approval broker. Retain PASS, FAIL, INSUFFICIENT_DATA and ERROR decisions in append-only storage. A deployment consumes exactly one signed authorization, verifies the target and expiry, rehashes the artifact through the same open file descriptor used by the deployment adapter, and fails closed on missing dynamic or approval evidence.
Independent verification
A separately credentialed verifier replays forged signature, omitted dynamic stage, omitted approval, stale policy, altered artifact, mixed routing/static evidence, duplicate authorization, wrong target, ledger outage and deployment-failure fixtures. Only the complete positive chain may be consumed once; every other fixture must emit a non-PASS decision or be rejected by the deployment sink.
AID-H-031.004-G002
Require named or dual approval for high-risk AI-generated changes before promotion.
Signed approval policy and closed request
High-risk approval is a separately signed input to the final promotion gate, not a UI badge or a caller-provided list. The approval broker verifies a signed request from the source-control boundary, a signed identity-and-role registry, a signed closed collection manifest, and every individual approver assertion. The values below are an example organizational policy.
{
"schema_version": "aidefend.h031.approval-policy.v1",
"policy_version": "2026.07.21.1",
"valid_until": "2026-10-01T00:00:00Z",
"max_clock_skew_seconds": 60,
"profiles": {
"high-risk-dual": {
"required_roles": ["platform_owner", "security_owner"],
"minimum_distinct_approvers": 2,
"maximum_lifetime_seconds": 900
}
}
}Verify distinct identities, roles, bindings and the complete assertion population
# File: approval/issue_h031_approval_decision.py
from __future__ import annotations
import hashlib
import json
import math
import os
import re
import secrets
import stat
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
CONTROL = Path("/etc/aidefend/h031")
INPUT = Path("/var/lib/aidefend/h031/approval-input")
ASSERTIONS = INPUT / "assertions"
OUTPUT = Path("/var/lib/aidefend/h031/approval-decisions")
FILES = {
"policy": CONTROL / "approval-policy.json",
"registry": CONTROL / "approval-identity-registry.json",
"routing": INPUT / "routing.json",
"subject": INPUT / "dynamic-subject.json",
"request": INPUT / "approval-request.json",
"collection": INPUT / "approval-collection.json",
}
KEYS = {
"policy": CONTROL / "keys/approval-policy.pub",
"registry": CONTROL / "keys/approval-identity-registry.pub",
"routing": CONTROL / "keys/routing.pub",
"subject": CONTROL / "keys/build-subject.pub",
"request": CONTROL / "keys/source-control-approval-request.pub",
"collection": CONTROL / "keys/approval-collection.pub",
}
APPROVER_KEY_DIR = (CONTROL / "approver-keys").resolve()
SIGNING_KEY = Path("/run/secrets/aidefend/h031-approval-broker.key")
VERIFY_KEY = CONTROL / "keys/approval-broker.pub"
COSIGN = "/usr/local/bin/cosign"
SHA256 = re.compile(r"^[a-f0-9]{64}$")
SAFE_FILE = re.compile(r"^approval-[A-Za-z0-9._-]{1,96}\.json$")
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,255}$")
COMMAND_TIMEOUT = float(os.environ["VERIFIED_H031_COMMAND_TIMEOUT_SECONDS"])
if not math.isfinite(COMMAND_TIMEOUT) or not 0 < COMMAND_TIMEOUT <= 300:
raise RuntimeError("verified command timeout is invalid")
FIELDS = {
"policy": {"schema_version", "policy_version", "valid_until",
"max_clock_skew_seconds", "profiles"},
"registry": {"schema_version", "registry_version", "identities"},
"routing": {"schema_version", "provenance_sha256", "policy_version", "selected_tier",
"required_checks", "required_review_roles", "base_commit", "head_commit",
"change_set_sha256"},
"subject": {"schema_version", "change_set_sha256", "routing_sha256",
"artifact_sha256", "artifact_media_type", "build_provenance_sha256",
"release_id", "issued_at"},
"request": {"schema_version", "request_id", "requester_id", "ticket_id",
"approval_profile_id", "change_set_sha256", "artifact_sha256",
"routing_sha256", "subject_sha256", "issued_at", "expires_at"},
"collection": {"schema_version", "request_id", "assertions", "closed_at"},
}
ASSERTION_FIELDS = {"schema_version", "approval_id", "request_id", "approver_id",
"role", "decision", "ticket_id", "change_set_sha256",
"artifact_sha256", "routing_sha256", "subject_sha256",
"approved_at", "expires_at", "nonce"}
class ApprovalStop(Exception):
def __init__(self, outcome: str, reason: str):
super().__init__(reason)
self.outcome = outcome
self.reason = reason
def canonical(value: Any) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False).encode("utf-8")
def sha256(raw: bytes) -> str:
return hashlib.sha256(raw).hexdigest()
def read_regular_bytes(path: Path) -> bytes:
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags)
with os.fdopen(descriptor, "rb") as handle:
before = os.fstat(handle.fileno())
if not stat.S_ISREG(before.st_mode) or before.st_size == 0:
raise ApprovalStop("INSUFFICIENT_DATA", "signed_input_not_regular")
raw = handle.read()
after = os.fstat(handle.fileno())
identity = lambda value: (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns)
if identity(before) != identity(after):
raise ApprovalStop("INSUFFICIENT_DATA", "signed_input_changed_during_read")
return raw
def strict_object(raw: bytes, label: str) -> dict[str, Any]:
def unique(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate {label} key: {key}")
result[key] = value
return result
value = json.loads(raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique,
parse_constant=lambda token: (_ for _ in ()).throw(
ValueError("non-finite JSON token: " + token)))
if not isinstance(value, dict):
raise ValueError(label + " must be an object")
return value
def verify_raw(raw: bytes, bundle_raw: bytes, key: Path, label: str) -> None:
try:
with tempfile.TemporaryDirectory(prefix="h031-approval-input-") as directory:
root = Path(directory)
payload = root / "payload.json"
bundle = root / "bundle.json"
payload.write_bytes(raw)
bundle.write_bytes(bundle_raw)
os.chmod(payload, 0o400)
os.chmod(bundle, 0o400)
subprocess.run([COSIGN, "verify-blob", "--key", str(key),
"--bundle", str(bundle), str(payload)],
check=True, capture_output=True, text=True,
timeout=COMMAND_TIMEOUT)
if payload.read_bytes() != raw or bundle.read_bytes() != bundle_raw:
raise ApprovalStop("ERROR", label + "_snapshot_changed")
except subprocess.CalledProcessError as error:
raise ApprovalStop("FAIL", label + "_signature_invalid") from error
def load_signed(name: str) -> tuple[dict[str, Any], str]:
path = FILES[name]
bundle = path.with_name(path.name + ".sigstore.json")
try:
raw = read_regular_bytes(path)
bundle_raw = read_regular_bytes(bundle)
except (FileNotFoundError, PermissionError, OSError) as error:
raise ApprovalStop("INSUFFICIENT_DATA", name + "_signed_input_missing") from error
verify_raw(raw, bundle_raw, KEYS[name], name)
value = strict_object(raw, name)
if set(value) != FIELDS[name] or raw not in {canonical(value), canonical(value) + b"\n"}:
raise ApprovalStop("FAIL", name + "_schema_or_canonical_form_invalid")
return value, sha256(canonical(value))
def parse_time(value: Any, reason: str) -> datetime:
if not isinstance(value, str):
raise ApprovalStop("FAIL", reason)
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as error:
raise ApprovalStop("FAIL", reason) from error
if parsed.utcoffset() is None:
raise ApprovalStop("FAIL", reason)
return parsed.astimezone(timezone.utc)
def require_sha256(value: Any, reason: str) -> str:
if not isinstance(value, str) or SHA256.fullmatch(value) is None:
raise ApprovalStop("FAIL", reason)
return value
def validate_identity_registry(registry: dict[str, Any]) -> dict[str, dict[str, Any]]:
if registry["schema_version"] != "aidefend.h031.approval-identities.v1":
raise ApprovalStop("FAIL", "identity_registry_version_unsupported")
identities = registry["identities"]
if not isinstance(identities, dict) or not identities:
raise ApprovalStop("INSUFFICIENT_DATA", "identity_registry_empty")
for approver_id, identity in identities.items():
if (not isinstance(approver_id, str) or SAFE_ID.fullmatch(approver_id) is None
or not isinstance(identity, dict)
or set(identity) != {"key_id", "roles", "active"}
or not isinstance(identity["key_id"], str)
or re.fullmatch(r"[A-Za-z0-9._-]{1,96}", identity["key_id"]) is None
or not isinstance(identity["roles"], list) or not identity["roles"]
or len(identity["roles"]) != len(set(identity["roles"]))
or any(not isinstance(role, str) or not role for role in identity["roles"])
or identity["active"] not in {True, False}):
raise ApprovalStop("FAIL", "identity_registry_entry_invalid")
return identities
def load_assertion(name: str, expected_sha: str,
identities: dict[str, dict[str, Any]]) -> tuple[dict[str, Any], str]:
if SAFE_FILE.fullmatch(name) is None or Path(name).name != name:
raise ApprovalStop("FAIL", "assertion_filename_invalid")
path = ASSERTIONS / name
bundle = path.with_name(path.name + ".sigstore.json")
raw = read_regular_bytes(path)
if require_sha256(expected_sha, "assertion_manifest_digest_invalid") != sha256(raw):
raise ApprovalStop("FAIL", "assertion_collection_digest_mismatch")
value = strict_object(raw, name)
if set(value) != ASSERTION_FIELDS or raw not in {canonical(value), canonical(value) + b"\n"}:
raise ApprovalStop("FAIL", "assertion_schema_or_canonical_form_invalid")
approver_id = value["approver_id"]
identity = identities.get(approver_id)
if not isinstance(identity, dict) or identity["active"] is not True:
raise ApprovalStop("FAIL", "approver_identity_unknown_or_inactive")
key = (APPROVER_KEY_DIR / (identity["key_id"] + ".pub")).resolve()
if key.parent != APPROVER_KEY_DIR:
raise ApprovalStop("FAIL", "approver_key_path_escape")
verify_raw(raw, read_regular_bytes(bundle), key, "approver_assertion")
return value, sha256(canonical(value))
def evaluate(context: dict[str, Any]) -> datetime:
policy, policy_sha = load_signed("policy")
registry, registry_sha = load_signed("registry")
routing, routing_sha = load_signed("routing")
subject, subject_sha = load_signed("subject")
request, request_sha = load_signed("request")
collection, collection_sha = load_signed("collection")
context.update(policy_sha256=policy_sha,
identity_registry_sha256=registry_sha,
collection_sha256=collection_sha)
now = datetime.now(timezone.utc)
if policy["schema_version"] != "aidefend.h031.approval-policy.v1":
raise ApprovalStop("FAIL", "approval_policy_version_unsupported")
if parse_time(policy["valid_until"], "approval_policy_expiry_invalid") <= now:
raise ApprovalStop("FAIL", "approval_policy_expired")
skew = policy["max_clock_skew_seconds"]
if isinstance(skew, bool) or not isinstance(skew, int) or not 0 <= skew <= 300:
raise ApprovalStop("FAIL", "approval_clock_skew_invalid")
if routing["schema_version"] != "aidefend.generated-change-routing.v2":
raise ApprovalStop("FAIL", "routing_version_unsupported")
if subject["schema_version"] != "aidefend.h031.dynamic-subject.v1":
raise ApprovalStop("FAIL", "subject_version_unsupported")
if request["schema_version"] != "aidefend.h031.approval-request.v1":
raise ApprovalStop("FAIL", "approval_request_version_unsupported")
if (request["routing_sha256"] != routing_sha
or request["subject_sha256"] != subject_sha
or request["change_set_sha256"] != routing["change_set_sha256"]
or request["change_set_sha256"] != subject["change_set_sha256"]
or request["artifact_sha256"] != subject["artifact_sha256"]):
raise ApprovalStop("FAIL", "approval_request_binding_mismatch")
issued = parse_time(request["issued_at"], "approval_request_time_invalid")
request_expires = parse_time(request["expires_at"], "approval_request_time_invalid")
if issued > now.replace(microsecond=0) + timedelta(seconds=skew):
raise ApprovalStop("FAIL", "approval_request_future_issued")
if request_expires <= now or request_expires <= issued:
raise ApprovalStop("FAIL", "approval_request_expired")
profiles = policy["profiles"]
profile = profiles.get(request["approval_profile_id"]) if isinstance(profiles, dict) else None
if not isinstance(profile, dict) or set(profile) != {
"required_roles", "minimum_distinct_approvers", "maximum_lifetime_seconds"
}:
raise ApprovalStop("FAIL", "approval_profile_missing")
required_roles = profile["required_roles"]
minimum = profile["minimum_distinct_approvers"]
maximum_lifetime = profile["maximum_lifetime_seconds"]
if (not isinstance(required_roles, list) or not required_roles
or len(required_roles) != len(set(required_roles))
or any(not isinstance(role, str) or not role for role in required_roles)
or isinstance(minimum, bool) or not isinstance(minimum, int)
or minimum < len(required_roles)
or isinstance(maximum_lifetime, bool) or not isinstance(maximum_lifetime, int)
or maximum_lifetime < 1):
raise ApprovalStop("FAIL", "approval_profile_invalid")
if set(routing["required_review_roles"]) != set(required_roles):
raise ApprovalStop("FAIL", "routing_review_roles_do_not_match_profile")
identities = validate_identity_registry(registry)
if (collection["schema_version"] != "aidefend.h031.approval-collection.v1"
or collection["request_id"] != request["request_id"]):
raise ApprovalStop("FAIL", "approval_collection_binding_mismatch")
closed_at = parse_time(collection["closed_at"], "approval_collection_time_invalid")
if closed_at > now.replace(microsecond=0) + timedelta(seconds=skew):
raise ApprovalStop("FAIL", "approval_collection_future_closed")
manifest = collection["assertions"]
if not isinstance(manifest, dict) or not manifest:
raise ApprovalStop("INSUFFICIENT_DATA", "approval_assertion_population_empty")
actual_names = {
path.name for path in ASSERTIONS.glob("approval-*.json")
if not path.name.endswith(".sigstore.json")
}
if actual_names != set(manifest):
raise ApprovalStop("INSUFFICIENT_DATA", "approval_assertion_population_incomplete_or_extra")
approver_ids: set[str] = set()
approval_ids: set[str] = set()
nonces: set[str] = set()
observed_roles: set[str] = set()
expires: list[datetime] = []
assertion_hashes: list[str] = []
for name in sorted(manifest):
assertion, assertion_sha = load_assertion(name, manifest[name], identities)
assertion_hashes.append(assertion_sha)
if (assertion["schema_version"] != "aidefend.h031.approver-assertion.v1"
or assertion["request_id"] != request["request_id"]
or assertion["ticket_id"] != request["ticket_id"]
or assertion["change_set_sha256"] != request["change_set_sha256"]
or assertion["artifact_sha256"] != request["artifact_sha256"]
or assertion["routing_sha256"] != routing_sha
or assertion["subject_sha256"] != subject_sha):
raise ApprovalStop("FAIL", "approver_assertion_binding_mismatch")
if assertion["decision"] != "approve":
raise ApprovalStop("FAIL", "approver_denied_request")
approver_id = assertion["approver_id"]
role = assertion["role"]
if approver_id == request["requester_id"]:
raise ApprovalStop("FAIL", "requester_self_approval_forbidden")
identity = identities[approver_id]
if role not in identity["roles"] or role not in required_roles:
raise ApprovalStop("FAIL", "approver_role_not_authorized_for_profile")
approved_at = parse_time(assertion["approved_at"], "assertion_time_invalid")
expires_at = parse_time(assertion["expires_at"], "assertion_time_invalid")
lifetime = (expires_at - approved_at).total_seconds()
if (approved_at < issued - timedelta(seconds=skew)
or approved_at > now + timedelta(seconds=skew)
or expires_at <= now or lifetime <= 0 or lifetime > maximum_lifetime):
raise ApprovalStop("FAIL", "approver_assertion_lifetime_invalid")
if (approver_id in approver_ids or assertion["approval_id"] in approval_ids
or assertion["nonce"] in nonces):
raise ApprovalStop("FAIL", "duplicate_approver_approval_or_nonce")
approver_ids.add(approver_id)
approval_ids.add(assertion["approval_id"])
nonces.add(assertion["nonce"])
observed_roles.add(role)
expires.append(expires_at)
if len(approver_ids) < minimum or not set(required_roles).issubset(observed_roles):
raise ApprovalStop("FAIL", "distinct_approver_or_role_coverage_insufficient")
decision_expires = min([request_expires, *expires])
context.update(request_id=request["request_id"], requester_id=request["requester_id"],
ticket_id=request["ticket_id"],
approval_profile_id=request["approval_profile_id"],
change_set_sha256=request["change_set_sha256"],
artifact_sha256=request["artifact_sha256"],
routing_sha256=routing_sha, subject_sha256=subject_sha,
approver_ids=sorted(approver_ids), approval_ids=sorted(approval_ids),
roles=sorted(observed_roles))
return decision_expires
def verify_output(path: Path) -> None:
subprocess.run([COSIGN, "verify-blob", "--key", str(VERIFY_KEY),
"--bundle", str(path) + ".sigstore.json", str(path)],
check=True, capture_output=True, text=True,
timeout=COMMAND_TIMEOUT)
def publish(record: dict[str, Any]) -> tuple[Path, str]:
OUTPUT.mkdir(parents=True, exist_ok=True, mode=0o750)
stem = "approval-" + secrets.token_urlsafe(18)
temporary = OUTPUT / ("." + stem + ".json.tmp")
final = OUTPUT / (stem + ".json")
temporary_bundle = Path(str(temporary) + ".sigstore.json")
final_bundle = Path(str(final) + ".sigstore.json")
payload = canonical(record) + b"\n"
with temporary.open("xb") as handle:
os.fchmod(handle.fileno(), 0o440)
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
subprocess.run([COSIGN, "sign-blob", "--yes", "--key", str(SIGNING_KEY),
"--bundle", str(temporary_bundle), str(temporary)],
check=True, capture_output=True, text=True,
timeout=COMMAND_TIMEOUT)
os.replace(temporary, final)
os.replace(temporary_bundle, final_bundle)
if final.read_bytes() != payload:
raise OSError("approval decision readback mismatch")
verify_output(final)
return final, sha256(payload)
def main() -> int:
context = {
"request_id": None, "requester_id": None, "ticket_id": None,
"approval_profile_id": None, "change_set_sha256": None,
"artifact_sha256": None, "routing_sha256": None, "subject_sha256": None,
"policy_sha256": None, "identity_registry_sha256": None,
"collection_sha256": None, "approver_ids": [], "approval_ids": [], "roles": [],
}
outcome, reason = "PASS", "distinct_qualified_approvals_verified"
expires_at = datetime.now(timezone.utc)
try:
expires_at = evaluate(context)
except ApprovalStop as error:
outcome, reason = error.outcome, error.reason
except (OSError, subprocess.SubprocessError):
outcome, reason = "ERROR", "trusted_approval_dependency_error"
except Exception:
outcome, reason = "ERROR", "unexpected_approval_broker_error"
now = datetime.now(timezone.utc)
record = {
"schema_version": "aidefend.h031.approval-decision.v2",
"control": "AID-H-031.004",
"outcome": outcome,
"reason_code": reason,
**context,
"issued_at": now.isoformat(),
"expires_at": expires_at.isoformat() if outcome == "PASS" else now.isoformat(),
}
path, digest = publish(record)
print(json.dumps({"outcome": outcome, "decision": str(path), "sha256": digest}))
return 0 if outcome == "PASS" else 2
if __name__ == "__main__":
raise SystemExit(main())
Action: The approval collection signer closes the complete assertion population before evaluation; the request signer, collection signer, approver keys and decision signer are separate identities. Missing assertions, duplicated identities, self-approval, wrong roles, stale or denied assertions, changed artifact/change set, unsigned UI comments, incomplete populations and verifier failures produce a signed non-PASS decision. Only a fresh PASS decision bound to the exact routing, subject, ticket and artifact is accepted by G001.
Independent verification
Replay one-approver, duplicate-approver, same-person-two-roles, requester-self-approval, inactive identity, wrong role, denial, omitted assertion, added assertion, forged collection, changed artifact, stale request, expired approval and broker-outage fixtures. The final promotion gate must reject every negative receipt and accept only the closed, distinct, role-complete positive population.
Open-source tools
- Cosign
- in-toto
- slsa-verifier
- Open Policy Agent (OPA)
- OpenTelemetry
- Sigstore Rekor v2 (promotion-evidence digest transparency anchoring; approval and promotion enforcement remain external)
Commercial / hosted tools
- GitHub Enterprise Cloud
- GitLab Ultimate
- JFrog Artifactory
- ServiceNow
Threat mappings
MITRE ATLAS
- AML.T0011 User Execution
- AML.T0010 AI Supply Chain Compromise
- AML.T0011.000 User Execution: Unsafe AI Artifacts
MAESTRO
- Supply Chain Attacks (Cross-Layer)
- Infrastructure-as-Code (IaC) Manipulation (L4)
- Privilege Escalation (Cross-Layer)
OWASP LLM Top 10 2026
- LLM10:2026 Improper Output Handling
OWASP ML Top 10 2023
- ML06:2023 AI Supply Chain Attacks
OWASP Top 10 for Agentic Applications 2026
- ASI04:2026 Agentic Supply Chain Vulnerabilities
- ASI05:2026 Unexpected Code Execution (RCE)
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- AISubtech-12.2.1 Code Detection / Malicious Code Output (requires exact artifact-bound verification receipts at final promotion)
- AISubtech-2.1.1 Execution Approval Bypass
- AITech-12.2 Insecure Output Handling (requires exact artifact-bound verification receipts at final promotion)
- AITech-2.1 Excessive Agency
Google Secure AI Framework 2.0 - Risks
- MDT: Model Deployment Tampering
- IIC: Insecure Integrated Component
Databricks AI Security Framework 3.0
- Platform 12.4: Unauthorized privileged access
- Platform 12.5: Poor security in the software development lifecycle