Scope boundary

Owns materialization, exact agent/release/authority/goal binding, signing or reuse of an already suitable signed artifact, publication, rotation/revocation metadata and complete gate-delivery verification for the agent-wide behavior contract. It does not independently redefine authority, prove semantic policy safety, admit arbitrary runtime configuration, or authorize actions.

AID-M-009.002: AID-H-017.005 consumes approved authority facts and binds the deployable contract; AID-M-009.002 defines concrete authority limits and action-risk classification. No capability can be added by the contract materializer.

AID-M-003.004: AID-H-017.005 binds the contract to the verified effective mission digest; AID-M-003.004 owns the trusted goal baseline and its meaning. The contract handoff does not approve a new mission.

AID-H-021.002: AID-H-017.005 verifies agent-wide contract binding and delivery to the complete gate population; AID-H-021.002 owns generic fail-closed startup and runtime artifact admission. Reuse its signing and verification infrastructure; the same signature or loaded-artifact fact receives no duplicate integrity credit.

AID-H-018.002: AID-H-017.005 publishes the signed agent-wide behavior contract; AID-H-018.002 evaluates the current action against active policy and verified context. The contract supplies authority facts but never pre-authorizes every request.

AID-H-018.007: AID-H-017.005 defines what the agent as a whole is intended to do; AID-H-018.007 validates and enforces the narrower permission manifest of each installed skill. Agent-level authority and skill-level authority are nested but independently enforced.

Implementation guidance

• AID-H-017.005-G001: Materialize a release-bound behavior contract from approved capability and mission inputs without redefining upstream authority.
• AID-H-017.005-G002: Independently verify exact contract delivery to every expected gate using per-instance authorized readback signers and a fresh replay-protected challenge.

AID-H-017.005-G001

Applicability and canonical ownership

Apply this method when the agent authorization platform consumes an agent-wide behavior contract. The authority service materializes the approved M-009.002 facts into a signed, closed capability profile and binds it to the current M-003.004 goal and exact release. This is a defined deployment adapter, not an assertion that the authority-envelope YAML and capability-profile schemas are identical. Obtain the expected tuple independently from the signature-verified deployment manifest.

# File: materialize_behavior_contract.py
from __future__ import annotations
import base64, hashlib, json, re
from dataclasses import dataclass
from datetime import datetime, timezone
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
import rfc8785

PROFILE_FIELDS = {"schema_version", "profile_version", "agent_id", "release_id", "agent_release_digest",
    "authority_envelope_sha256", "goal_manifest_sha256", "policy_version", "cert_id", "issuer",
    "issued_at", "expires_at", "capabilities", "critical_operations"}
SHA256 = re.compile(r"^[0-9a-f]{64}$")

@dataclass(frozen=True)
class VerifiedReleaseContext:
    agent_id: str
    release_id: str
    agent_release_digest: str
    authority_envelope_sha256: str
    goal_manifest_sha256: str
    policy_version: str

def duplicate_free(pairs):
    result = {}
    for key, value in pairs:
        if key in result: raise ValueError("duplicate profile key")
        result[key] = value
    return result

