Scope boundary

Owns verification of Agent Cards, discovery records, registry and name-resolution provenance, endpoint and protocol selection, advertised capabilities, TTL, and revocation before an agent route is trusted.

AID-H-004.003: AID-H-004.004 proves which endpoint and advertised capability record may be used; AID-H-004.003 secures the actual message, stream, task, and authorization lifecycle after connection. A verified Agent Card does not authorize every later request.

AID-H-024: AID-H-004.004 verifies agent-level discovery and endpoint provenance; AID-H-024 resolves and pins tool or MCP names, versions, descriptors, and contracts. Agent identity discovery and tool-resolution integrity operate at different namespaces.

Implementation guidance

• AID-H-004.004-G001: Verify signed A2A Agent Cards before discovery, routing, capability selection, or endpoint use.
• AID-H-004.004-G002: Secure ANS Resolution Verification & TTL Management
• AID-H-004.004-G003: Mandatory Revocation Check (CRL) for Agent Identities
• AID-H-004.004-G004: Verify exact capability-manifest bytes against a signed discovery record and issue a signed routing decision.

AID-H-004.004-G001

Production implementation: verify an A2A v1.0 Agent Card with the official SDK

Pin an a2a-sdk release that declares A2A v1.0 compatibility in the workload lockfile and record the package and image digests. The trusted resolver passes the exact TLS-verified response bytes and the configured card URL into this verifier; model output and card fields never select trust roots or routes.

Signed trust-policy contract

Store a root-owned, signature-verified policy with exact fields schema_version, policy_version, not_after, allowed_algorithms, keys_by_kid, and routes. Each exact card URL route binds approved signing keys, endpoint origins, A2A bindings and versions, agent versions, skill IDs, extension URIs, capability flags, and a card-size ceiling. Pin the policy digest in deployment configuration so an old but validly signed policy cannot be replayed.

# File: a2a/verify_agent_card_v1.py
from __future__ import annotations
import math

import hashlib
import json
import os
import stat
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

from a2a.client.card_resolver import parse_agent_card
from a2a.utils.signing import create_signature_verifier
from jwt import PyJWK

def _required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number of seconds")
    return value

AGENT_CARD_POLICY_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H004_AGENT_CARD_POLICY_VERIFY_TIMEOUT_SECONDS")



class AgentCardRejected(RuntimeError):
    pass


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise AgentCardRejected(f"duplicate JSON member: {key}")
        result[key] = value
    return result


def decode_unique_json(raw: bytes) -> dict[str, Any]:
    try:
        value = json.loads(raw, object_pairs_hook=unique_object, parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise AgentCardRejected("invalid JSON") from exc
    if not isinstance(value, dict):
        raise AgentCardRejected("JSON root must be an object")
    return value


def require_root_owned_regular(path: Path) -> None:
    metadata = path.lstat()
    if path.is_symlink() or not stat.S_ISREG(metadata.st_mode):
        raise AgentCardRejected(f"non-regular trust input: {path}")
    if metadata.st_uid != 0 or metadata.st_mode & 0o022:
        raise AgentCardRejected(f"trust input is not root-owned and write-protected: {path}")


def verified_policy_bytes(path: Path, signature: Path, verify_key: Path) -> bytes:
    for item in (path, signature, verify_key):
        require_root_owned_regular(item)
    policy_bytes = path.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(prefix="a2a-trust-policy-", suffix=".json", delete=False) as handle:
            handle.write(policy_bytes)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [
                "cosign", "verify-blob",
                "--key", str(verify_key),
                "--bundle", str(signature),
                str(snapshot),
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=AGENT_CARD_POLICY_VERIFY_TIMEOUT_SECONDS,
        )
        return policy_bytes
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)


