AID-H-009.005: Confidential Inference & Remote Attestation

Version 1.20261015 · Actionable sub-technique

Run sensitive inference workloads inside attested confidential-computing environments that can protect model weights, prompts, embeddings, and serving-state caches from the host, hypervisor, and cloud operator. For modern LLM workloads this usually means VM-level TEEs such as Intel TDX or AMD SEV-SNP, paired with GPU confidential computing where accelerated inference is required, rather than assuming a classic SGX enclave can hold large model weights.

Pillars: infra

Lifecycle: building, operation

Scope boundary

Owns confidential inference and key or data release based on fresh CPU, accelerator-set, runtime, interconnect, and supported topology attestation. Multi-GPU policy must distinguish Hopper and Blackwell profiles and must never infer encrypted interconnect or topology assurance from per-GPU attestation alone.

AID-H-005.002: AID-H-009.005 protects plaintext inference inside an attested confidential runtime; AID-H-005.002 keeps supported computations on ciphertext and may use an enclave only for unsupported operations. TEE confidentiality and homomorphic computation have different trust and performance assumptions.

AID-D-004.002: AID-H-009.005 uses attestation as a preventive key and data release gate; AID-D-004.002 verifies and reports confidential-runtime attestation state as a detective integrity control. Preventive secret release and independent runtime monitoring should not collapse into one decision path.

Implementation guidance

AID-H-009.005-G001

Gate model-key and data release on fresh CPU, complete accelerator-set, and supported secure-interconnect attestation evidence from the confidential inference runtime.
Concept:

Put remote attestation in the inference boot and model-key release path. The release service must validate fresh, vendor-signed evidence for the CPU TEE, workload image, debug posture, every assigned accelerator, and any topology or interconnect properties required by the signed deployment policy before it calls KMS or exposes model material. An audit screenshot is not enforcement.

Step 1: Sign the complete release policy

Publish a versioned confidential_inference/release_policy.yaml that pins the exact attestation and JWKS endpoints, audience, issuers, algorithms, accepted workload measurements, platform claims, maximum evidence age, network deadlines, decoded-response byte caps, and AWS SDK attempt budget. Sign the exact policy bytes and distribute the verification key through a separate trust path.

cosign verify-blob --key /opt/aidefend/trust/confidential-release-policy.pub \
  --bundle confidential_inference/release_policy.yaml.sig \
  confidential_inference/release_policy.yaml

An unsigned or malformed policy is an ERROR; missing platform prerequisites are NOT_APPLICABLE, not a pass. The service below verifies an immutable snapshot so a file replacement after signature verification cannot change the policy being enforced.

Step 2: Gate model-key release on bounded, authenticated evidence
# File: confidential_inference/key_release.py
from __future__ import annotations

import base64
import hashlib
import json
import math
import os
import re
import secrets
import subprocess
import tempfile
import time
from pathlib import Path
from typing import Any, Callable
from urllib.parse import urlsplit, urlunsplit

import boto3
from botocore.config import Config
import jwt
import requests
import yaml

SAFE_ALGORITHMS = {
    "RS256", "RS384", "RS512", "PS256", "PS384", "PS512",
    "ES256", "ES384", "ES512", "EdDSA",
}
TOP_LEVEL_FIELDS = {
    "policy_id", "attestor_url", "jwks_url", "audience", "issuers",
    "algorithms", "allowed_verifiers", "allowed_tee_types", "allowed_workloads", "debug_allowed",
    "max_attestation_age_seconds", "clock_skew_seconds", "gpu", "transport",
    "key_release",
}


class UniqueKeyLoader(yaml.SafeLoader):
    pass


def _unique_mapping(loader: UniqueKeyLoader, node: yaml.MappingNode, deep: bool = False) -> dict:
    result: dict = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        if key in result:
            raise ValueError(f"duplicate policy key: {key}")
        result[key] = loader.construct_object(value_node, deep=deep)
    return result


UniqueKeyLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _unique_mapping
)