def materialize_contract(profile_envelope, *, trusted_profile_keys, authorized_profile_key_ids,
                         expected: VerifiedReleaseContext, now: datetime):
    if set(profile_envelope) != {"payload_b64", "signature_b64", "key_id"}: raise ValueError("profile envelope differs")
    kid = profile_envelope["key_id"]
    if kid not in authorized_profile_key_ids or kid not in trusted_profile_keys: raise ValueError("unauthorized capability authority")
    key = trusted_profile_keys[kid]
    if not isinstance(key, Ed25519PublicKey): raise ValueError("profile verifier expects Ed25519")
    raw = base64.b64decode(profile_envelope["payload_b64"], validate=True)
    key.verify(base64.b64decode(profile_envelope["signature_b64"], validate=True), raw)
    profile = json.loads(raw.decode("utf-8"), object_pairs_hook=duplicate_free,
        parse_constant=lambda x: (_ for _ in ()).throw(ValueError("non-finite JSON")))
    if not isinstance(profile, dict) or set(profile) != PROFILE_FIELDS or profile["schema_version"] != "aidefend.agent-capability-profile.v2": raise ValueError("profile schema differs")
    if rfc8785.dumps(profile) != raw: raise ValueError("profile bytes are not canonical")
    for field in ("profile_version", "agent_id", "release_id", "policy_version", "cert_id", "issuer"):
        if not isinstance(profile[field], str) or not profile[field].strip(): raise ValueError("missing profile identity")
    for field in ("agent_release_digest", "authority_envelope_sha256", "goal_manifest_sha256"):
        if not isinstance(profile[field], str) or not SHA256.fullmatch(profile[field]): raise ValueError("invalid binding digest")
    for field in expected.__dataclass_fields__:
        if profile[field] != getattr(expected, field): raise ValueError("release/authority/goal binding differs")
    issued = datetime.fromisoformat(profile["issued_at"].replace("Z", "+00:00"))
    expires = datetime.fromisoformat(profile["expires_at"].replace("Z", "+00:00"))
    if now.tzinfo is None or issued.tzinfo is None or expires.tzinfo is None or not issued <= now < expires:
        raise ValueError("profile is future, expired or has an invalid interval")
    caps = profile["capabilities"]
    if not isinstance(caps, dict) or set(caps) != {"allowed_tools", "file_access_scopes", "network_access"}: raise ValueError("capability schema differs")
    tools = caps["allowed_tools"]
    if not isinstance(tools, dict) or not tools: raise ValueError("missing tools")
    for name, limits in tools.items():
        if not isinstance(name, str) or not name or not isinstance(limits, dict): raise ValueError("invalid tool")
        for limit, value in limits.items():
            if not isinstance(limit, str) or not limit or type(value) is not int or value < 1: raise ValueError("invalid tool limit")
    scopes = caps["file_access_scopes"]
    if not isinstance(scopes, list): raise ValueError("invalid file scope population")
    for scope in scopes:
        if not isinstance(scope, dict) or set(scope) != {"path", "permissions"} or not isinstance(scope["path"], str) or not scope["path"].startswith("/"):
            raise ValueError("invalid file scope")
        permissions = scope["permissions"]
        if not isinstance(permissions, list) or not permissions or any(p not in {"read", "write"} for p in permissions) or len(permissions) != len(set(permissions)):
            raise ValueError("invalid file permissions")
    network = caps["network_access"]
    if not isinstance(network, dict) or set(network) != {"allowed_domains"}: raise ValueError("invalid network scope")
    domains = network["allowed_domains"]
    if not isinstance(domains, list) or any(not isinstance(d, str) or not d for d in domains) or len(domains) != len(set(domains)): raise ValueError("invalid domains")
    critical = profile["critical_operations"]
    if not isinstance(critical, list) or any(not isinstance(x, str) for x in critical) or len(critical) != len(set(critical)) or not set(critical).issubset(tools): raise ValueError("invalid critical operations")
    contract = {**profile, "schema_version": "aidefend.behavior-contract.v2",
                "source_profile_sha256": hashlib.sha256(raw).hexdigest()}
    return rfc8785.dumps(contract)

Publish exact bytes and preserve the upstream handoff

Pin pyca/cryptography and the RFC8785 canonicalizer in a hash-locked release image. The concrete profile adapter above uses Ed25519; another verified signing path is acceptable with equivalent canonical-byte, authorized-issuer and binding properties. Sign the returned exact bytes through the protected release signer, publish immutable bytes and its signature/trust metadata together, and include the canonical contract SHA-256 and issuer key ID in the signed deployment manifest. The agent and model cannot select issuer keys or sign contracts.

If an existing signed authority/configuration artifact already expresses the exact required contract and release/mission binding, deliver that artifact directly through the same reviewed handoff; a second certificate or second signature is not mandatory and earns no extra integrity credit. The profile digest/version and authority/goal digests preserve provenance across an actual adapter conversion. Compare the complete declared capability population with the upstream approved facts and deployment manifests, test wrong-agent/release/authority/goal binding and attempted capability inflation, then independently read back the published bytes. H-018 verifies the contract facts at the action boundary and still decides each exact request.

AID-H-017.005-G002

Separate contract issuance from readback authority

The signed release manifest supplies the expected contract signer and digest. Separately, the signed deployment trust inventory maps every expected gate instance to its authorized readback key IDs and issuer; the public JWKS is signature-verified before use. A key appearing in the JWKS is not permission to attest for every instance. A contract-issuance key may also attest for a gate only when that dual role is explicitly authorized in the same per-instance inventory. Physical key separation is not a universal requirement.

# File: verify_behavior_contract_handoff.py
from __future__ import annotations
from dataclasses import dataclass
import hashlib, json, math, sqlite3, time
from urllib.parse import urlsplit
from pathlib import Path
from contextlib import closing
import os
import jwt
import requests

