Scope boundary

Owns fresh nonce-bound remote-attestation verification for supported confidential inference runtimes, including verifier and runtime identity, measurement, policy, freshness, single-use challenge, and explicit unsupported or unverifiable states.

AID-D-004.001: AID-D-004.002 verifies a fresh signed measurement of the confidential runtime environment; AID-D-004.001 verifies static stored artifact hashes and signatures. Remote attestation does not replace artifact-manifest verification, and static integrity does not prove the current trusted-execution state.

AID-D-004.008: AID-D-004.002 verifies the platform attestation measurement and runtime identity; AID-D-004.008 observes the complete eligible population of code and modules loaded by the deployed process. An approved attestation measurement may not expose every dynamic module load, while module monitoring is not hardware-backed attestation.

AID-D-004.003: AID-D-004.002 verifies confidential-runtime measurements; AID-D-004.003 compares live application, service, and policy configuration with approved state. Attestation and configuration drift have different evidence roots even when both describe one runtime.

Implementation guidance

• AID-D-004.002-G001: Verify a fresh, nonce-bound, signed remote-attestation result against the expected runtime identity and measurement before releasing traffic or secrets.

AID-D-004.002-G001

Before you begin

Use this only when a remote-attestation verifier can return a signed result for the runtime.

One-time challenge and signed-policy verifier

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")
JWKS_TIMEOUT_SECONDS = _aidefend_positive_float("D004002_JWKS_TIMEOUT_SECONDS")
JWKS_CONNECT_TIMEOUT_SECONDS = _aidefend_positive_float("D004002_JWKS_CONNECT_TIMEOUT_SECONDS")
POLICY_VERIFY_TIMEOUT_SECONDS = _aidefend_positive_float("D004002_POLICY_VERIFY_TIMEOUT_SECONDS")
JWKS_MAX_RESPONSE_BYTES = _aidefend_positive_int("D004002_JWKS_MAX_RESPONSE_BYTES")
JWKS_RESPONSE_CHUNK_BYTES = _aidefend_positive_int("D004002_JWKS_RESPONSE_CHUNK_BYTES")
if JWKS_MAX_RESPONSE_BYTES < 2 or JWKS_RESPONSE_CHUNK_BYTES > JWKS_MAX_RESPONSE_BYTES:
    raise RuntimeError("signed JWKS response bounds are invalid")

# File: attestation/verify_runtime.py
# Dependencies: boto3>=1.35,<2, httpx>=0.27,<1,
# PyJWT>=2.9,<3, cryptography>=43,<45

import hashlib
import json
import os
import secrets
import ssl
import subprocess
import time
import uuid
from pathlib import Path
import hmac
import re
import tempfile
from urllib.parse import urlsplit

import boto3
import httpx
import jwt
from botocore.exceptions import ClientError
from jwt.algorithms import RSAAlgorithm

CHALLENGE_TABLE = os.environ["ATTESTATION_CHALLENGE_TABLE"]
POLICY_PATH = Path(os.environ["MEASUREMENT_POLICY_PATH"])
POLICY_SIGNATURE = Path(os.environ["MEASUREMENT_POLICY_SIGNATURE_PATH"])
POLICY_PUBLIC_KEY = Path(os.environ["MEASUREMENT_POLICY_PUBLIC_KEY_PATH"])
JWKS_URL = os.environ["ATTESTATION_JWKS_URL"]
JWKS_CA = os.environ["ATTESTATION_JWKS_CA_BUNDLE"]
ddb = boto3.client("dynamodb")


class AttestationError(Exception):
    pass