def _positive_int(value: object, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise ValueError(f"{field} must be a positive integer")
    return value


def _positive_number(value: object, field: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{field} must be numeric")
    result = float(value)
    if not math.isfinite(result) or result <= 0:
        raise ValueError(f"{field} must be finite and positive")
    return result


def _unique_strings(value: object, field: str) -> list[str]:
    if (
        not isinstance(value, list) or not value
        or any(not isinstance(item, str) or not item for item in value)
        or len(value) != len(set(value))
    ):
        raise ValueError(f"{field} must contain unique non-empty strings")
    return value


def canonical_https(value: object, field: str) -> str:
    if not isinstance(value, str) or not value:
        raise ValueError(f"{field} must be a canonical HTTPS URL")
    parts = urlsplit(value)
    if (
        parts.scheme != "https" or not parts.hostname
        or parts.username is not None or parts.password is not None
        or parts.port is not None or parts.query or parts.fragment
        or "\\" in value or parts.netloc != parts.hostname
    ):
        raise ValueError(f"{field} must be HTTPS without credentials, port, query, or fragment")
    canonical = urlunsplit(("https", parts.hostname, parts.path, "", ""))
    if value != canonical or (parts.path not in ("", "/") and parts.path.endswith("/")):
        raise ValueError(f"{field} is not canonical")
    return value


def strict_json(raw: bytes, label: str) -> dict[str, Any]:
    def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
        value: dict[str, Any] = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"{label} contains duplicate key: {key}")
            value[key] = item
        return value

    def reject_constant(value: str) -> None:
        raise ValueError(f"{label} contains non-finite constant: {value}")

    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicates,
        parse_constant=reject_constant,
    )
    if not isinstance(value, dict):
        raise ValueError(f"{label} root must be an object")
    return value


def bounded_json(response: requests.Response, maximum_bytes: int, label: str) -> dict[str, Any]:
    length = response.headers.get("Content-Length")
    if length is not None and (not length.isdigit() or int(length) > maximum_bytes):
        raise PermissionError(f"{label} exceeds signed byte cap")
    raw = bytearray()
    for chunk in response.iter_content(chunk_size=min(65536, maximum_bytes + 1)):
        if chunk:
            raw.extend(chunk)
        if len(raw) > maximum_bytes:
            raise PermissionError(f"{label} exceeds signed byte cap")
    return strict_json(bytes(raw), label)


def verify_policy_snapshot(path: Path, signature: Path, public_key: Path) -> bytes:
    policy_bytes = path.read_bytes()
    timeout = _positive_number(
        float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"]),
        "AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS",
    )
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(prefix="aidefend-policy-", suffix=".yaml", 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(public_key),
             "--bundle", str(signature), str(snapshot)],
            check=True, capture_output=True, text=True, timeout=timeout,
        )
        return policy_bytes
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)