def utc_time(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise AgentCardRejected("not_after must include a UTC offset")
    return parsed.astimezone(timezone.utc)


def load_trust_policy() -> tuple[dict[str, Any], str]:
    raw = verified_policy_bytes(
        Path(os.environ["A2A_TRUST_POLICY"]),
        Path(os.environ["A2A_TRUST_POLICY_SIGNATURE"]),
        Path(os.environ["A2A_TRUST_POLICY_VERIFY_KEY"]),
    )
    digest = hashlib.sha256(raw).hexdigest()
    if digest != os.environ["A2A_TRUST_POLICY_SHA256"].lower():
        raise AgentCardRejected("trust policy digest is not the deployment-pinned digest")
    policy = decode_unique_json(raw)
    required = {
        "schema_version", "policy_version", "not_after",
        "allowed_algorithms", "keys_by_kid", "routes",
    }
    if set(policy) != required:
        raise AgentCardRejected("trust policy schema differs")
    if policy["schema_version"] != "aidefend.a2a-card-trust-policy.v1":
        raise AgentCardRejected("unsupported trust policy schema")
    if utc_time(policy["not_after"]) <= datetime.now(timezone.utc):
        raise AgentCardRejected("trust policy expired")
    algorithms = policy["allowed_algorithms"]
    supported = {"ES256", "PS256", "RS256", "EdDSA"}
    if (
        not isinstance(algorithms, list)
        or not algorithms
        or len(algorithms) != len(set(algorithms))
        or not set(algorithms) <= supported
    ):
        raise AgentCardRejected("invalid JWS algorithm policy")
    if not isinstance(policy["keys_by_kid"], dict) or not policy["keys_by_kid"]:
        raise AgentCardRejected("empty signing-key policy")
    if not isinstance(policy["routes"], dict) or not policy["routes"]:
        raise AgentCardRejected("empty route policy")
    return policy, digest


def https_origin(value: str) -> str:
    try:
        parsed = urlparse(value)
        port = parsed.port
    except ValueError as exc:
        raise AgentCardRejected("invalid endpoint URL") from exc
    if (
        parsed.scheme != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.fragment
    ):
        raise AgentCardRejected("endpoint must be an absolute HTTPS URL without userinfo or fragment")
    host = parsed.hostname.encode("idna").decode("ascii").lower()
    suffix = "" if port in (None, 443) else f":{port}"
    return f"https://{host}{suffix}"


def verify_agent_card(
    card_bytes: bytes,
    *,
    trusted_card_url: str,
    client_protocol_bindings: set[str],
) -> dict[str, Any]:
    policy, policy_sha256 = load_trust_policy()
    route = policy["routes"].get(trusted_card_url)
    if not isinstance(route, dict):
        raise AgentCardRejected("card URL has no exact trusted route")
    https_origin(trusted_card_url)
    route_fields = {
        "allowed_kids", "allowed_endpoint_origins",
        "allowed_protocol_bindings", "allowed_protocol_versions",
        "allowed_agent_versions", "allowed_skill_ids",
        "allowed_extension_uris", "allowed_capabilities", "max_card_bytes",
    }
    if set(route) != route_fields:
        raise AgentCardRejected("route policy schema differs")
    if not client_protocol_bindings:
        raise AgentCardRejected("client supports no protocol binding")
    if (
        not isinstance(route["max_card_bytes"], int)
        or isinstance(route["max_card_bytes"], bool)
        or route["max_card_bytes"] <= 0
        or len(card_bytes) > route["max_card_bytes"]
    ):
        raise AgentCardRejected("Agent Card exceeds the signed size limit")

    raw_card = decode_unique_json(card_bytes)
    card = parse_agent_card(raw_card)
    if not card.signatures:
        raise AgentCardRejected("signed Agent Card required")

    allowed_kids = set(route["allowed_kids"])
    resolved_kids: set[str] = set()

    def key_provider(kid: str | None, jku: str | None) -> PyJWK:
        if not kid or kid not in allowed_kids:
            raise AgentCardRejected("signature kid is not approved for this route")
        record = policy["keys_by_kid"].get(kid)
        if not isinstance(record, dict) or set(record) != {"jwk", "jku", "revoked"}:
            raise AgentCardRejected("signing-key record schema differs")
        if record["revoked"] is not False:
            raise AgentCardRejected("signing key is revoked")
        if jku is not None and jku != record["jku"]:
            raise AgentCardRejected("protected jku is not the pinned JWKS URL")
        jwk_data = record["jwk"]
        private_members = {"d", "p", "q", "dp", "dq", "qi", "oth", "k"}
        if (
            not isinstance(jwk_data, dict)
            or jwk_data.get("kty") not in {"EC", "OKP", "RSA"}
            or private_members & set(jwk_data)
        ):
            raise AgentCardRejected("trust policy must contain a public asymmetric JWK")
        resolved_kids.add(kid)
        return PyJWK.from_dict(jwk_data)

    verifier = create_signature_verifier(
        key_provider,
        algorithms=list(policy["allowed_algorithms"]),
    )
    verifier(card)

    allowed_origins = set(route["allowed_endpoint_origins"])
    allowed_bindings = set(route["allowed_protocol_bindings"])
    allowed_versions = set(route["allowed_protocol_versions"])
    interfaces: list[dict[str, Any]] = []
    seen_interfaces: set[tuple[str, str, str, str]] = set()
    for interface in card.supported_interfaces:
        row = (
            interface.url,
            interface.protocol_binding,
            interface.protocol_version,
            interface.tenant,
        )
        if row in seen_interfaces:
            raise AgentCardRejected("duplicate supportedInterfaces entry")
        seen_interfaces.add(row)
        if https_origin(interface.url) not in allowed_origins:
            raise AgentCardRejected("interface origin is not approved")
        if interface.protocol_binding not in allowed_bindings:
            raise AgentCardRejected("protocol binding is not approved")
        if interface.protocol_version not in allowed_versions:
            raise AgentCardRejected("protocol version is not approved")
        interfaces.append(
            {
                "url": interface.url,
                "protocol_binding": interface.protocol_binding,
                "protocol_version": interface.protocol_version,
                "tenant": interface.tenant or None,
            }
        )
    if not interfaces:
        raise AgentCardRejected("Agent Card has no supported interface")

    selected = next(
        (
            item for item in interfaces
            if item["protocol_binding"] in client_protocol_bindings
        ),
        None,
    )
    if selected is None:
        raise AgentCardRejected("no mutually supported approved interface")

    if card.version not in set(route["allowed_agent_versions"]):
        raise AgentCardRejected("agent version is not approved")
    skill_ids = [skill.id for skill in card.skills]
    if (
        not skill_ids
        or len(skill_ids) != len(set(skill_ids))
        or set(skill_ids) != set(route["allowed_skill_ids"])
    ):
        raise AgentCardRejected("skill population differs from the approved route")
    extension_uris = [item.uri for item in card.capabilities.extensions]
    if (
        len(extension_uris) != len(set(extension_uris))
        or set(extension_uris) != set(route["allowed_extension_uris"])
    ):
        raise AgentCardRejected("extension population differs from the approved route")
    effective_capabilities = {
        "streaming": bool(card.capabilities.streaming),
        "push_notifications": bool(card.capabilities.push_notifications),
        "extended_agent_card": bool(card.capabilities.extended_agent_card),
    }
    if effective_capabilities != route["allowed_capabilities"]:
        raise AgentCardRejected("capability flags differ from the approved route")

    return {
        "schema_version": "aidefend.verified-a2a-route.v1",
        "card_url": trusted_card_url,
        "card_sha256": hashlib.sha256(card_bytes).hexdigest(),
        "trust_policy_version": policy["policy_version"],
        "trust_policy_sha256": policy_sha256,
        "agent_name": card.name,
        "agent_version": card.version,
        "resolved_allowed_kids": sorted(resolved_kids),
        "signature_count": len(card.signatures),
        "supported_interfaces": interfaces,
        "selected_interface": selected,
        "skill_ids": sorted(skill_ids),
        "extension_uris": sorted(extension_uris),
        "capabilities": effective_capabilities,
    }


def run_with_status(**kwargs) -> dict:
    try:
        return {"status": "PASS", "evidence": verify_agent_card(**kwargs)}
    except AgentCardRejected as exc:
        return {"status": "FAIL", "reason": str(exc)}
    except Exception as exc:
        return {"status": "ERROR", "reason": type(exc).__name__}

Enforcement: Call verify_agent_card before discovery, route selection, capability use, or client construction. It delegates schema-aware A2A v1.0 canonicalization and JWS verification to the official SDK, requires a locally pinned public key, validates every supportedInterfaces member, and chooses the first mutually supported approved binding in signed order. Preserve the returned route receipt, exact card and policy bytes, TLS/registry fetch evidence, SDK and verifier digests, and independent re-fetch result. Sign the canonical receipt outside the agent runtime and retain it in append-only or WORM storage. A card signature authenticates discovery metadata only; request authentication, tenant binding, delegation, and per-action authorization remain mandatory.

AID-H-004.004-G002

Concept:

When an agent resolver trusts ANS-style discovery, it must verify the registry signature over the exact endpoint record bytes before using the endpoint, then cache the result only until the shorter of the signed expiry and the signed TTL. Missing registry data, invalid signatures, malformed or oversized responses, stale records, redirects, or non-canonical HTTPS endpoints are routing denials. Load the exact registry origin, positive deadline/body/TTL caps, key ID, and trust key from a signature-verified discovery profile.

Production resolver pattern

# File: discovery/ans_resolver.py
from __future__ import annotations
import math
import os

import base64
import json
import re
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Mapping
from urllib.parse import quote, urlparse

import requests
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

def _required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number of seconds")
    return value

ANS_REGISTRY_FETCH_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H004_ANS_REGISTRY_FETCH_TIMEOUT_SECONDS")


def _required_positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be a positive integer")
    return value


ANS_RECORD_MAX_TTL_SECONDS = _required_positive_int("VERIFIED_H004_ANS_RECORD_MAX_TTL_SECONDS")
ANS_REGISTRY_RESPONSE_MAX_BYTES = _required_positive_int("VERIFIED_H004_ANS_REGISTRY_RESPONSE_MAX_BYTES")



class DiscoveryValidationError(RuntimeError):
    """Discovery record failed security validation."""


@dataclass(frozen=True)
class CacheEntry:
    record: dict
    expires_at_epoch: float


class ANSResolver:
    def __init__(
        self,
        *,
        registry_url: str,
        registry_public_key_pem: bytes,
        allowed_key_id: str,
        http_client: requests.Session | None = None,
    ):
        parsed = urlparse(registry_url)
        if (
            parsed.scheme != "https" or not parsed.hostname
            or parsed.username is not None or parsed.password is not None
            or parsed.query or parsed.fragment or parsed.path not in {"", "/"}
            or registry_url.endswith("/") or parsed.netloc != parsed.netloc.lower()
        ):
            raise ValueError("ANS registry URL must be the exact canonical HTTPS origin")
        self.registry_url = registry_url
        self.allowed_key_id = allowed_key_id
        self.registry_public_key = serialization.load_pem_public_key(registry_public_key_pem)
        self.http = http_client or requests.Session()
        self.cache: dict[str, CacheEntry] = {}

    def resolve_agent(self, ans_name: str) -> dict:
        if not re.fullmatch(r"[a-z0-9][a-z0-9.-]{1,126}", ans_name):
            raise DiscoveryValidationError("invalid ANS name syntax")

        cached = self.cache.get(ans_name)
        now = time.time()
        if cached and cached.expires_at_epoch > now:
            return cached.record
        if cached:
            del self.cache[ans_name]

        envelope = self._fetch_signed_record(ans_name)
        raw_data = envelope["data"]
        signature = base64.b64decode(envelope["signature_b64"], validate=True)
        if envelope["key_id"] != self.allowed_key_id:
            raise DiscoveryValidationError("ANS registry signing key is not approved")

        try:
            self.registry_public_key.verify(
                signature,
                raw_data.encode("utf-8"),
                padding.PSS(
                    mgf=padding.MGF1(hashes.SHA256()),
                    salt_length=padding.PSS.MAX_LENGTH,
                ),
                hashes.SHA256(),
            )
        except InvalidSignature as exc:
            raise DiscoveryValidationError("ANS record signature verification failed") from exc

        record = self._validate_record(ans_name, raw_data)
        ttl_seconds = int(record["ttl_seconds"])
        signed_expiry = self._parse_utc(record["expires_at"]).timestamp()
        self.cache[ans_name] = CacheEntry(
            record=record,
            expires_at_epoch=min(now + ttl_seconds, signed_expiry),
        )
        return record

    def _fetch_signed_record(self, name: str) -> Mapping[str, str]:
        with self.http.get(
            f"{self.registry_url}/v1/records/{quote(name, safe='')}",
            headers={"Accept": "application/json"},
            timeout=ANS_REGISTRY_FETCH_TIMEOUT_SECONDS,
            allow_redirects=False,
            stream=True,
        ) as response:
            if response.status_code != 200:
                raise DiscoveryValidationError(f"ANS registry returned HTTP {response.status_code}")
            media = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
            if media != "application/json":
                raise DiscoveryValidationError("ANS registry returned an unexpected media type")
            raw = bytearray()
            for chunk in response.iter_content(chunk_size=min(65536, ANS_REGISTRY_RESPONSE_MAX_BYTES + 1)):
                if chunk:
                    raw.extend(chunk)
                if len(raw) > ANS_REGISTRY_RESPONSE_MAX_BYTES:
                    raise DiscoveryValidationError("ANS response exceeds the signed decoded-byte cap")
        envelope = json.loads(bytes(raw), object_pairs_hook=self._no_duplicates, parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")))
        required = {"data", "signature_b64", "key_id"}
        if set(envelope) != required or not all(isinstance(envelope[item], str) for item in required):
            raise DiscoveryValidationError("ANS registry response schema differs")
        return envelope

    def _validate_record(self, ans_name: str, raw_data: str) -> dict:
        try:
            record = json.loads(raw_data, object_pairs_hook=self._no_duplicates, parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")))
        except json.JSONDecodeError as exc:
            raise DiscoveryValidationError("ANS record data is not valid JSON") from exc

        required = {
            "ans_name",
            "agent_id",
            "endpoint",
            "protocol_version",
            "capability_manifest_sha256",
            "ttl_seconds",
            "issued_at",
            "expires_at",
        }
        if set(record) != required:
            raise DiscoveryValidationError("ANS record schema differs")
        if record["ans_name"] != ans_name:
            raise DiscoveryValidationError("ANS record name binding differs")
        endpoint = urlparse(record["endpoint"])
        if (
            endpoint.scheme != "https" or not endpoint.hostname
            or endpoint.username is not None or endpoint.password is not None
            or endpoint.query or endpoint.fragment or not endpoint.path.startswith("/")
            or endpoint.netloc != endpoint.netloc.lower()
        ):
            raise DiscoveryValidationError("ANS endpoint must be a canonical HTTPS URL")
        if not re.fullmatch(r"[a-f0-9]{64}", record["capability_manifest_sha256"]):
            raise DiscoveryValidationError("capability manifest hash is not sha256 hex")
        ttl_seconds = int(record["ttl_seconds"])
        if ttl_seconds <= 0 or ttl_seconds > ANS_RECORD_MAX_TTL_SECONDS:
            raise DiscoveryValidationError("ANS TTL is outside approved bounds")
        if self._parse_utc(record["expires_at"]) <= datetime.now(timezone.utc):
            raise DiscoveryValidationError("ANS record has expired")
        return record

    @staticmethod
    def _no_duplicates(pairs: list[tuple[str, object]]) -> dict:
        result = {}
        for key, value in pairs:
            if key in result:
                raise DiscoveryValidationError(f"duplicate JSON key: {key}")
            result[key] = value
        return result

    @staticmethod
    def _parse_utc(value: str) -> datetime:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        if parsed.tzinfo is None:
            raise DiscoveryValidationError("timestamp must include timezone")
        return parsed.astimezone(timezone.utc)

Action: Treat discovery as a signed routing input. Verify the signature, key ID, schema, name binding, endpoint scheme, manifest hash, TTL, and expiry before writing a route decision. If the registry is unreachable or the record cannot be verified, deny routing rather than falling back to stale or unsigned discovery metadata.

AID-H-004.004-G003

Before you begin

Apply AID-H-004.004-G003 only when an agent endpoint, Agent Card, registry record, ANS-style name, capability manifest, or revocation state is discovered dynamically.

Run revocation validation after certificate-path validation and before a discovered endpoint, Agent Card, registry record, or agent identity is admitted. Fetch the CRL only through an authenticated, allowlisted channel; verify the CRL issuer, signature, scope, freshness, key usage, and monotonic CRL number before checking the leaf serial. A parseable list or absent serial alone is never sufficient.

# File: identity/verified_crl.py
from __future__ import annotations
import math

import hashlib
import json
import os
import re
import subprocess
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path

from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.x509.oid import ExtensionOID

def _required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number of seconds")
    return value

REVOCATION_DECISION_SIGN_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H004_REVOCATION_DECISION_SIGN_TIMEOUT_SECONDS")
REVOCATION_DECISION_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H004_REVOCATION_DECISION_VERIFY_TIMEOUT_SECONDS")



@dataclass(frozen=True)
class CrlPolicy:
    policy_version: str
    expected_distribution_point: str
    maximum_crl_age: timedelta
    maximum_clock_skew: timedelta
    require_next_update: bool = True


@dataclass(frozen=True)
class RevocationDecision:
    schema_version: str
    control_id: str
    outcome: str
    reason_code: str
    certificate_serial_hex: str
    issuer_sha256: str
    crl_sha256: str
    distribution_point: str
    crl_number: int
    crl_last_update: str
    crl_next_update: str | None
    policy_version: str
    policy_sha256: str
    decided_at: str


def required_extension(certificate, oid):
    try:
        return certificate.extensions.get_extension_for_oid(oid).value
    except x509.ExtensionNotFound as exc:
        raise PermissionError(f"required_extension_missing:{oid.dotted_string}") from exc


def verify_crl(
    *,
    agent_cert_pem: bytes,
    issuer_cert_pem: bytes,
    crl_der: bytes,
    source_distribution_point: str,
    policy: CrlPolicy,
    policy_sha256: str,
    state_store,
    now: datetime | None = None,
) -> RevocationDecision:
    now = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
    if (
        not policy.policy_version
        or not policy.expected_distribution_point.startswith("https://")
        or policy.maximum_crl_age <= timedelta(0)
        or policy.maximum_clock_skew < timedelta(0)
        or not re.fullmatch(r"[a-f0-9]{64}", policy_sha256)
    ):
        raise ValueError("signed_crl_policy_is_invalid")
    cert = x509.load_pem_x509_certificate(agent_cert_pem)
    issuer = x509.load_pem_x509_certificate(issuer_cert_pem)
    crl = x509.load_der_x509_crl(crl_der)

    # This verifies the leaf's direct issuer/signature. Production path building
    # must separately validate the issuer to a pinned trust anchor.
    cert.verify_directly_issued_by(issuer)
    if crl.issuer != issuer.subject or not crl.is_signature_valid(issuer.public_key()):
        raise PermissionError("crl_issuer_or_signature_invalid")
    if source_distribution_point != policy.expected_distribution_point:
        raise PermissionError("crl_distribution_point_mismatch")

    constraints = required_extension(issuer, ExtensionOID.BASIC_CONSTRAINTS)
    usage = required_extension(issuer, ExtensionOID.KEY_USAGE)
    if not constraints.ca or not usage.crl_sign:
        raise PermissionError("issuer_not_authorized_to_sign_crl")

    try:
        delta = crl.extensions.get_extension_for_oid(
            ExtensionOID.DELTA_CRL_INDICATOR
        )
    except x509.ExtensionNotFound:
        delta = None
    if delta is not None:
        raise PermissionError("delta_crl_requires_explicit_merge_support")

    try:
        idp = crl.extensions.get_extension_for_oid(
            ExtensionOID.ISSUING_DISTRIBUTION_POINT
        ).value
    except x509.ExtensionNotFound:
        idp = None
    if idp is not None:
        if (
            idp.relative_name is not None
            or idp.only_contains_user_certs
            or idp.only_contains_ca_certs
            or idp.only_contains_attribute_certs
            or idp.only_some_reasons is not None
            or idp.indirect_crl
        ):
            raise PermissionError("partitioned_or_indirect_crl_is_unsupported")
        idp_uris = {
            name.value for name in (idp.full_name or [])
            if isinstance(name, x509.UniformResourceIdentifier)
        }
        if idp_uris and idp_uris != {policy.expected_distribution_point}:
            raise PermissionError("issuing_distribution_point_scope_mismatch")

    number = required_extension(crl, ExtensionOID.CRL_NUMBER).crl_number
    last_update = crl.last_update_utc
    next_update = crl.next_update_utc
    if last_update > now + policy.maximum_clock_skew:
        raise PermissionError("crl_last_update_in_future")
    if now - last_update > policy.maximum_crl_age:
        raise PermissionError("crl_exceeds_policy_age")
    if policy.require_next_update and next_update is None:
        raise PermissionError("crl_next_update_missing")
    if next_update is not None and now - policy.maximum_clock_skew >= next_update:
        raise PermissionError("crl_expired")

    issuer_key = issuer.fingerprint(hashes.SHA256()).hex()
    crl_sha256 = hashlib.sha256(crl_der).hexdigest()
    current_state = {"crl_number": number, "crl_sha256": crl_sha256}
    previous = state_store.read_crl_state(issuer_key, source_distribution_point)
    if previous is not None and number < previous["crl_number"]:
        raise PermissionError("crl_number_rollback")
    if (
        previous is not None
        and number == previous["crl_number"]
        and crl_sha256 != previous["crl_sha256"]
    ):
        raise PermissionError("same_number_different_crl_bytes")
    if previous != current_state:
        if not state_store.compare_and_set_crl_state(
            issuer_key, source_distribution_point, previous, current_state
        ):
            raise RuntimeError("crl_state_concurrent_update_or_unavailable")

    revoked = crl.get_revoked_certificate_by_serial_number(cert.serial_number)
    return RevocationDecision(
        schema_version="aidefend.h004004-revocation-decision.v2",
        control_id="AID-H-004.004",
        outcome="FAIL" if revoked is not None else "PASS",
        reason_code="certificate_revoked" if revoked is not None else "not_listed_in_verified_crl",
        certificate_serial_hex=f"{cert.serial_number:x}",
        issuer_sha256=issuer_key,
        crl_sha256=crl_sha256,
        distribution_point=source_distribution_point,
        crl_number=number,
        crl_last_update=last_update.isoformat(),
        crl_next_update=next_update.isoformat() if next_update else None,
        policy_version=policy.policy_version,
        policy_sha256=policy_sha256,
        decided_at=now.isoformat(),
    )


def evaluate_sign_and_read_back(*, decision_path: Path, **verify_args) -> dict:
    policy = verify_args["policy"]
    policy_sha256 = verify_args["policy_sha256"]
    try:
        decision = asdict(verify_crl(**verify_args))
    except Exception as exc:
        decision = {
            "schema_version": "aidefend.h004004-revocation-decision.v2",
            "control_id": "AID-H-004.004",
            "outcome": "ERROR",
            "reason_code": type(exc).__name__,
            "certificate_input_sha256": hashlib.sha256(
                verify_args["agent_cert_pem"]
            ).hexdigest(),
            "issuer_input_sha256": hashlib.sha256(
                verify_args["issuer_cert_pem"]
            ).hexdigest(),
            "crl_sha256": hashlib.sha256(verify_args["crl_der"]).hexdigest(),
            "distribution_point": verify_args["source_distribution_point"],
            "policy_version": policy.policy_version,
            "policy_sha256": policy_sha256,
            "decided_at": datetime.now(timezone.utc).isoformat(),
        }

    payload = json.dumps(
        decision, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8") + b"\n"
    decision_path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(
        decision_path,
        os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC,
        0o600,
    )
    try:
        os.write(fd, payload)
        os.fsync(fd)
    finally:
        os.close(fd)
    signature_path = decision_path.with_suffix(".sig")
    subprocess.run([
        "cosign", "sign-blob", "--yes",
        "--key", "env://H004004_DECISION_SIGNING_KEY",
        "--bundle", str(signature_path), str(decision_path),
    ], check=True, timeout=REVOCATION_DECISION_SIGN_TIMEOUT_SECONDS)
    subprocess.run([
        "cosign", "verify-blob",
        "--key", os.environ["H004004_DECISION_VERIFY_KEY"],
        "--bundle", str(signature_path), str(decision_path),
    ], check=True, capture_output=True, text=True, timeout=REVOCATION_DECISION_VERIFY_TIMEOUT_SECONDS)
    if decision_path.read_bytes() != payload:
        raise RuntimeError("signed_revocation_decision_readback_mismatch")
    return decision

Verification, outcomes, and evidence

Evidence: retain exact signed policy, certificate/issuer fingerprints and serial, trust-anchor and chain-validation receipt, CRL bytes/digest/signature algorithm/IDP, distribution point, number, thisUpdate/nextUpdate, prior and resulting atomic state, fetch receipt, signed decision/readback, and verifier build digest. A separate verifier reloads the retained bytes and reproduces signature, scope, freshness, monotonicity, serial lookup, and complete admitted-identity population.

AID-H-004.004-G004

Complete admission unit

The routing gateway verifies one signed discovery record and one signed policy, hashes the exact manifest bytes before parsing, validates strict schemas and validity, then signs a decision that binds the agent, registry, endpoint, manifest, record, policy, and verifier build. The gateway routes only a signed PASS; a manifest hash match by itself is not authorization.

# File: discovery/verify_capability_manifest.py
from __future__ import annotations
import math

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

def _required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number of seconds")
    return value

CAPABILITY_ARTIFACT_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H004_CAPABILITY_ARTIFACT_VERIFY_TIMEOUT_SECONDS")
CAPABILITY_DECISION_SIGN_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H004_CAPABILITY_DECISION_SIGN_TIMEOUT_SECONDS")
CAPABILITY_DECISION_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H004_CAPABILITY_DECISION_VERIFY_TIMEOUT_SECONDS")



SHA256 = re.compile(r"^[a-f0-9]{64}$")
POLICY_FIELDS = {
    "schema_version", "policy_version", "registry_id",
    "maximum_record_seconds", "maximum_record_bytes", "maximum_manifest_bytes",
    "required_manifest_schema", "verifier_image_digest",
}
RECORD_FIELDS = {
    "schema_version", "record_id", "record_version", "registry_id",
    "agent_id", "endpoint", "capability_manifest_sha256",
    "capability_manifest_version", "policy_version", "issued_at",
    "expires_at", "signing_key_id",
}
MANIFEST_FIELDS = {
    "schema_version", "agent_id", "manifest_version", "capabilities",
    "tool_scopes", "data_scopes", "network_destinations",
}


def canonical(value: dict) -> bytes:
    return json.dumps(
        value, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8") + b"\n"


def parse_time(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.utcoffset() is None:
        raise ValueError("timestamp_without_timezone")
    return parsed.astimezone(timezone.utc)


def verified_bytes(path: Path, signature: Path, key: str) -> bytes:
    before = path.read_bytes()
    subprocess.run([
        "cosign", "verify-blob", "--key", key,
        "--bundle", str(signature), str(path),
    ], check=True, capture_output=True, text=True, timeout=CAPABILITY_ARTIFACT_VERIFY_TIMEOUT_SECONDS)
    after = path.read_bytes()
    if not before or not hmac.compare_digest(before, after):
        raise RuntimeError("signed_artifact_changed_during_verification")
    return before


def strict_object(raw: bytes, fields: set[str], label: str) -> dict:
    value = json.loads(raw.decode("utf-8", errors="strict"))
    if not isinstance(value, dict) or set(value) != fields:
        raise ValueError(f"{label}_schema_mismatch")
    return value


def evaluate(
    *,
    policy_path: Path,
    policy_signature: Path,
    record_path: Path,
    record_signature: Path,
    manifest_path: Path,
    expected_agent_id: str,
    expected_endpoint: str,
    output: Path,
) -> dict:
    policy_raw = verified_bytes(
        policy_path, policy_signature, os.environ["DISCOVERY_POLICY_VERIFY_KEY"]
    )
    policy_sha256 = hashlib.sha256(policy_raw).hexdigest()
    policy = strict_object(policy_raw, POLICY_FIELDS, "policy")
    record_raw = b""
    manifest_raw = b""
    now = datetime.now(timezone.utc)
    try:
        if (
            policy["schema_version"] != "aidefend.discovery-policy.v1"
            or not isinstance(policy["policy_version"], str)
            or not policy["policy_version"]
            or not isinstance(policy["maximum_record_seconds"], int)
            or isinstance(policy["maximum_record_seconds"], bool)
            or policy["maximum_record_seconds"] < 1
            or not isinstance(policy["maximum_record_bytes"], int)
            or isinstance(policy["maximum_record_bytes"], bool)
            or policy["maximum_record_bytes"] < 1
            or not isinstance(policy["maximum_manifest_bytes"], int)
            or isinstance(policy["maximum_manifest_bytes"], bool)
            or policy["maximum_manifest_bytes"] < 1
            or not isinstance(policy["verifier_image_digest"], str)
            or not re.fullmatch(r".+@sha256:[a-f0-9]{64}", policy["verifier_image_digest"])
        ):
            raise ValueError("signed_policy_is_invalid")
        if (
            record_path.stat().st_size > policy["maximum_record_bytes"]
            or manifest_path.stat().st_size > policy["maximum_manifest_bytes"]
        ):
            raise ValueError("discovery_artifact_exceeds_signed_policy_limit")
        record_raw = verified_bytes(
            record_path, record_signature,
            os.environ["DISCOVERY_RECORD_VERIFY_KEY"],
        )
        manifest_raw = manifest_path.read_bytes()
        if (
            len(record_raw) > policy["maximum_record_bytes"]
            or len(manifest_raw) > policy["maximum_manifest_bytes"]
        ):
            raise ValueError("discovery_artifact_exceeds_signed_policy_limit")
        record = strict_object(record_raw, RECORD_FIELDS, "record")
        issued = parse_time(record["issued_at"])
        expires = parse_time(record["expires_at"])
        if (
            record["schema_version"] != "aidefend.discovery-record.v1"
            or record["registry_id"] != policy["registry_id"]
            or record["policy_version"] != policy["policy_version"]
            or record["agent_id"] != expected_agent_id
            or record["endpoint"] != expected_endpoint
            or not issued <= now < expires
            or expires <= issued
            or expires - issued > timedelta(seconds=policy["maximum_record_seconds"])
        ):
            raise PermissionError("discovery_record_binding_or_validity_failed")
        manifest_sha256 = hashlib.sha256(manifest_raw).hexdigest()
        if (
            not SHA256.fullmatch(record["capability_manifest_sha256"])
            or manifest_sha256 != record["capability_manifest_sha256"]
        ):
            raise PermissionError("capability_manifest_digest_mismatch")
        manifest = strict_object(manifest_raw, MANIFEST_FIELDS, "manifest")
        if (
            manifest["schema_version"] != policy["required_manifest_schema"]
            or manifest["agent_id"] != expected_agent_id
            or manifest["manifest_version"] != record["capability_manifest_version"]
        ):
            raise PermissionError("capability_manifest_binding_failed")
        for field in ("capabilities", "tool_scopes", "data_scopes", "network_destinations"):
            values = manifest[field]
            if not isinstance(values, list) or values != sorted(set(values)):
                raise ValueError(f"{field}_must_be_sorted_unique_strings")
            if any(not isinstance(item, str) or not item for item in values):
                raise ValueError(f"{field}_contains_invalid_value")
        outcome, reason = "PASS", "manifest_and_discovery_record_verified"
    except PermissionError as exc:
        manifest_sha256 = hashlib.sha256(manifest_raw).hexdigest()
        outcome, reason = "FAIL", type(exc).__name__
    except Exception as exc:
        manifest_sha256 = hashlib.sha256(manifest_raw).hexdigest()
        outcome, reason = "ERROR", type(exc).__name__

    decision = {
        "schema_version": "aidefend.capability-admission.v1",
        "control_id": "AID-H-004.004",
        "outcome": outcome,
        "reason_code": reason,
        "agent_id": expected_agent_id,
        "endpoint_sha256": hashlib.sha256(expected_endpoint.encode()).hexdigest(),
        "registry_id": policy.get("registry_id"),
        "record_sha256": hashlib.sha256(record_raw).hexdigest(),
        "manifest_sha256": manifest_sha256,
        "policy_version": policy.get("policy_version"),
        "policy_sha256": policy_sha256,
        "verifier_image_digest": policy.get("verifier_image_digest"),
        "decided_at": now.isoformat(),
    }
    output.parent.mkdir(parents=True, exist_ok=True)
    payload = canonical(decision)
    fd = os.open(
        output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o600
    )
    try:
        os.write(fd, payload)
        os.fsync(fd)
    finally:
        os.close(fd)
    subprocess.run([
        "cosign", "sign-blob", "--yes",
        "--key", "env://CAPABILITY_DECISION_SIGNING_KEY",
        "--bundle", str(output.with_suffix(".sig")), str(output),
    ], check=True, timeout=CAPABILITY_DECISION_SIGN_TIMEOUT_SECONDS)
    subprocess.run([
        "cosign", "verify-blob",
        "--key", os.environ["CAPABILITY_DECISION_VERIFY_KEY"],
        "--bundle", str(output.with_suffix(".sig")), str(output),
    ], check=True, capture_output=True, text=True, timeout=CAPABILITY_DECISION_VERIFY_TIMEOUT_SECONDS)
    return decision

Before you begin

Apply AID-H-004.004-G004 only when an agent endpoint, Agent Card, registry record, ANS-style name, capability manifest, or revocation state is discovered dynamically.

Keep the action safely bounded

deny resolution and routing on unsigned, stale, revoked, ambiguous, rolled-back, mismatched, over-capable, or unverifiable discovery data.

Open-source tools

• SPIRE
• py-spiffe
• pyca/cryptography
• jcs Python package
• jwcrypto
• Requests
• step-ca
• OpenSSL

Commercial / hosted tools

• Palo Alto Networks Next-Generation Trust Security (certificate lifecycle management)
• HashiCorp Vault Enterprise
• HCP Vault Dedicated
• AWS Private CA
• GlobalSign Atlas

Threat mappings

MITRE ATLAS

• AML.T0073 Impersonation
• AML.T0074 Masquerading (ANS prevents agent masquerading through verified identity)
• AML.T0091 Use Alternate Authentication Material (ANS revocation checks invalidate compromised credentials)

MAESTRO

• Agent Impersonation (L7)
• Compromised Agent Registry (L7)
• Malicious Agent Discovery (L7) (ANS prevents malicious agents from being discovered as legitimate)
• Agent Identity Attack (L7) (ANS cryptographic identity binding prevents agent identity attacks)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI03:2026 Identity and Privilege Abuse
• ASI04:2026 Agentic Supply Chain Vulnerabilities (verified discovery prevents supply chain agent compromise)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-3.1.1 Identity Obfuscation
• AISubtech-3.1.2 Trusted Agent Spoofing
• AISubtech-4.1.1 Rogue Agent Introduction
• AISubtech-4.3.2 Namespace Collision
• AISubtech-4.3.5 Capability Inflation
• AITech-3.1 Masquerading / Obfuscation / Impersonation
• AITech-4.1 Agent Injection
• AITech-4.3 Protocol Manipulation

Google Secure AI Framework 2.0 - Risks

• IIC: Insecure Integrated Component (ANS prevents insecure agent discovery)

Databricks AI Security Framework 3.0

• Agents - Core 13.9: Identity Spoofing & Impersonation