AID-M-009.006: Capability-Tier Control-Profile Binding & Assurance Gate

Version 1.20261015 · Actionable sub-technique

Bind an agent's approved autonomy, authority envelope, and blast-radius tier to a mandatory control profile before the agent is promoted or allowed to retain elevated privileges.

Pillars: app, infra

Lifecycle: scoping, validation, operation

Scope boundary

Owns the binding from an already classified autonomy, authority, capability, and blast-radius tier to the mandatory preventive, detective, containment, eviction, and recovery control profile, then gates promotion on current signed evidence for that profile.

AID-M-009.001: AID-M-009.006 determines which defensive controls an autonomy and capability tier requires; AID-M-009.001 determines how independently the agent may operate by function. The control profile consumes the autonomy classification but does not define it.

AID-M-009.002: AID-M-009.006 binds an aggregate authority and blast-radius tier to required assurance evidence; AID-M-009.002 defines the underlying action limits and risk classes. The assurance gate proves required safeguards, while the authority envelope defines what actions remain in bounds.

Implementation guidance

AID-M-009.006-G001

Maintain a control-profile binding matrix that maps each autonomy / authority / blast-radius tier to the minimum required AIDEFEND controls.
Concept:

Do not approve an agent tier in isolation. A tier only means something if it automatically selects the controls that must be present before that tier is allowed in production. The binding matrix, agent architecture record, and evidence index must each be signed by their own authority. A separate profile evaluator verifies those inputs, resolves the required controls, and signs the result consumed by the promotion gate.

Step 1: Define a tier-to-control matrix
# File: policy/agent_control_profiles.yaml
schema_version: aidefend.agent-control-profiles.v1
policy_version: "2026.07.1"
tier_profile_map:
  tier_0_assistive_read_only: tier_0_assistive_read_only
  tier_1_supervised_low_impact: tier_1_supervised_low_impact
  tier_2_autonomous_business_impact: tier_2_autonomous_business_impact
  tier_3_high_impact_or_irreversible: tier_3_high_impact_or_irreversible
max_architecture_evidence_age_seconds: 86400
profiles:
  tier_0_assistive_read_only:
    description: "No external side effects; can read approved context and draft responses."
    required_controls:
      - AID-M-009.001
      - AID-D-005.004
  tier_1_supervised_low_impact:
    description: "May call approved low-risk tools, but material side effects require approval."
    required_controls:
      - AID-M-009.001
      - AID-M-009.002
      - AID-H-018
      - AID-D-010
      - AID-I-003
  tier_2_autonomous_business_impact:
    description: "May execute bounded business workflows with reversible side effects."
    required_controls:
      - AID-M-009.002
      - AID-M-009.003
      - AID-H-018
      - AID-D-010
      - AID-I-003
      - AID-I-005
      - AID-R-003
    conditional_controls:
      - prerequisite: active_runtime_termination
        control: AID-E-002
      - prerequisite: durable_agent_state_teardown
        control: AID-E-005
  tier_3_high_impact_or_irreversible:
    description: "Can affect money, credentials, production deployment, destructive operations, or regulated data."
    required_controls:
      - AID-M-009.003
      - AID-M-009.004
      - AID-H-018
      - AID-H-021
      - AID-D-010
      - AID-D-016.001
      - AID-I-003
      - AID-I-005
      - AID-R-001
      - AID-R-003
    conditional_controls:
      - prerequisite: active_runtime_termination
        control: AID-E-002
      - prerequisite: durable_agent_state_teardown
        control: AID-E-005

The two eviction prerequisites above are architecture facts, not tier defaults. Mark active_runtime_termination applicable only when an executing process, pod, worker, run, or lease can persist; mark durable_agent_state_teardown applicable only when server-side session, memory, webhook, registration, queued-job, or schedule state can persist. A signed architecture inventory must supply each state and its evidence digest. Unknown state is INSUFFICIENT_DATA, not NOT_APPLICABLE.

Step 2: Evaluate the matrix against the system control inventory
from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
CONTROL_PROFILE_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("M009006_CONTROL_PROFILE_COMMAND_TIMEOUT_SECONDS")

# File: policy/evaluate_control_profile.py

import hashlib
import json
import os
import stat
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path

import yaml


def verify_blob(path: Path, signature: Path, key: Path) -> None:
    subprocess.run(
        [
            "cosign", "verify-blob", "--key", str(key),
            "--bundle", str(signature), str(path),
        ],
        check=True, capture_output=True, text=True, timeout=CONTROL_PROFILE_COMMAND_TIMEOUT_SECONDS,
    )


class StrictSafeLoader(yaml.SafeLoader):
    pass


def construct_unique_mapping(loader, node, deep=False):
    loader.flatten_mapping(node)
    value = {}
    for key_node, item_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        try:
            duplicate = key in value
        except TypeError as error:
            raise ValueError("YAML contains an unhashable mapping key") from error
        if duplicate:
            raise ValueError(f"duplicate YAML key: {key}")
        value[key] = loader.construct_object(item_node, deep=deep)
    return value


StrictSafeLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
    construct_unique_mapping,
)


def strict_yaml_bytes(raw: bytes, label: str) -> object:
    try:
        text = raw.decode("utf-8", errors="strict")
        loader = StrictSafeLoader(text)
        try:
            return loader.get_single_data()
        finally:
            loader.dispose()
    except (UnicodeDecodeError, yaml.YAMLError, ValueError) as error:
        raise ValueError(f"{label} is not strict UTF-8 YAML") from error


def read_regular_bytes(path: Path) -> bytes:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    with os.fdopen(descriptor, "rb") as handle:
        status = os.fstat(handle.fileno())
        if not stat.S_ISREG(status.st_mode) or status.st_size == 0:
            raise ValueError(f"missing, empty, or unsafe signed YAML: {path}")
        return handle.read()


def write_private_snapshot(path: Path, raw: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags, 0o400)
    with os.fdopen(descriptor, "wb") as handle:
        handle.write(raw)
        handle.flush()
        os.fsync(handle.fileno())


def load_verified_yaml(path: Path, signature: Path, key: Path) -> tuple[dict, str]:
    payload_raw = read_regular_bytes(path)
    signature_raw = read_regular_bytes(signature)
    with tempfile.TemporaryDirectory(prefix="aidefend-control-profile-") as directory:
        root = Path(directory)
        root.chmod(0o700)
        payload_snapshot = root / "payload.yaml"
        signature_snapshot = root / "payload.sig"
        write_private_snapshot(payload_snapshot, payload_raw)
        write_private_snapshot(signature_snapshot, signature_raw)
        verify_blob(payload_snapshot, signature_snapshot, key)
        if read_regular_bytes(payload_snapshot) != payload_raw or read_regular_bytes(signature_snapshot) != signature_raw:
            raise ValueError("private signed-YAML snapshot changed during verification")
    value = strict_yaml_bytes(payload_raw, str(path))
    if not isinstance(value, dict):
        raise ValueError(f"verified YAML is not an object: {path}")
    return value, hashlib.sha256(payload_raw).hexdigest()

AUTONOMY_FIELDS = {"autonomous_execute", "autonomous_delegate", "autonomous_persist"}
AUTHORITY_FIELDS = {"material_side_effects", "irreversible_actions", "money_or_credentials", "production_deployment", "regulated_data", "human_approval_required"}
CAPABILITY_FIELDS = {"approved_tool_calls", "external_side_effects"}
ARCHITECTURE_EVIDENCE_ROOT = Path("evidence/architecture-facts").resolve()


def boolean_facts(value: object, fields: set[str], label: str) -> dict:
    if not isinstance(value, dict) or set(value) != fields or any(not isinstance(value[field], bool) for field in fields):
        raise SystemExit(f"{label} fact schema differs")
    return value


def derive_tier(agent_record: dict) -> tuple[str, dict]:
    autonomy = boolean_facts(agent_record["autonomy_facts"], AUTONOMY_FIELDS, "autonomy")
    authority = boolean_facts(agent_record["authority_facts"], AUTHORITY_FIELDS, "authority")
    capability = boolean_facts(agent_record["capability_facts"], CAPABILITY_FIELDS, "capability")
    blast = agent_record["blast_radius_facts"]
    if not isinstance(blast, dict) or set(blast) != {"scope"} or blast["scope"] not in {"none", "single_task", "bounded_business", "high_impact"}:
        raise SystemExit("blast-radius fact schema differs")
    if authority["irreversible_actions"] or authority["money_or_credentials"] or authority["production_deployment"] or authority["regulated_data"] or blast["scope"] == "high_impact": tier = "tier_3_high_impact_or_irreversible"
    elif autonomy["autonomous_execute"] or autonomy["autonomous_delegate"] or autonomy["autonomous_persist"] or capability["external_side_effects"] or blast["scope"] == "bounded_business": tier = "tier_2_autonomous_business_impact"
    elif capability["approved_tool_calls"] or authority["material_side_effects"] or authority["human_approval_required"] or blast["scope"] == "single_task": tier = "tier_1_supervised_low_impact"
    else: tier = "tier_0_assistive_read_only"
    return tier, {"autonomy_facts": autonomy, "authority_facts": authority, "capability_facts": capability, "blast_radius_facts": blast}