def load_policy(path: str, signature: str, public_key: str) -> tuple[dict[str, Any], str]:
    raw_bytes = verify_policy_snapshot(Path(path), Path(signature), Path(public_key))
    policy = yaml.load(raw_bytes.decode("utf-8", errors="strict"), Loader=UniqueKeyLoader)
    if not isinstance(policy, dict) or set(policy) != TOP_LEVEL_FIELDS:
        raise ValueError("release policy schema mismatch")
    if not isinstance(policy["policy_id"], str) or not policy["policy_id"]:
        raise ValueError("policy_id is invalid")
    policy["attestor_url"] = canonical_https(policy["attestor_url"], "attestor_url")
    policy["jwks_url"] = canonical_https(policy["jwks_url"], "jwks_url")
    if not isinstance(policy["audience"], str) or not policy["audience"]:
        raise ValueError("audience is invalid")
    policy["issuers"] = _unique_strings(policy["issuers"], "issuers")
    policy["algorithms"] = _unique_strings(policy["algorithms"], "algorithms")
    policy["allowed_verifiers"] = _unique_strings(policy["allowed_verifiers"], "allowed_verifiers")
    if not set(policy["algorithms"]).issubset(SAFE_ALGORITHMS):
        raise ValueError("policy permits an unsafe JWT algorithm")
    policy["allowed_tee_types"] = _unique_strings(
        policy["allowed_tee_types"], "allowed_tee_types"
    )
    if not isinstance(policy["debug_allowed"], bool):
        raise ValueError("debug_allowed must be boolean")
    _positive_int(policy["max_attestation_age_seconds"], "max_attestation_age_seconds")
    _positive_int(policy["clock_skew_seconds"], "clock_skew_seconds")

    workloads = policy["allowed_workloads"]
    if not isinstance(workloads, dict) or not workloads:
        raise ValueError("allowed_workloads must be a non-empty object")
    for workload_id, expected in workloads.items():
        if not isinstance(workload_id, str) or not workload_id:
            raise ValueError("workload ID is invalid")
        if not isinstance(expected, dict) or set(expected) != {"measurement", "image_digest"}:
            raise ValueError("workload schema mismatch")
        if re.fullmatch(r"[0-9a-f]{64}", expected["measurement"]) is None:
            raise ValueError("workload measurement is invalid")
        if re.fullmatch(r"sha256:[0-9a-f]{64}", expected["image_digest"]) is None:
            raise ValueError("image digest is invalid")

    transport = policy["transport"]
    if not isinstance(transport, dict) or set(transport) != {
        "connect_timeout_seconds", "read_timeout_seconds",
        "attestation_response_max_bytes", "jwks_response_max_bytes",
        "aws_total_max_attempts",
    }:
        raise ValueError("transport schema mismatch")
    _positive_number(transport["connect_timeout_seconds"], "connect timeout")
    _positive_number(transport["read_timeout_seconds"], "read timeout")
    _positive_int(transport["attestation_response_max_bytes"], "attestation cap")
    _positive_int(transport["jwks_response_max_bytes"], "JWKS cap")
    _positive_int(transport["aws_total_max_attempts"], "AWS total attempts")

    gpu = policy["gpu"]
    if not isinstance(gpu, dict) or set(gpu) != {
        "required", "expected_gpu_count", "expected_switch_count",
        "allowed_profiles", "require_topology_attestation",
        "allowed_multi_gpu_modes", "require_interconnect_encryption",
        "allowed_interconnect_encryption_modes",
    }:
        raise ValueError("GPU policy schema mismatch")
    if not isinstance(gpu["required"], bool):
        raise ValueError("gpu.required must be boolean")
    for field in ("expected_gpu_count", "expected_switch_count"):
        if isinstance(gpu[field], bool) or not isinstance(gpu[field], int) or gpu[field] < 0:
            raise ValueError(f"gpu.{field} must be a non-negative integer")
    for field in ("require_topology_attestation", "require_interconnect_encryption"):
        if not isinstance(gpu[field], bool):
            raise ValueError(f"gpu.{field} must be boolean")
    for field in (
        "allowed_profiles", "allowed_multi_gpu_modes",
        "allowed_interconnect_encryption_modes",
    ):
        gpu[field] = _unique_strings(gpu[field], f"gpu.{field}")
    if gpu["required"] and gpu["expected_gpu_count"] < 1:
        raise ValueError("GPU enforcement requires at least one expected GPU")
    if gpu["require_topology_attestation"] and gpu["expected_switch_count"] < 1:
        raise ValueError("topology attestation requires expected switch evidence")
    if gpu["require_interconnect_encryption"] and gpu["expected_gpu_count"] < 2:
        raise ValueError("interconnect encryption applies only to multi-GPU profiles")

    key_release = policy["key_release"]
    if not isinstance(key_release, dict) or set(key_release) != {
        "kms_key_id", "encrypted_model_key_s3_uri"
    }:
        raise ValueError("key_release schema mismatch")
    if not isinstance(key_release["kms_key_id"], str) or not key_release["kms_key_id"]:
        raise ValueError("KMS key ID is invalid")
    if not isinstance(key_release["encrypted_model_key_s3_uri"], str):
        raise ValueError("encrypted key URI is invalid")
    return policy, hashlib.sha256(raw_bytes).hexdigest()


def build_attestation_session(verify: str, cert: tuple[str, str]) -> requests.Session:
    session = requests.Session()
    session.trust_env = False
    session.verify = verify
    session.cert = cert
    return session


def fetch_jwks(policy: dict[str, Any], session: requests.Session) -> dict[str, Any]:
    transport = policy["transport"]
    with session.get(
        policy["jwks_url"], allow_redirects=False, stream=True,
        timeout=(transport["connect_timeout_seconds"], transport["read_timeout_seconds"]),
    ) as response:
        response.raise_for_status()
        return bounded_json(response, transport["jwks_response_max_bytes"], "JWKS")