REQUIRED_CLAIMS = {"iss", "aud", "sub", "jti", "iat", "nbf", "exp", "nonce",
    "agent_id", "release_id", "agent_release_digest", "cert_id", "contract_sha256",
    "contract_signer_key_id", "policy_version", "contract_state"}

@dataclass(frozen=True)
class ExpectedContract:
    agent_id: str
    release_id: str
    agent_release_digest: str
    cert_id: str
    contract_sha256: str
    contract_signer_key_id: str
    policy_version: str
    readback_issuer: str
    expected_instances: tuple[str, ...]
    instance_readback_key_ids: dict[str, tuple[str, ...]]
    challenge_nonce: str
    maximum_readback_age_seconds: float
    clock_skew_seconds: float
    connect_timeout_seconds: float
    read_timeout_seconds: float
    response_max_bytes: int

def load_jwks(jwks):
    if not isinstance(jwks, dict) or set(jwks) != {"keys"} or not isinstance(jwks["keys"], list): raise ValueError("readback JWKS schema differs")
    keys = {}
    for item in jwks["keys"]:
        if not isinstance(item, dict) or not isinstance(item.get("kid"), str) or not item["kid"] or item["kid"] in keys:
            raise ValueError("missing or duplicate readback kid")
        if item.get("kty") != "EC" or item.get("crv") != "P-256" or item.get("alg") != "ES256" or item.get("use") != "sig" or "d" in item:
            raise ValueError("readback key type, algorithm or use differs")
        keys[item["kid"]] = jwt.PyJWK.from_dict(item)
    if not keys: raise ValueError("readback JWKS is empty")
    return keys

def verify_readback_jwt(token, *, keys, expected):
    header = jwt.get_unverified_header(token)
    if set(header) != {"alg", "typ", "kid"} or header["alg"] != "ES256" or header["typ"] != "JWT": raise jwt.InvalidTokenError("readback JOSE header differs")
    kid = header["kid"]
    if not isinstance(kid, str) or kid not in keys: raise jwt.InvalidKeyError("readback kid unknown")
    claims = jwt.decode(token, keys[kid].key, algorithms=["ES256"], audience="behavior-contract-release",
        issuer=expected.readback_issuer, leeway=expected.clock_skew_seconds, options={"require": sorted(REQUIRED_CLAIMS)})
    if set(claims) != REQUIRED_CLAIMS: raise jwt.InvalidTokenError("readback claims differ")
    instance = claims["sub"]
    if not isinstance(instance, str) or instance not in expected.instance_readback_key_ids:
        raise jwt.InvalidTokenError("unexpected instance")
    if kid not in expected.instance_readback_key_ids[instance]: raise jwt.InvalidKeyError("signer not authorized for this instance")
    if any(type(claims[f]) is not int for f in ("iat", "nbf", "exp")) or not claims["iat"] < claims["exp"] or not claims["nbf"] < claims["exp"]:
        raise jwt.InvalidTokenError("invalid readback time claims")
    if not -expected.clock_skew_seconds <= time.time() - claims["iat"] <= expected.maximum_readback_age_seconds + expected.clock_skew_seconds:
        raise jwt.InvalidTokenError("readback is stale or future-dated")
    if claims["nonce"] != expected.challenge_nonce: raise jwt.InvalidTokenError("readback challenge differs")
    if not isinstance(claims["jti"], str) or not claims["jti"]: raise jwt.InvalidTokenError("missing readback jti")
    return {**claims, "verified_readback_key_id": kid}

def provision_replay_store(database_path):
    """Administrator-only, once per protected persistent volume; never call at startup."""
    path = Path(database_path).absolute()
    fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    os.close(fd)
    try:
        with closing(sqlite3.connect(path.as_uri() + "?mode=rw", uri=True)) as conn, conn:
            conn.execute("PRAGMA synchronous=FULL")
            conn.execute("CREATE TABLE behavior_readback_replay (identity TEXT PRIMARY KEY)")
    except BaseException:
        # Leave failed provisioning for an administrator to inspect; no automatic reset.
        raise