def required_controls_for(
    profile_id: str,
    profiles: dict,
    applicability: dict,
) -> tuple[set[str], dict]:
    try:
        profile = profiles["profiles"][profile_id]
    except KeyError as exc:
        raise SystemExit(f"unknown control profile: {profile_id}") from exc
    required = set(profile["required_controls"])
    resolved = {}
    for rule in profile.get("conditional_controls", []):
        if not isinstance(rule, dict) or set(rule) != {"prerequisite", "control"}:
            raise SystemExit("conditional control schema differs")
        prerequisite = rule["prerequisite"]
        record = applicability.get(prerequisite)
        if (
            not isinstance(record, dict)
            or set(record) != {"state", "evidence_path", "evidence_sha256", "observed_at", "expires_at"}
            or record["state"] not in {"APPLICABLE", "NOT_APPLICABLE"}
            or not isinstance(record["evidence_path"], str) or not record["evidence_path"]
            or not isinstance(record["evidence_sha256"], str) or len(record["evidence_sha256"]) != 64
            or set(record["evidence_sha256"]) - set("0123456789abcdef")
        ):
            raise SystemExit(f"unmeasured prerequisite: {prerequisite}")
        evidence_path = (ARCHITECTURE_EVIDENCE_ROOT / record["evidence_path"]).resolve()
        if not evidence_path.is_relative_to(ARCHITECTURE_EVIDENCE_ROOT) or hashlib.sha256(read_regular_bytes(evidence_path)).hexdigest() != record["evidence_sha256"]:
            raise SystemExit(f"architecture evidence differs: {prerequisite}")
        observed = datetime.fromisoformat(record["observed_at"].replace("Z", "+00:00")).astimezone(timezone.utc)
        expires = datetime.fromisoformat(record["expires_at"].replace("Z", "+00:00")).astimezone(timezone.utc)
        now = datetime.now(timezone.utc)
        if observed > now or expires <= now or expires <= observed or (now - observed).total_seconds() > profiles["max_architecture_evidence_age_seconds"]:
            raise SystemExit(f"architecture evidence is stale: {prerequisite}")
        resolved[prerequisite] = record
        if record["state"] == "APPLICABLE":
            required.add(rule["control"])
    return required, resolved


def evaluate_profile(agent_record: dict, evidence_inventory: dict, profiles: dict) -> dict:
    if (
        set(profiles) != {"schema_version", "policy_version", "tier_profile_map", "max_architecture_evidence_age_seconds", "profiles"}
        or profiles.get("schema_version") != "aidefend.agent-control-profiles.v1"
        or not isinstance(profiles.get("policy_version"), str)
        or not profiles["policy_version"]
        or not isinstance(profiles.get("tier_profile_map"), dict)
        or set(profiles["tier_profile_map"]) != {"tier_0_assistive_read_only", "tier_1_supervised_low_impact", "tier_2_autonomous_business_impact", "tier_3_high_impact_or_irreversible"}
        or isinstance(profiles.get("max_architecture_evidence_age_seconds"), bool)
        or not isinstance(profiles.get("max_architecture_evidence_age_seconds"), (int, float))
        or profiles["max_architecture_evidence_age_seconds"] <= 0
        or not isinstance(profiles.get("profiles"), dict)
        or not profiles["profiles"]
    ):
        raise SystemExit("control-profile policy schema differs")
    for profile_name, profile in profiles["profiles"].items():
        allowed_keys = {"description", "required_controls", "conditional_controls"}
        if (
            not isinstance(profile_name, str) or not profile_name
            or not isinstance(profile, dict)
            or not {"description", "required_controls"}.issubset(profile)
            or set(profile) - allowed_keys
            or not isinstance(profile["description"], str)
            or not profile["description"].strip()
            or not isinstance(profile["required_controls"], list)
            or not profile["required_controls"]
            or any(not isinstance(item, str) or not item for item in profile["required_controls"])
            or len(profile["required_controls"]) != len(set(profile["required_controls"]))
            or not isinstance(profile.get("conditional_controls", []), list)
        ):
            raise SystemExit(f"control-profile entry schema differs: {profile_name}")
    required_agent_fields = {
        "agent_id", "agent_version", "candidate_tier", "control_profile",
        "autonomy_facts", "authority_facts", "capability_facts", "blast_radius_facts",
        "control_applicability",
    }
    if set(agent_record) != required_agent_fields:
        raise SystemExit("signed agent architecture record schema differs")
    derived_tier, derivation_facts = derive_tier(agent_record)
    profile_id = profiles["tier_profile_map"].get(derived_tier)
    if agent_record["candidate_tier"] != derived_tier or agent_record["control_profile"] != profile_id or profile_id not in profiles["profiles"]:
        raise SystemExit("candidate tier or control profile differs from architecture facts")
    required, applicability = required_controls_for(
        profile_id,
        profiles,
        agent_record["control_applicability"],
    )
    agent_evidence = evidence_inventory.get("agents", {}).get(
        agent_record["agent_id"]
    )
    if not isinstance(agent_evidence, dict):
        raise SystemExit("signed evidence index has no complete agent population")
    receipt_digests = {}
    for control, record in agent_evidence.items():
        if (
            not isinstance(record, dict)
            or set(record) != {"receipt_sha256"}
            or not isinstance(record["receipt_sha256"], str)
            or len(record["receipt_sha256"]) != 64
            or set(record["receipt_sha256"]) - set("0123456789abcdef")
        ):
            raise SystemExit(f"invalid evidence index record for {control}")
        receipt_digests[control] = record["receipt_sha256"]
    implemented = set(receipt_digests)
    missing = sorted(required - implemented)
    return {
        "schema_version": "aidefend.control-profile-result.v1",
        "agent_id": agent_record["agent_id"],
        "agent_version": agent_record["agent_version"],
        "candidate_tier": derived_tier,
        "control_profile": profile_id,
        "tier_derivation_facts": derivation_facts,
        "control_profile_policy_version": profiles["policy_version"],
        "required_controls": sorted(required),
        "evidence_receipt_sha256_by_control": {
            control: receipt_digests[control]
            for control in sorted(required & implemented)
        },
        "missing_controls": missing,
        "resolved_control_applicability": applicability,
        "status": "RESOLVED" if not missing else "INCOMPLETE",
    }