def _write_private_snapshot(path: Path, payload: bytes) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def _strict_json_bytes(raw: bytes) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key: " + key)
            value[key] = item
        return value

    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def capture_verified_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str,
    timeout_seconds: float,
    proof_flag: str = "--bundle",
) -> bytes:
    if payload_path == proof_path or proof_flag != "--bundle":
        raise AttestationError("signed payload/proof inputs are invalid")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
    )) or key_reference.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else Path(key_reference).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-d004002-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "proof"
        _write_private_snapshot(payload_snapshot, payload_raw)
        _write_private_snapshot(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _write_private_snapshot(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(proof_snapshot), str(payload_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise AttestationError("private signature-verification snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)
    return payload_raw
def sha256_bytes(value):
    return hashlib.sha256(value).hexdigest()


def load_signed_policy():
    raw = capture_verified_bytes(
        POLICY_PATH,
        POLICY_SIGNATURE,
        str(POLICY_PUBLIC_KEY),
        POLICY_VERIFY_TIMEOUT_SECONDS,
    )
    policy = _strict_json_bytes(raw)
    required = {
        "schema_version", "policy_version", "verifier_issuer",
        "verifier_audience", "allowed_measurements_by_runtime",
        "challenge_ttl_seconds", "max_result_age_seconds",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise AttestationError("measurement policy schema mismatch")
    if policy["schema_version"] != "aidefend.attestation_policy.v1":
        raise AttestationError("unsupported measurement policy")
    if (
        not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
        or not isinstance(policy["verifier_issuer"], str)
        or not isinstance(policy["verifier_audience"], str)
    ):
        raise AttestationError("measurement policy identity invalid")
    for field in ("challenge_ttl_seconds", "max_result_age_seconds"):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise AttestationError(f"{field} must be a positive integer")
    measurements = policy["allowed_measurements_by_runtime"]
    if not isinstance(measurements, dict) or not measurements:
        raise AttestationError("measurement allowlist is empty")
    for runtime_id, values in measurements.items():
        if (
            not isinstance(runtime_id, str) or not runtime_id
            or not isinstance(values, list) or not values
            or len(values) != len(set(values))
            or any(
                not isinstance(value, str)
                or len(value) != 64
                or any(ch not in "0123456789abcdef" for ch in value)
                for value in values
            )
        ):
            raise AttestationError("measurement allowlist entry invalid")
    return policy, sha256_bytes(raw)


def issue_challenge(runtime_id):
    policy, policy_sha256 = load_signed_policy()
    if runtime_id not in policy["allowed_measurements_by_runtime"]:
        raise AttestationError("runtime is absent from signed measurement policy")
    nonce = secrets.token_bytes(32)
    challenge_id = uuid.uuid4().hex
    now = int(time.time())
    ddb.put_item(
        TableName=CHALLENGE_TABLE,
        Item={
            "challenge_id": {"S": challenge_id},
            "runtime_id": {"S": runtime_id},
            "nonce_sha256": {"S": sha256_bytes(nonce)},
            "policy_version": {"S": policy["policy_version"]},
            "policy_sha256": {"S": policy_sha256},
            "status": {"S": "ISSUED"},
            "issued_at": {"N": str(now)},
            "expires_at": {
                "N": str(now + policy["challenge_ttl_seconds"])
            },
        },
        ConditionExpression="attribute_not_exists(challenge_id)",
    )
    return {
        "challenge_id": challenge_id,
        "nonce": nonce.hex(),
        "runtime_id": runtime_id,
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
    }


def reject_duplicate_keys(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise AttestationError(f"duplicate JWKS key: {key}")
        value[key] = item
    return value


def jwks_key(token):
    parsed = urlsplit(JWKS_URL)
    if (parsed.scheme != "https" or not parsed.hostname or parsed.username
            or parsed.password or parsed.fragment):
        raise AttestationError("JWKS URL is not one canonical HTTPS endpoint")
    tls = ssl.create_default_context(cafile=JWKS_CA)
    timeout = httpx.Timeout(JWKS_TIMEOUT_SECONDS, connect=JWKS_CONNECT_TIMEOUT_SECONDS)
    with httpx.Client(verify=tls, timeout=timeout, follow_redirects=False) as client:
        with client.stream("GET", JWKS_URL, headers={"Accept-Encoding": "identity"}) as response:
            if 300 <= response.status_code < 400:
                raise AttestationError("JWKS redirect is forbidden")
            response.raise_for_status()
            if response.headers.get("Content-Encoding", "identity").lower() != "identity":
                raise AttestationError("encoded JWKS response is forbidden")
            declared = response.headers.get("Content-Length")
            if declared is not None and (not declared.isdigit() or int(declared) > JWKS_MAX_RESPONSE_BYTES):
                raise AttestationError("JWKS Content-Length is invalid")
            chunks, observed = [], 0
            for chunk in response.iter_bytes(JWKS_RESPONSE_CHUNK_BYTES):
                observed += len(chunk)
                if observed > JWKS_MAX_RESPONSE_BYTES:
                    raise AttestationError("JWKS response exceeds signed bound")
                chunks.append(chunk)
    if declared is not None and int(declared) != observed:
        raise AttestationError("JWKS Content-Length differs from observed bytes")
    try:
        document = json.loads(b"".join(chunks).decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise AttestationError("JWKS is not strict UTF-8 JSON") from exc
    if not isinstance(document, dict) or set(document) != {"keys"} or not isinstance(document["keys"], list):
        raise AttestationError("JWKS schema differs")
    header = jwt.get_unverified_header(token)
    if header.get("alg") != "RS256" or not isinstance(header.get("kid"), str):
        raise AttestationError("unexpected verifier token header")
    matches = [key for key in document["keys"]
               if isinstance(key, dict) and key.get("kid") == header["kid"] and key.get("alg") == "RS256"]
    if len(matches) != 1:
        raise AttestationError("verifier signing key unavailable or ambiguous")
    return RSAAlgorithm.from_jwk(json.dumps(matches[0]))


def verify_and_consume(token):
    policy, policy_sha256 = load_signed_policy()
    key = jwks_key(token)
    claims = jwt.decode(
        token,
        key,
        algorithms=["RS256"],
        issuer=policy["verifier_issuer"],
        audience=policy["verifier_audience"],
        leeway=5,
        options={"require": [
            "iss", "aud", "iat", "nbf", "exp", "jti", "challenge_id",
            "nonce_sha256", "runtime_id", "measurement_sha256",
            "policy_sha256", "result",
        ]},
    )
    now = int(time.time())
    if claims["result"] not in {"VERIFIED", "FAILED"}:
        raise AttestationError("attestation result enumeration invalid")
    if (
        claims["policy_sha256"] != policy_sha256
        or now - int(claims["iat"]) > policy["max_result_age_seconds"]
    ):
        raise AttestationError("attestation policy binding or freshness invalid")
    challenge = ddb.get_item(
        TableName=CHALLENGE_TABLE,
        Key={"challenge_id": {"S": claims["challenge_id"]}},
        ConsistentRead=True,
    ).get("Item")
    if challenge is None:
        raise AttestationError("unknown challenge")
    expected = {
        "runtime_id": challenge["runtime_id"]["S"],
        "nonce_sha256": challenge["nonce_sha256"]["S"],
        "policy_sha256": challenge["policy_sha256"]["S"],
    }
    if any(claims.get(field) != value for field, value in expected.items()):
        raise AttestationError("attestation is not bound to issued challenge")
    if challenge["policy_version"]["S"] != policy["policy_version"]:
        raise AttestationError("challenge measurement policy version changed")
    if (
        challenge["status"]["S"] != "ISSUED"
        or int(challenge["expires_at"]["N"]) < now
    ):
        raise AttestationError("challenge expired or already consumed")
    allowed = policy["allowed_measurements_by_runtime"].get(
        claims["runtime_id"], []
    )
    findings = []
    if claims["result"] != "VERIFIED":
        findings.append("verifier_reported_attestation_failure")
    if claims["measurement_sha256"] not in allowed:
        findings.append("runtime_measurement_not_approved")

    try:
        ddb.update_item(
            TableName=CHALLENGE_TABLE,
            Key={"challenge_id": {"S": claims["challenge_id"]}},
            UpdateExpression="SET #s=:used, consumed_at=:now, result_jti=:jti",
            ConditionExpression="#s=:issued AND expires_at >= :now",
            ExpressionAttributeNames={"#s": "status"},
            ExpressionAttributeValues={
                ":used": {"S": "CONSUMED"},
                ":issued": {"S": "ISSUED"},
                ":now": {"N": str(now)},
                ":jti": {"S": claims["jti"]},
            },
        )
    except ClientError as exc:
        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":
            raise AttestationError("challenge replay or expiry") from exc
        raise
    return {
        "schema_version": "aidefend.d004002-attestation-result.v1",
        "control_status": "PASS",
        "finding_status": "FINDING" if findings else "NO_FINDING",
        "findings": findings,
        "runtime_id": claims["runtime_id"],
        "measurement_sha256": claims["measurement_sha256"],
        "challenge_id": claims["challenge_id"],
        "result_jti": claims["jti"],
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
    }


def safe_verify(token):
    try:
        return verify_and_consume(token)
    except Exception as exc:
        return {
            "schema_version": "aidefend.d004002-attestation-result.v1",
            "control_status": "ERROR",
            "finding_status": "UNKNOWN",
            "error_class": type(exc).__name__,
        }

Evidence: retain the signed measurement policy and signature, JWKS digest and key ID, challenge issuance/consumption record, compact verifier token, verified claims, policy digest/version, and result. Independently replay policy and token signatures and challenge bindings. Test caller-created nonce, wrong runtime/policy/measurement, expired result, reused challenge, concurrent replay, policy/JWKS/challenge-store outage, unknown key, and partial consumption.

Open-source tools

• Intel SGX SDK
• Open Enclave SDK
• AWS Nitro Enclaves SDK
• Keylime
• Veraison
• Cosign

Commercial / hosted tools

• Microsoft Azure Attestation
• Google Cloud Confidential Computing
• AWS Nitro Enclaves
• Intel Trust Authority

Threat mappings

MITRE ATLAS

• AML.T0010 AI Supply Chain Compromise
• AML.T0010.001 AI Supply Chain Compromise: AI Software
• AML.T0107 Exploitation for Defense Evasion

MAESTRO

• Orchestration Attacks (L4)
• Compromised Container Images (L4)

OWASP LLM Top 10 2026

• LLM02:2026 Sensitive Information Disclosure (fresh attestation prevents secrets from entering an unapproved confidential-inference runtime)
• LLM04:2026 Supply Chain

OWASP ML Top 10 2023

• ML06:2023 AI Supply Chain Attacks

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities

NIST Adversarial Machine Learning 2025

• NISTAML.051 Model Poisoning (Supply Chain)

Cisco Integrated AI Security and Safety Framework

• AITech-9.3 Dependency / Plugin Compromise (integrity changes visible in the attested model measurement)

Google Secure AI Framework 2.0 - Risks

• MDT: Model Deployment Tampering

Databricks AI Security Framework 3.0

• Model 7.3: ML Supply chain vulnerabilities
• Model 7.4: Source code control attack