class ReplayGuard:
    def __init__(self, database_path, busy_timeout_seconds):
        if type(busy_timeout_seconds) not in (int, float) or not math.isfinite(busy_timeout_seconds) or busy_timeout_seconds <= 0:
            raise ValueError("positive replay-store deadline required")
        self.database_uri = Path(database_path).absolute().as_uri() + "?mode=rw"
        self.timeout = busy_timeout_seconds
    def _record(self, values):
        # Existing-only opening fails if the volume/database disappeared. The
        # protected store and its backup retain records for the policy window.
        with closing(sqlite3.connect(self.database_uri, uri=True, timeout=self.timeout)) as conn, conn:
            conn.execute("PRAGMA synchronous=FULL")
            conn.execute("BEGIN IMMEDIATE")
            if any(conn.execute("SELECT 1 FROM behavior_readback_replay WHERE identity=?", (value,)).fetchone() for value in values):
                return False
            conn.executemany("INSERT INTO behavior_readback_replay VALUES (?)", [(value,) for value in values])
        return True
    def begin(self, issuer, nonce):
        # Reserve before collection, including incomplete/error results. Retry
        # with a newly generated challenge, never captured partial responses.
        return self._record(["nonce:" + issuer + ":" + nonce])
    def consume(self, issuer, nonce, jtis):
        return self._record(["jti:" + issuer + ":" + jti for jti in jtis])

def verify_handoff(*, control_plane_url, bearer_token, expected, client_cert, ca_bundle,
                   signed_jwks, replay_guard):
    try:
        endpoint = urlsplit(control_plane_url)
        if endpoint.scheme != "https" or not endpoint.hostname or endpoint.username or endpoint.password or endpoint.query or endpoint.fragment:
            raise ValueError("invalid fixed authenticated HTTPS endpoint")
        for value in (expected.agent_id, expected.release_id, expected.cert_id, expected.contract_signer_key_id,
                      expected.policy_version, expected.readback_issuer, expected.challenge_nonce, bearer_token):
            if not isinstance(value, str) or not value: raise ValueError("missing trusted identity or challenge")
        for value in (expected.contract_sha256, expected.agent_release_digest):
            if not isinstance(value, str) or len(value) != 64 or set(value) - set("0123456789abcdef"): raise ValueError("invalid expected digest")
        instances = expected.expected_instances
        if not isinstance(instances, tuple) or not instances or any(not isinstance(x, str) or not x for x in instances) or len(instances) != len(set(instances)):
            raise ValueError("invalid expected instance population")
        if not isinstance(expected.instance_readback_key_ids, dict) or set(expected.instance_readback_key_ids) != set(instances):
            raise ValueError("instance trust population differs")
        for allowed in expected.instance_readback_key_ids.values():
            if not isinstance(allowed, tuple) or not allowed or any(not isinstance(k, str) or not k for k in allowed) or len(allowed) != len(set(allowed)):
                raise ValueError("invalid per-instance signer allowlist")
        for value in (expected.connect_timeout_seconds, expected.read_timeout_seconds, expected.maximum_readback_age_seconds):
            if type(value) not in (int, float) or not math.isfinite(value) or value <= 0: raise ValueError("invalid policy deadline")
        if type(expected.clock_skew_seconds) not in (int, float) or not math.isfinite(expected.clock_skew_seconds) or expected.clock_skew_seconds < 0:
            raise ValueError("invalid policy clock-skew allowance")
        if type(expected.response_max_bytes) is not int or expected.response_max_bytes <= 0: raise ValueError("invalid decoded-byte cap")
        keys = load_jwks(signed_jwks)
        if not set().union(*map(set, expected.instance_readback_key_ids.values())).issubset(keys): raise ValueError("missing authorized readback key")
        if not replay_guard.begin(expected.readback_issuer, expected.challenge_nonce):
            return {"status": "FAIL", "reason": "readback_replay"}
        with requests.get(control_plane_url.rstrip("/") + "/v1/behavior-contract-readback",
            params={"agent_id": expected.agent_id, "release_id": expected.release_id, "nonce": expected.challenge_nonce},
            headers={"Authorization": "Bearer " + bearer_token, "Accept": "application/json"}, cert=client_cert,
            verify=ca_bundle, timeout=(expected.connect_timeout_seconds, expected.read_timeout_seconds),
            allow_redirects=False, stream=True) as response:
            response.raise_for_status()
            if response.status_code != 200: raise ValueError("readback did not return HTTP 200")
            raw = bytearray()
            for chunk in response.iter_content(chunk_size=min(65536, expected.response_max_bytes + 1)):
                raw.extend(chunk)
                if len(raw) > expected.response_max_bytes: raise ValueError("readback exceeds decoded-byte cap")
        body = json.loads(bytes(raw))
        if not isinstance(body, dict) or set(body) != {"readback_jwts"} or not isinstance(body["readback_jwts"], list): raise ValueError("response schema differs")
    except (requests.RequestException, jwt.PyJWTError, KeyError, TypeError, ValueError, sqlite3.Error, OSError) as exc:
        return {"status": "ERROR", "reason": type(exc).__name__}
    observed = {}; seen_jtis = set(); invalid_tokens = []; mismatches = []
    fields = {"agent_id": expected.agent_id, "release_id": expected.release_id,
        "agent_release_digest": expected.agent_release_digest, "cert_id": expected.cert_id,
        "contract_sha256": expected.contract_sha256, "contract_signer_key_id": expected.contract_signer_key_id,
        "policy_version": expected.policy_version, "contract_state": "ACTIVE"}
    for index, token in enumerate(body["readback_jwts"]):
        try:
            if not isinstance(token, str) or not token: raise jwt.InvalidTokenError("empty readback")
            claims = verify_readback_jwt(token, keys=keys, expected=expected)
            instance = claims["sub"]
            if instance in observed or claims["jti"] in seen_jtis: raise jwt.InvalidTokenError("duplicate instance or readback jti")
            seen_jtis.add(claims["jti"]); observed[instance] = claims
            different = sorted(f for f, value in fields.items() if claims[f] != value)
            if different: mismatches.append({"instance_id": instance, "fields": different})
        except (jwt.PyJWTError, KeyError, TypeError, ValueError) as exc:
            invalid_tokens.append({"index": index, "reason": type(exc).__name__})
    missing = sorted(set(instances) - observed.keys())
    if invalid_tokens or mismatches: status, reason = "FAIL", "invalid_or_mismatched_signed_readback"
    elif missing: status, reason = "INSUFFICIENT_DATA", "missing_gate_readback"
    else:
        try:
            accepted = replay_guard.consume(expected.readback_issuer, expected.challenge_nonce, sorted(seen_jtis))
            status, reason = ("PASS", "complete_exact_signed_handoff") if accepted else ("FAIL", "readback_replay")
        except (sqlite3.Error, OSError): status, reason = "ERROR", "replay_store_unavailable"
    return {"status": status, "reason": reason, "checked_instances": sorted(observed),
            "missing_instances": missing, "invalid_tokens": invalid_tokens, "mismatches": mismatches,
            "verified_readback_key_ids": {instance: row["verified_readback_key_id"] for instance, row in observed.items()}}