def main() -> None:
    profiles, profiles_sha256 = load_verified_yaml(
        Path("policy/agent_control_profiles.yaml"),
        Path("policy/agent_control_profiles.sig"),
        Path("keys/control-profile-authority.pub"),
    )
    agent, agent_sha256 = load_verified_yaml(
        Path("agents/support-agent-prod.yaml"),
        Path("agents/support-agent-prod.sig"),
        Path("keys/agent-architecture-authority.pub"),
    )
    evidence, evidence_sha256 = load_verified_yaml(
        Path("evidence/control_inventory.yaml"),
        Path("evidence/control_inventory.sig"),
        Path("keys/evidence-index-authority.pub"),
    )
    result = evaluate_profile(agent, evidence, profiles)
    result.update({
        "profile_policy_sha256": profiles_sha256,
        "agent_record_sha256": agent_sha256,
        "evidence_index_sha256": evidence_sha256,
    })
    result_path = Path("artifacts/control_profile_result.json")
    result_path.write_text(
        json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )
    subprocess.run(
        [
            "cosign", "sign-blob", "--yes",
            "--key", "env://CONTROL_PROFILE_EVALUATOR_SIGNING_KEY",
            "--bundle", str(result_path.with_suffix(".sig")),
            str(result_path),
        ],
        check=True, timeout=CONTROL_PROFILE_COMMAND_TIMEOUT_SECONDS,
    )
    verify_blob(
        result_path, result_path.with_suffix(".sig"),
        Path("keys/control-profile-evaluator.pub"),
    )
    if result["status"] != "RESOLVED":
        raise SystemExit(
            "control profile resolution incomplete: missing evidence references "
            + ", ".join(result["missing_controls"])
        )


if __name__ == "__main__":
    main()
Step 3: Treat the signed profile decision as release evidence

Persist the exact signed result and its three verified input digests with the agent release record. The result identifies the required-control set derived from exact autonomy, authority, capability, and blast-radius facts. Its RESOLVED state is deliberately not PASS and is not itself proof that each control passed. The assurance gate below must independently verify every referenced receipt.

Negative parser and snapshot fixtures

Run these tests whenever the signed YAML schema, signer, or control-profile evaluator changes.

# File: tests/test_control_profile_inputs.py
from __future__ import annotations

import hashlib
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

os.environ.setdefault("AIDEFEND_RUNTIME_PROFILE_VERSION", "test-v1")
os.environ.setdefault("AIDEFEND_RUNTIME_PROFILE_SHA256", "1" * 64)
os.environ.setdefault("M009006_CONTROL_PROFILE_COMMAND_TIMEOUT_SECONDS", "2")

from policy.evaluate_control_profile import (  # noqa: E402
    load_verified_yaml,
    strict_yaml_bytes,
)


class ControlProfileInputTests(unittest.TestCase):
    def test_duplicate_yaml_key_is_rejected(self):
        with self.assertRaises(ValueError):
            strict_yaml_bytes(b"schema_version: v1\nschema_version: v2\n", "fixture")

    def test_source_replacement_cannot_change_verified_snapshot(self):
        original = b"schema_version: fixture.v1\nvalue: original\n"
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            payload = root / "input.yaml"
            signature = root / "input.sig"
            payload.write_bytes(original)
            signature.write_bytes(b"signature")

            def replace_source(payload_snapshot, signature_snapshot, _key):
                self.assertNotEqual(payload_snapshot, payload)
                self.assertNotEqual(signature_snapshot, signature)
                payload.write_bytes(b"schema_version: attacker.v1\n")

            with patch(
                "policy.evaluate_control_profile.verify_blob",
                side_effect=replace_source,
            ):
                value, digest = load_verified_yaml(
                    payload, signature, root / "authority.pub",
                )
            self.assertEqual(value["value"], "original")
            self.assertEqual(digest, hashlib.sha256(original).hexdigest())