def verify_token(token: str, nonce: str, policy: dict[str, Any], jwks: dict[str, Any]) -> dict[str, Any]:
    header = jwt.get_unverified_header(token)
    algorithm, key_id = header.get("alg"), header.get("kid")
    if algorithm not in policy["algorithms"] or not isinstance(key_id, str) or not key_id:
        raise PermissionError("attestation header denied")
    keys = jwks.get("keys")
    if not isinstance(keys, list):
        raise PermissionError("JWKS keys must be an array")
    candidates = [
        item for item in keys
        if isinstance(item, dict) and item.get("kid") == key_id
        and item.get("alg") in (None, algorithm)
        and item.get("use") in (None, "sig")
        and ("key_ops" not in item or (
            isinstance(item["key_ops"], list) and "verify" in item["key_ops"]
        ))
    ]
    if len(candidates) != 1:
        raise PermissionError("attestation signing key is missing or ambiguous")
    signing_key = jwt.PyJWK.from_dict(candidates[0], algorithm=algorithm).key
    claims = jwt.decode(
        token, signing_key, algorithms=[algorithm], audience=policy["audience"],
        issuer=policy["issuers"], leeway=policy["clock_skew_seconds"],
        options={"require": [
            "iss", "exp", "iat", "nonce", "verifier", "tee_type",
            "workload_id", "measurement", "image_digest",
        ]},
    )
    now = int(time.time())
    issued_at = claims["iat"]
    if isinstance(issued_at, bool) or not isinstance(issued_at, int):
        raise PermissionError("attestation iat is invalid")
    if issued_at > now + policy["clock_skew_seconds"]:
        raise PermissionError("attestation was issued in the future")
    if now - issued_at > policy["max_attestation_age_seconds"]:
        raise PermissionError("attestation is stale")
    if not isinstance(claims["nonce"], str) or not secrets.compare_digest(claims["nonce"], nonce):
        raise PermissionError("attestation nonce mismatch")
    if claims["verifier"] not in policy["allowed_verifiers"]:
        raise PermissionError("attestation verifier denied")
    if claims["tee_type"] not in policy["allowed_tee_types"]:
        raise PermissionError("TEE type denied")
    if not isinstance(claims.get("debug_enabled", False), bool):
        raise PermissionError("debug claim is malformed")
    if claims.get("debug_enabled", False) and not policy["debug_allowed"]:
        raise PermissionError("debug-enabled workload denied")
    expected = policy["allowed_workloads"].get(claims["workload_id"])
    if not expected or claims["measurement"] != expected["measurement"] or claims["image_digest"] != expected["image_digest"]:
        raise PermissionError("workload measurement denied")

    gpu_policy = policy["gpu"]
    if gpu_policy["required"]:
        devices = claims.get("gpu_devices")
        if not isinstance(devices, list) or len(devices) != gpu_policy["expected_gpu_count"]:
            raise PermissionError("GPU evidence population is incomplete")
        identities: set[str] = set()
        for device in devices:
            identity = device.get("ueid_sha256") if isinstance(device, dict) else None
            if not isinstance(identity, str) or re.fullmatch(r"[0-9a-f]{64}", identity) is None or identity in identities:
                raise PermissionError("GPU identity is invalid or duplicated")
            identities.add(identity)
            if device.get("attested") is not True or device.get("confidential_compute") is not True:
                raise PermissionError("GPU confidential-compute evidence failed")
            if device.get("profile") not in gpu_policy["allowed_profiles"]:
                raise PermissionError("GPU profile denied")
        switches = claims.get("switch_devices", [])
        if not isinstance(switches, list) or len(switches) != gpu_policy["expected_switch_count"]:
            raise PermissionError("switch evidence population is incomplete")
        if any(not isinstance(item, dict) or item.get("attested") is not True for item in switches):
            raise PermissionError("switch evidence failed")
        if gpu_policy["require_topology_attestation"] and claims.get("topology_attested") is not True:
            raise PermissionError("topology attestation is missing")
        if len(devices) > 1 and claims.get("multi_gpu_mode") not in gpu_policy["allowed_multi_gpu_modes"]:
            raise PermissionError("multi-GPU mode denied")
        if gpu_policy["require_interconnect_encryption"] and (
            claims.get("interconnect_encryption_attested") is not True
            or claims.get("interconnect_encryption_mode") not in gpu_policy["allowed_interconnect_encryption_modes"]
        ):
            raise PermissionError("interconnect encryption not proven")
    return claims