Deploy and verify the handoff

Generate a fresh unpredictable challenge nonce for each reconciliation. Obtain the fixed authenticated HTTPS endpoint, exact instance population, per-instance signer allowlists, public JWKS, finite request/readback/replay deadlines and decoded-byte cap from the verified deployment trust policy. Use workload identity/mTLS for collection and a protected persistent replay database that survives restarts. Do not infer authenticity from a returned signature_verified field, and never compare the gate JWT key ID directly with the contract issuer key ID.

Run after deploy, scale-out, contract/key rotation, revocation and rollback. Preserve token digests rather than bearer JWTs, release/trust/JWKS digests, expected/observed populations, verified gate signer identities, nonce/replay disposition and the exact contract/policy bindings. A missing gate remains insufficient evidence; forged or misbound signers, stale/replayed challenges, duplicate/extra instances, wrong release/digest/issuer or revoked/non-active contracts fail. Collection or trust/replay-store errors are errors, never a pass. Independently exercise valid separate signers, unauthorized cross-instance signing, an unapproved contract key used as a gate key, expressly authorized dual-role keys, token replay after restart, missing replicas and bounded collection failure. H-018 independently proves action-time enforcement.

Set clock_skew_seconds from the verified deployment policy; apply the same bounded allowance to JWT validity and readback age. Provision the protected replay database once through provision_replay_store before service activation. Runtime opening requires the existing database and schema. A missing volume or store is an error, never permission to reset replay history. Reserve each fresh challenge before collecting readbacks; an incomplete collection or transport error consumes that challenge, so retry with a new nonce.

Open-source tools

• pyca/cryptography
• PyJWT
• rfc8785 (Python canonical JSON implementation)

Commercial / hosted tools

• AWS Key Management Service (AWS KMS)
• Azure Key Vault
• Google Cloud Key Management Service
• Cerbos Hub
• Permit.io

Threat mappings

MITRE ATLAS

• AML.T0081 Modify AI Agent Configuration (detect tampered behavior-contract content or substitution during signed agent/release-bound delivery to authorization gates; does not prove safe policy semantics or action authorization)

MAESTRO

• N/A

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-5.2.1 Agent Profile Tampering (signed agent behavior-contract integrity; runtime authorization remains separate)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• N/A