if __name__ == "__main__":
    unittest.main()

Action: Keep the profile author, architecture authority, evidence indexer, evaluator, and promotion gate as separately authorized identities. A self-asserted tier, opaque applicability digest, stale or mismatched referenced architecture evidence, missing signature, unknown prerequisite, missing receipt reference, or missing required control denies promotion.

Profile real control independence

When a capability tier requires defense in depth, record whether selected controls share the same writable policy, credential, runtime, network enforcement point, telemetry path, or disable authority. Correlated controls may still be required, but they do not satisfy an independence requirement merely because they have different AIDEFEND IDs. AID-H-021.004 owns the enforcement result that the workload cannot modify egress, load balancers, routes, security groups, monitoring, or kill switches; this profile consumes that result and does not create another PASS. Any break-glass path that crosses layers must be policy-selected, quorum-approved, just-in-time, time-bounded, independently alerted, and unable to erase its audit trail.

AID-M-009.006-G002

Require the signed control evidence named by the capability-tier profile before promotion, and fail closed when a required record is missing, invalid, or stale.
Gate contract

RESOLVED from the profile evaluator is deliberately non-PASS. The promotion gate must independently reproduce the tier, selected profile, and conditional-control set from a separately signed release manifest and promotion policy. It may approve only after resolving the digest-addressed release and every receipt-referenced evidence artifact to the exact bytes named by their digests.

{
  "schema_version": "aidefend.agent-release-manifest.v2",
  "agent_id": "support-agent-prod",
  "agent_version": "2026.06.24.3",
  "agent_record_sha256": "5123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
  "agent_release_digest": "sha256:4123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
  "artifact_uri": "oci://registry.example.com/agents/support@sha256:4123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
  "tier_derivation_facts": {
    "autonomy_facts": {"autonomous_execute": true, "autonomous_delegate": false, "autonomous_persist": false},
    "authority_facts": {"material_side_effects": true, "irreversible_actions": false, "money_or_credentials": false, "production_deployment": false, "regulated_data": false, "human_approval_required": true},
    "capability_facts": {"approved_tool_calls": true, "external_side_effects": true},
    "blast_radius_facts": {"scope": "bounded_business"}
  },
  "resolved_control_applicability": {
    "active_runtime_termination": {"state": "APPLICABLE", "evidence_path": "runtime-termination.json", "evidence_sha256": "6123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", "observed_at": "2026-08-29T12:00:00Z", "expires_at": "2026-08-30T12:00:00Z"}
  }
}

The separately signed promotion policy supplies the tier-to-profile map, each profile's unconditional and conditional controls, evidence freshness limits, approved receipt keys, and bounded resolver timeout. These are deployment policy values, not AIDEFEND defaults.

Fail-closed verifier
from __future__ import annotations

import hashlib
import json
import os
import re
import stat
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path

ROOT = Path("/evidence").resolve()
RECEIPT_ROOT = (ROOT / "receipts").resolve()
ARTIFACT_ROOT = (ROOT / "artifacts").resolve()
ARCHITECTURE_ROOT = (ROOT / "architecture-facts").resolve()
SHA = re.compile(r"^[0-9a-f]{64}$")
DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
VERIFIED: dict[Path, bytes] = {}


class Deny(RuntimeError):
    pass


def regular_bytes(path: Path) -> bytes:
    fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    with os.fdopen(fd, "rb") as handle:
        metadata = os.fstat(handle.fileno())
        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size == 0:
            raise Deny(f"unsafe or empty input: {path}")
        return handle.read()


def private_file(path: Path, raw: bytes) -> None:
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL
                 | getattr(os, "O_NOFOLLOW", 0), 0o400)
    with os.fdopen(fd, "wb") as handle:
        handle.write(raw)
        handle.flush()
        os.fsync(handle.fileno())


def verify_blob(path: Path, signature: Path, key: Path, timeout: float) -> bytes:
    raw, signature_raw, key_raw = regular_bytes(path), regular_bytes(signature), regular_bytes(key)
    with tempfile.TemporaryDirectory(prefix="aidefend-promotion-") as directory:
        stage = Path(directory)
        stage.chmod(0o700)
        private_file(stage / "payload", raw)
        private_file(stage / "payload.sig", signature_raw)
        private_file(stage / "verify.pub", key_raw)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(stage / "verify.pub"),
             "--bundle", str(stage / "payload.sig"), str(stage / "payload")],
            check=True, capture_output=True, timeout=timeout,
        )
        if regular_bytes(stage / "payload") != raw:
            raise Deny("verified snapshot changed")
    VERIFIED[path] = raw
    return raw