def request_attestation(nonce: str, policy: dict[str, Any], session: requests.Session) -> str:
    transport = policy["transport"]
    with session.post(
        policy["attestor_url"], json={"nonce": nonce}, allow_redirects=False,
        stream=True,
        timeout=(transport["connect_timeout_seconds"], transport["read_timeout_seconds"]),
    ) as response:
        response.raise_for_status()
        body = bounded_json(
            response, transport["attestation_response_max_bytes"], "attestation"
        )
    if set(body) != {"attestation_token"} or not isinstance(body["attestation_token"], str):
        raise PermissionError("attestation response schema mismatch")
    return body["attestation_token"]


def encrypted_key_blob(uri: str, s3_client: Any) -> bytes:
    if not uri.startswith("s3://"):
        raise ValueError("encrypted key URI must use s3://")
    bucket, separator, key = uri[5:].partition("/")
    if not separator or not bucket or not key:
        raise ValueError("encrypted key URI is invalid")
    body = s3_client.get_object(Bucket=bucket, Key=key)["Body"]
    encoded = body.read(65537)
    body.close()
    if not encoded or len(encoded) > 65536:
        raise ValueError("encrypted key blob is empty or oversized")
    return base64.b64decode(encoded.strip(), validate=True)


def release_key(
    policy_path: str, signature: str, public_key: str,
    ca_bundle: str, client_cert: tuple[str, str],
    key_consumer: Callable[[memoryview], None],
) -> dict[str, Any]:
    policy, policy_digest = load_policy(policy_path, signature, public_key)
    nonce = secrets.token_urlsafe(32)
    with build_attestation_session(ca_bundle, client_cert) as attestation_session:
        token = request_attestation(nonce, policy, attestation_session)
        jwks = fetch_jwks(policy, attestation_session)
    claims = verify_token(token, nonce, policy, jwks)

    transport = policy["transport"]
    sdk_config = Config(
        connect_timeout=transport["connect_timeout_seconds"],
        read_timeout=transport["read_timeout_seconds"],
        retries={
            "mode": "standard",
            "total_max_attempts": transport["aws_total_max_attempts"],
        },
    )
    aws_session = boto3.session.Session()
    s3_client = aws_session.client("s3", config=sdk_config)
    kms_client = aws_session.client("kms", config=sdk_config)
    ciphertext = encrypted_key_blob(
        policy["key_release"]["encrypted_model_key_s3_uri"], s3_client
    )
    result = kms_client.decrypt(
        CiphertextBlob=ciphertext, KeyId=policy["key_release"]["kms_key_id"]
    )
    plaintext = result.get("Plaintext")
    if not isinstance(plaintext, bytes) or not plaintext:
        raise PermissionError("KMS returned no plaintext")
    working_key = bytearray(plaintext)
    result.clear()
    del plaintext
    try:
        key_consumer(memoryview(working_key))
    finally:
        working_key[:] = b"\x00" * len(working_key)

    return {
        "outcome": "PASS", "policy_id": policy["policy_id"],
        "policy_sha256": policy_digest,
        "attestation_token_sha256": hashlib.sha256(token.encode("utf-8")).hexdigest(),
        "nonce_sha256": hashlib.sha256(nonce.encode("utf-8")).hexdigest(),
        "issuer": claims["iss"], "workload_id": claims["workload_id"],
        "image_digest": claims["image_digest"], "attestation_iat": claims["iat"],
        "gpu_count": len(claims.get("gpu_devices", [])),
        "switch_count": len(claims.get("switch_devices", [])),
    }
Step 3: Replay the decision independently

Retain the raw vendor evidence, pinned adapter and support-profile versions, signed-policy bytes and digest, normalized claims, hashed nonce and token, complete assigned-device population, and KMS decision in append-only evidence. A separate verifier should replay the signature, JWT, freshness, measurement, device-population, topology, and policy checks against those exact artifacts. Emit PASS only when the release path actually withheld the key until every applicable check succeeded. A mismatch blocks release and records the failed predicate; missing evidence leaves the decision incomplete; an absent confidential-computing prerequisite goes through applicability assessment; and verifier or dependency faults stop release for operator recovery. Re-attest on boot, model reload, node replacement, image or driver change, topology change, policy change, and before the signed maximum age expires. Use a TEE-native key handle or sealed loader when managed-runtime copies make bytearray zeroization insufficient.

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-H-009.005: Confidential Inference & Remote Attestation | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service