def strict_json(raw: bytes, label: str) -> dict:
    def unique(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise Deny(f"duplicate JSON key in {label}: {key}")
            value[key] = item
        return value
    value = json.loads(raw.decode("utf-8", errors="strict"),
                       object_pairs_hook=unique,
                       parse_constant=lambda item: (_ for _ in ()).throw(Deny(item)))
    if not isinstance(value, dict):
        raise Deny(f"{label} is not an object")
    return value


def load_signed(path: Path, signature: Path, key: Path, timeout: float) -> dict:
    return strict_json(verify_blob(path, signature, key, timeout), str(path))


def beneath(root: Path, relative: object) -> Path:
    if not isinstance(relative, str) or not relative or Path(relative).is_absolute():
        raise Deny("unsafe relative evidence path")
    result = (root / relative).resolve()
    if not result.is_relative_to(root):
        raise Deny("evidence path escapes its root")
    return result


def timestamp(value: object) -> datetime:
    if not isinstance(value, str):
        raise Deny("timestamp is absent")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise Deny("timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


AUTONOMY = {"autonomous_execute", "autonomous_delegate", "autonomous_persist"}
AUTHORITY = {"material_side_effects", "irreversible_actions", "money_or_credentials",
             "production_deployment", "regulated_data", "human_approval_required"}
CAPABILITY = {"approved_tool_calls", "external_side_effects"}


def booleans(value: object, fields: set[str], label: str) -> dict:
    if (not isinstance(value, dict) or set(value) != fields
            or any(not isinstance(value[field], bool) for field in fields)):
        raise Deny(f"{label} facts differ")
    return value


def derive_tier(facts: object) -> str:
    if not isinstance(facts, dict) or set(facts) != {
        "autonomy_facts", "authority_facts", "capability_facts", "blast_radius_facts"
    }:
        raise Deny("tier fact population differs")
    autonomy = booleans(facts["autonomy_facts"], AUTONOMY, "autonomy")
    authority = booleans(facts["authority_facts"], AUTHORITY, "authority")
    capability = booleans(facts["capability_facts"], CAPABILITY, "capability")
    blast = facts["blast_radius_facts"]
    if (not isinstance(blast, dict) or set(blast) != {"scope"}
            or blast["scope"] not in {"none", "single_task", "bounded_business", "high_impact"}):
        raise Deny("blast-radius facts differ")
    if (authority["irreversible_actions"] or authority["money_or_credentials"]
            or authority["production_deployment"] or authority["regulated_data"]
            or blast["scope"] == "high_impact"):
        return "tier_3_high_impact_or_irreversible"
    if (autonomy["autonomous_execute"] or autonomy["autonomous_delegate"]
            or autonomy["autonomous_persist"] or capability["external_side_effects"]
            or blast["scope"] == "bounded_business"):
        return "tier_2_autonomous_business_impact"
    if (capability["approved_tool_calls"] or authority["material_side_effects"]
            or authority["human_approval_required"] or blast["scope"] == "single_task"):
        return "tier_1_supervised_low_impact"
    return "tier_0_assistive_read_only"


def derive_required(profile_id: str, applicability: object,
                    policy: dict, now: datetime) -> list[str]:
    contract = policy["control_profiles"].get(profile_id)
    if not isinstance(contract, dict) or set(contract) != {"required_controls", "conditional_controls"}:
        raise Deny("profile contract is absent")
    required, rules = contract["required_controls"], contract["conditional_controls"]
    if (not isinstance(required, list) or not required
            or any(not isinstance(item, str) or not item for item in required)
            or len(required) != len(set(required)) or not isinstance(rules, list)):
        raise Deny("profile contract differs")
    conditional = {}
    for rule in rules:
        if (not isinstance(rule, dict) or set(rule) != {"prerequisite", "control"}
                or any(not isinstance(rule[field], str) or not rule[field]
                       for field in ("prerequisite", "control"))
                or rule["prerequisite"] in conditional):
            raise Deny("conditional-control rule differs")
        conditional[rule["prerequisite"]] = rule["control"]
    if not isinstance(applicability, dict) or set(applicability) != set(conditional):
        raise Deny("conditional applicability population differs")
    result = set(required)
    for prerequisite, control in conditional.items():
        row = applicability[prerequisite]
        if (not isinstance(row, dict)
                or set(row) != {"state", "evidence_path", "evidence_sha256",
                                "observed_at", "expires_at"}
                or row["state"] not in {"APPLICABLE", "NOT_APPLICABLE"}
                or not isinstance(row["evidence_sha256"], str)
                or not SHA.fullmatch(row["evidence_sha256"])):
            raise Deny(f"conditional row differs: {prerequisite}")
        evidence = beneath(ARCHITECTURE_ROOT, row["evidence_path"])
        if hashlib.sha256(regular_bytes(evidence)).hexdigest() != row["evidence_sha256"]:
            raise Deny(f"conditional evidence differs: {prerequisite}")
        observed, expires = timestamp(row["observed_at"]), timestamp(row["expires_at"])
        if (observed > now or expires <= now or expires <= observed
                or now - observed > timedelta(
                    seconds=policy["max_architecture_evidence_age_seconds"])):
            raise Deny(f"conditional evidence is stale: {prerequisite}")
        if row["state"] == "APPLICABLE":
            result.add(control)
    return sorted(result)


def resolve_artifact(uri: object, expected: str, timeout: float) -> str:
    if not isinstance(uri, str) or not uri or not SHA.fullmatch(expected):
        raise Deny("artifact binding differs")
    if uri.startswith("file://"):
        raw = regular_bytes(Path(uri[7:]).resolve())
        if not Path(uri[7:]).resolve().is_relative_to(ARTIFACT_ROOT):
            raise Deny("artifact path escapes its root")
    elif uri.startswith("oci://"):
        reference = uri[6:]
        if "@sha256:" not in reference or reference.rsplit("@sha256:", 1)[1] != expected:
            raise Deny("OCI reference is not bound to the expected digest")
        raw = subprocess.run(["oras", "manifest", "fetch", reference],
                             check=True, capture_output=True, timeout=timeout).stdout
        if not raw:
            raise Deny("OCI resolver returned no manifest bytes")
    else:
        raise Deny("unsupported artifact resolver")
    actual = hashlib.sha256(raw).hexdigest()
    if actual != expected:
        raise Deny("resolved artifact bytes differ")
    return actual


def verify(now: datetime) -> dict:
    policy = load_signed(
        Path("/etc/aidefend/promotion/policy.json"),
        Path("/etc/aidefend/promotion/policy.sig"),
        Path("/opt/aidefend/trust/promotion-policy.pub"), 30,
    )
    policy_fields = {"schema_version", "policy_version", "max_evidence_age_seconds",
                     "max_architecture_evidence_age_seconds", "command_timeout_seconds",
                     "release_manifest_verify_key", "receipt_verify_keys",
                     "tier_profile_map", "control_profiles"}
    if (set(policy) != policy_fields
            or policy["schema_version"] != "aidefend.promotion-assurance-policy.v2"
            or any(isinstance(policy[field], bool) or not isinstance(policy[field], (int, float))
                   or policy[field] <= 0
                   for field in ("max_evidence_age_seconds",
                                 "max_architecture_evidence_age_seconds",
                                 "command_timeout_seconds"))
            or not isinstance(policy["tier_profile_map"], dict)
            or not isinstance(policy["control_profiles"], dict)
            or not isinstance(policy["receipt_verify_keys"], dict)):
        raise Deny("promotion policy differs")
    timeout = float(policy["command_timeout_seconds"])

    profile_path = ROOT / "control_profile_result.json"
    profile = load_signed(profile_path, ROOT / "control_profile_result.sig",
                          Path("/opt/aidefend/trust/control-profile-evaluator.pub"), timeout)
    release_path = ROOT / "agent_release_manifest.json"
    release = load_signed(release_path, ROOT / "agent_release_manifest.sig",
                          Path(policy["release_manifest_verify_key"]), timeout)
    package_path = ROOT / "agent_assurance_package.json"
    package = load_signed(package_path, ROOT / "agent_assurance_package.sig",
                          Path("/opt/aidefend/trust/evidence-package-assembler.pub"), timeout)

    profile_fields = {"schema_version", "agent_id", "agent_version", "candidate_tier",
                      "control_profile", "tier_derivation_facts",
                      "control_profile_policy_version", "required_controls",
                      "evidence_receipt_sha256_by_control", "missing_controls",
                      "resolved_control_applicability", "status",
                      "profile_policy_sha256", "agent_record_sha256",
                      "evidence_index_sha256"}
    release_fields = {"schema_version", "agent_id", "agent_version", "agent_record_sha256",
                      "agent_release_digest", "artifact_uri", "tier_derivation_facts",
                      "resolved_control_applicability"}
    package_fields = {"schema_version", "agent_id", "agent_version", "candidate_tier",
                      "agent_release_digest", "control_profile_result_sha256", "evidence"}
    if (set(profile) != profile_fields
            or profile["schema_version"] != "aidefend.control-profile-result.v1"
            or profile["status"] != "RESOLVED" or profile["missing_controls"] != []):
        raise Deny("profile resolution is incomplete")
    if (set(release) != release_fields
            or release["schema_version"] != "aidefend.agent-release-manifest.v2"
            or not DIGEST.fullmatch(str(release["agent_release_digest"]))
            or release["agent_record_sha256"] != profile["agent_record_sha256"]
            or release["tier_derivation_facts"] != profile["tier_derivation_facts"]
            or release["resolved_control_applicability"]
               != profile["resolved_control_applicability"]):
        raise Deny("release manifest differs")
    if (set(package) != package_fields
            or package["schema_version"] != "aidefend.agent-assurance-package.v1"
            or package["control_profile_result_sha256"]
               != hashlib.sha256(VERIFIED[profile_path]).hexdigest()
            or any(package[field] != profile[field]
                   for field in ("agent_id", "agent_version", "candidate_tier"))
            or package["agent_release_digest"] != release["agent_release_digest"]
            or any(release[field] != profile[field]
                   for field in ("agent_id", "agent_version"))):
        raise Deny("assurance package differs")

    tier = derive_tier(release["tier_derivation_facts"])
    selected_profile = policy["tier_profile_map"].get(tier)
    required = derive_required(selected_profile,
                               release["resolved_control_applicability"], policy, now)
    if (profile["candidate_tier"] != tier
            or profile["control_profile"] != selected_profile
            or profile["required_controls"] != required):
        raise Deny("tier, profile, or condition generation differs")
    release_sha = resolve_artifact(
        release["artifact_uri"],
        release["agent_release_digest"].removeprefix("sha256:"), timeout,
    )

    if (not isinstance(package["evidence"], dict)
            or set(package["evidence"]) != set(required)
            or set(profile["evidence_receipt_sha256_by_control"]) != set(required)):
        raise Deny("receipt population differs")
    receipts, evidence_artifacts = {}, {}
    for control in required:
        locator = package["evidence"][control]
        if not isinstance(locator, dict) or set(locator) != {"receipt_path", "signature_path"}:
            raise Deny(f"receipt locator differs: {control}")
        receipt_path = beneath(RECEIPT_ROOT, locator["receipt_path"])
        signature_path = beneath(RECEIPT_ROOT, locator["signature_path"])
        key = policy["receipt_verify_keys"].get(control)
        if not isinstance(key, str) or not key:
            raise Deny(f"approved receipt key is absent: {control}")
        receipt = load_signed(receipt_path, signature_path, Path(key), timeout)
        receipt_fields = {"schema_version", "control", "agent_id", "agent_version",
                          "agent_release_digest", "status", "checked_at",
                          "artifact_uri", "artifact_sha256"}
        receipt_sha = hashlib.sha256(VERIFIED[receipt_path]).hexdigest()
        if (set(receipt) != receipt_fields
                or receipt["schema_version"] != "aidefend.control-assurance-receipt.v1"
                or receipt["control"] != control or receipt["status"] != "PASS"
                or receipt["agent_id"] != profile["agent_id"]
                or receipt["agent_version"] != profile["agent_version"]
                or receipt["agent_release_digest"] != release["agent_release_digest"]
                or receipt_sha != profile["evidence_receipt_sha256_by_control"][control]
                or not isinstance(receipt["artifact_sha256"], str)
                or not SHA.fullmatch(receipt["artifact_sha256"])):
            raise Deny(f"receipt differs: {control}")
        checked = timestamp(receipt["checked_at"])
        if checked > now or now - checked > timedelta(
                seconds=policy["max_evidence_age_seconds"]):
            raise Deny(f"receipt is stale: {control}")
        evidence_artifacts[control] = resolve_artifact(
            receipt["artifact_uri"], receipt["artifact_sha256"], timeout
        )
        receipts[control] = receipt_sha

    return {
        "schema_version": "aidefend.agent-promotion-decision.v2",
        "decision": "APPROVE",
        "agent_id": profile["agent_id"],
        "agent_version": profile["agent_version"],
        "candidate_tier": tier,
        "release_artifact_sha256": release_sha,
        "verified_receipt_sha256_by_control": receipts,
        "verified_evidence_artifact_sha256_by_control": evidence_artifacts,
        "decided_at": now.isoformat(),
    }


result_path = ROOT / "agent_promotion_decision.json"
try:
    result = verify(datetime.now(timezone.utc))
    exit_code = 0
except Deny as error:
    result = {"schema_version": "aidefend.agent-promotion-decision.v2",
              "decision": "DENY", "reason": str(error),
              "decided_at": datetime.now(timezone.utc).isoformat()}
    exit_code = 3
except Exception as error:
    result = {"schema_version": "aidefend.agent-promotion-decision.v2",
              "decision": "ERROR", "reason": type(error).__name__,
              "decided_at": datetime.now(timezone.utc).isoformat()}
    exit_code = 2
result_path.write_text(json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n",
                       encoding="utf-8")
raise SystemExit(exit_code)
Admission handoff

Run the verifier under a read-only evidence identity, sign its exact decision bytes with a separate promotion-gate key, and have the privilege controller accept only a fresh signed APPROVE bound to the candidate release digest. A self-asserted tier, a merely RESOLVED profile, stale condition facts, an unresolvable OCI digest, a receipt-only artifact claim, or any missing byte population is non-PASS and preserves the current tier.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-M-009.006: Capability-Tier Control-Profile Binding & Assurance Gate | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service