Scope boundary

Owns deployment or cluster admission based on hardware and firmware identity, measured boot or attestation evidence, approved accelerator inventory, freshness, and policy binding. It does not establish purchase custody or manage firmware patching.

AID-H-009.001: AID-H-003.004 consumes trusted boot and hardware identity evidence to admit a node or workload; AID-H-009.001 establishes the hardware root of trust and secure-boot chain. Root-of-trust evidence is an input, while cluster admission is the policy decision over the exact deployment.

AID-H-009.002: AID-H-003.004 blocks nodes whose firmware or driver state is not approved; AID-H-009.002 owns firmware and driver vulnerability, patch, rollout, and rollback management. Admission verifies current state but does not perform lifecycle patching.

AID-H-009.003: AID-H-003.004 admits hardware from technical attestation and inventory evidence; AID-H-009.003 verifies purchase receipts, serial custody, shipment authenticity, and tamper-evident handling. Custody provenance and runtime technical attestation are complementary hardware trust claims.

Implementation guidance

• AID-H-003.004-G001: Secure-boot GPUs/TPUs; attestation via TPM/CCA or NVIDIA Confidential Computing.
• AID-H-003.004-G002: Use fresh nonce-bound TPM quotes and an independent verifier to admit hosts or release workload credentials only after the approved measured-boot state is proven.
• AID-H-003.004-G003: Evaluate observed firmware versions against a version-controlled admission policy and quarantine nodes that fail onboarding or re-admission checks.
• AID-H-003.004-G004: Run side-channel/fault-injection self-tests during maintenance windows.

AID-H-003.004-G001

Production implementation: verify fresh OIDC attestation, never a caller verdict

The admission controller creates a single-use nonce and receives the provider token directly over the attestation channel. A signature-verified policy pins issuer, audience, discovery and JWKS URLs, accepted algorithms, maximum age, and exact required measurements.

# File: cluster_admission/verify_oidc_attestation.py
from __future__ import annotations

import json
import math
import time
from urllib.parse import urlsplit

import httpx
import jwt


POLICY_FIELDS = {
    "policy_version", "issuer", "audience", "discovery_url", "jwks_uri",
    "algorithms", "maximum_token_bytes", "maximum_token_age_seconds",
    "attestation_http_timeout_seconds", "maximum_discovery_bytes",
    "maximum_jwks_bytes", "required_claims",
}


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


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 canonical_https(raw: object, field: str) -> str:
    if not isinstance(raw, str) or not raw or raw != raw.strip():
        raise ValueError(f"{field} is empty or non-canonical")
    parsed = urlsplit(raw)
    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
    ):
        raise ValueError(f"{field} must be canonical HTTPS")
    return raw


def strict_json(raw: bytes) -> object:
    def unique_object(pairs: list[tuple[str, object]]) -> dict:
        value: dict = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"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(f"invalid JSON constant: {value}")
        ),
    )


def read_limited(response: httpx.Response, max_bytes: int) -> bytes:
    parts: list[bytes] = []
    total = 0
    for chunk in response.iter_bytes():
        room = max_bytes + 1 - total
        parts.append(chunk[: max(room, 0)])
        total += min(len(chunk), max(room, 0))
        if len(chunk) > room or total > max_bytes:
            raise ValueError("OIDC response exceeds signed byte bound")
    return b"".join(parts)


def get_json(
    client: httpx.Client,
    url: str,
    *,
    max_bytes: int,
    media_types: set[str],
) -> object:
    with client.stream(
        "GET",
        url,
        headers={"Accept": "application/json", "Accept-Encoding": "identity"},
    ) as response:
        if response.status_code != 200:
            raise PermissionError(
                f"OIDC endpoint returned HTTP {response.status_code}"
            )
        media_type = response.headers.get("content-type", "").split(";", 1)[0]
        if media_type.lower() not in media_types:
            raise ValueError("OIDC endpoint media type differs")
        raw = read_limited(response, max_bytes)
    return strict_json(raw)


def verify_token(
    token: str,
    policy: dict,
    expected_nonce: str,
    replay_store,
) -> dict:
    if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
        raise ValueError("attestation policy schema differs")
    if (
        not isinstance(token, str)
        or not token.isascii()
        or token.count(".") != 2
        or not isinstance(expected_nonce, str)
        or not expected_nonce
    ):
        raise ValueError("attestation token or nonce is invalid")

    token_cap = positive_int(policy["maximum_token_bytes"], "maximum_token_bytes")
    discovery_cap = positive_int(
        policy["maximum_discovery_bytes"], "maximum_discovery_bytes"
    )
    jwks_cap = positive_int(policy["maximum_jwks_bytes"], "maximum_jwks_bytes")
    maximum_age = positive_int(
        policy["maximum_token_age_seconds"], "maximum_token_age_seconds"
    )
    timeout_seconds = positive_number(
        policy["attestation_http_timeout_seconds"],
        "attestation_http_timeout_seconds",
    )
    if len(token.encode("ascii")) > token_cap:
        raise ValueError("attestation token exceeds signed byte bound")

    issuer = canonical_https(policy["issuer"], "issuer")
    discovery_url = canonical_https(policy["discovery_url"], "discovery_url")
    jwks_uri = canonical_https(policy["jwks_uri"], "jwks_uri")
    if (
        not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
        or not isinstance(policy["audience"], str)
        or not policy["audience"]
        or not isinstance(policy["algorithms"], list)
        or not policy["algorithms"]
        or len(policy["algorithms"]) != len(set(policy["algorithms"]))
        or any(
            not isinstance(algorithm, str) or not algorithm
            for algorithm in policy["algorithms"]
        )
        or not isinstance(policy["required_claims"], dict)
        or not policy["required_claims"]
    ):
        raise ValueError("signed attestation policy values are invalid")

    with httpx.Client(
        timeout=httpx.Timeout(timeout_seconds),
        follow_redirects=False,
    ) as client:
        metadata = get_json(
            client,
            discovery_url,
            max_bytes=discovery_cap,
            media_types={"application/json"},
        )
        if (
            not isinstance(metadata, dict)
            or metadata.get("issuer") != issuer
            or metadata.get("jwks_uri") != jwks_uri
        ):
            raise PermissionError("issuer discovery differs from signed policy")
        jwks = get_json(
            client,
            jwks_uri,
            max_bytes=jwks_cap,
            media_types={
                "application/json",
                "application/jwk-set+json",
            },
        )
    if (
        not isinstance(jwks, dict)
        or not isinstance(jwks.get("keys"), list)
        or not jwks["keys"]
    ):
        raise ValueError("JWKS schema differs")
    key_set = jwt.PyJWKSet.from_dict(jwks)

    header = jwt.get_unverified_header(token)
    if (
        not isinstance(header, dict)
        or not isinstance(header.get("kid"), str)
        or not header["kid"]
        or header.get("alg") not in policy["algorithms"]
    ):
        raise PermissionError("attestation header is not approved")
    candidates = [
        key for key in key_set.keys if key.key_id == header["kid"]
    ]
    if len(candidates) != 1:
        raise PermissionError("attestation key is absent or ambiguous")
    claims = jwt.decode(
        token,
        candidates[0].key,
        algorithms=policy["algorithms"],
        issuer=issuer,
        audience=policy["audience"],
        leeway=0,
        options={
            "require": [
                "iss", "sub", "aud", "exp", "iat", "nbf",
                "jti", "nonce",
            ]
        },
    )
    now = int(time.time())
    for field in ("iat", "nbf", "exp"):
        if isinstance(claims[field], bool) or not isinstance(claims[field], int):
            raise PermissionError(f"attestation {field} must be an integer")
    if (
        claims["nonce"] != expected_nonce
        or claims["iat"] > now
        or now - claims["iat"] > maximum_age
        or claims["nbf"] > now
        or claims["exp"] <= now
        or not isinstance(claims["jti"], str)
        or not claims["jti"]
        or not isinstance(claims["sub"], str)
        or not claims["sub"]
    ):
        raise PermissionError("attestation identity, nonce, or freshness differs")
    if not replay_store.put_if_absent(
        claims["jti"], expires_at=claims["exp"]
    ):
        raise PermissionError("attestation token replayed")
    for claim, expected in policy["required_claims"].items():
        if not isinstance(claim, str) or not claim or claims.get(claim) != expected:
            raise PermissionError("attested measurement differs: " + str(claim))
    return {
        "schema_version": "aidefend.node-attestation.v1",
        "issuer": claims["iss"],
        "subject": claims["sub"],
        "audience": policy["audience"],
        "jti": claims["jti"],
        "issued_at": claims["iat"],
        "expires_at": claims["exp"],
        "policy_version": policy["policy_version"],
        "measurements": {
            name: claims[name] for name in policy["required_claims"]
        },
    }

Action: Stream discovery and JWKS through signed positive byte caps, reject redirects and non-canonical HTTPS authorities, require HTTP 200 plus an approved JSON media type, reject duplicate keys, and bind issuer/JWKS exactly before validating the capped token. Bind the signed admission receipt to node identity, nonce, token digest, verified claims, firmware policy, cluster/pool, verifier image, and decision. Read back the admitted node label/taint and replay with wrong nonce, audience, issuer, key, measurement, expiry, and JTI. Any verifier or replay-store error quarantines the node.

Before you begin

Apply AID-H-003.004-G001 only when AI workloads depend on accelerator, host firmware, confidential-computing, secure-boot, or attested cluster state.

AID-H-003.004-G002

Concept:

AID-H-009.001 establishes the local measured-boot state; this AID-H-003.004 method independently proves that state to a remote verifier before cluster admission or credential release. This guidance is best treated as a reference architecture because the exact verifier, PCR policy, and credential-release workflow depend on your TPM stack, cloud environment, and secret-delivery system.

Step 1: Load a concrete signed verifier policy

The asset and attestation authorities publish one final JSON policy with the concrete asset identity, PCR digests, and credential lifetime. Treat it as a signed input artifact, not an edit-time template. The admission job verifies an immutable byte snapshot, validates every typed field, then installs only those verified bytes.

set -euo pipefail
umask 077
policy=attestation/training_vm_policy.json
signature=attestation/training_vm_policy.json.sig
snapshot=$(mktemp)
trap 'rm -f "$snapshot"' EXIT
install -m 0600 "$policy" "$snapshot"
cosign verify-blob --key /opt/aidefend/trust/attestation-policy.pub \
  --bundle "$signature" "$snapshot"
jq -e '
  (type == "object") and
  ((keys | sort) == ["agent_id","failure_action","nonce_binding","policy_version","required_pcrs","schema_version","secret_release","verifier"]) and
  (.schema_version == "aidefend.remote-attestation-policy.v1") and
  ((.policy_version | type) == "string" and (.policy_version | length) > 0) and
  ((.agent_id | type) == "string" and (.agent_id | length) > 0) and
  (.verifier == "keylime") and (.nonce_binding == "required") and
  ((.required_pcrs | type) == "object" and (.required_pcrs | length) > 0) and
  all(.required_pcrs | to_entries[]; (.key | test("^[0-9]+$")) and (.value | test("^sha256:[0-9a-f]{64}$"))) and
  (.secret_release.mode == "allow_only_on_success") and
  (.secret_release.issued_credential_ttl_seconds as $ttl | ($ttl | type) == "number" and $ttl >= 1 and $ttl == ($ttl | floor)) and
  (.failure_action == {"network_join":"deny","secret_delivery":"deny","incident_signal":"host_attestation_failed"})
' "$snapshot" >/dev/null
install -D -m 0440 "$snapshot" /etc/aidefend/attestation/training-vm-policy.json
sha256sum "$snapshot" > evidence/training-vm-policy.sha256

Step 2: Enforce attestation before secrets or workloads are released

The attestor should present a TPM quote bound to the verifier's nonce. The verifier checks the AIK chain, validates the quote signature, compares PCR values to the approved baseline, and only then issues a short-lived credential or workload admission token.

Action: In high-security environments, deploy a remote attestation service like Keylime. Configure sensitive AI workloads so they can only start or receive secrets after successful attestation against an approved PCR policy.

Keep the action safely bounded

Request fresh nonce/audience-bound vendor evidence, validate signature/certificate/revocation, full required measurements, firmware/TCB/debug/rollback and asset identity, then release workload credentials only to a matching host; deny stale, partial or unknown evidence.

Verify safely

A separately credentialed verifier independently reads back or replays each host/workload admission with exact nonce, verifier policy, PCR/measurement/TCB baseline and complete fleet population, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-003.004-G003

Concept:

Supply-chain firmware assurance at cluster admission requires observed facts, not procurement assertions. Treat accelerator firmware like any other production release artifact: collect a version inventory, compare it against an approved onboarding baseline, and quarantine nodes that violate policy before they serve or train models.

Step 1: Collect accelerator and host firmware facts

Query GPU VBIOS and host BIOS versions with vendor-supported tooling, and emit a structured JSON record for your inventory or admission pipeline.

# File: firmware_audit/collect_firmware.py
from __future__ import annotations

import json
import math
import os
import socket
import subprocess


def required_timeout() -> float:
    value = float(os.environ["VERIFIED_H003_FIRMWARE_COMMAND_TIMEOUT_SECONDS"])
    if not math.isfinite(value) or value <= 0: raise ValueError("firmware command timeout must be finite and positive")
    return value

FIRMWARE_COMMAND_TIMEOUT_SECONDS = required_timeout()

def run_checked(command: list[str]) -> str:
    return subprocess.run(
        command,
        check=True,
        capture_output=True,
        text=True,
        timeout=FIRMWARE_COMMAND_TIMEOUT_SECONDS,
    ).stdout.strip()


def collect_firmware_inventory() -> dict:
    gpu_rows = run_checked(
        [
            "nvidia-smi",
            "--query-gpu=name,serial,vbios_version,driver_version",
            "--format=csv,noheader",
        ]
    ).splitlines()

    gpus = []
    for row in gpu_rows:
        name, serial, vbios_version, driver_version = [part.strip() for part in row.split(",")]
        gpus.append(
            {
                "name": name,
                "serial": serial,
                "vbios_version": vbios_version,
                "driver_version": driver_version,
            }
        )

    return {
        "hostname": socket.gethostname(),
        "bios_vendor": run_checked(["dmidecode", "-s", "bios-vendor"]),
        "bios_version": run_checked(["dmidecode", "-s", "bios-version"]),
        "gpus": gpus,
    }


if __name__ == "__main__":
    print(json.dumps(collect_firmware_inventory(), indent=2))

Step 2: Compare inventory against a version-controlled policy

Maintain an approved firmware policy in Git and fail closed on both outdated and explicitly blocked versions.

# File: firmware_audit/policy.yaml
hardware_policies:
  host_bios:
    vendor: "Dell Inc."
    min_version: "2.18.1"

  gpu_vbios:
    "NVIDIA A100-SXM4-80GB":
      min_version: "92.00.45.00.05"
      blocked_versions:
        - "92.00.19.00.01"

# File: firmware_audit/evaluate_firmware.py
from __future__ import annotations

import json
from pathlib import Path

import yaml
from packaging.version import Version

from firmware_audit.collect_firmware import collect_firmware_inventory


def evaluate() -> list[dict]:
    inventory = collect_firmware_inventory()
    policy = yaml.safe_load(Path("firmware_audit/policy.yaml").read_text(encoding="utf-8"))
    findings: list[dict] = []

    host_policy = policy["hardware_policies"]["host_bios"]
    if inventory["bios_vendor"] != host_policy["vendor"]:
        findings.append(
            {
                "asset": inventory["hostname"],
                "control": "host_bios",
                "status": "unapproved_vendor",
                "observed": inventory["bios_vendor"],
                "required_vendor": host_policy["vendor"],
            }
        )
    elif Version(inventory["bios_version"]) < Version(host_policy["min_version"]):
        findings.append(
            {
                "asset": inventory["hostname"],
                "control": "host_bios",
                "status": "out_of_policy",
                "observed": inventory["bios_version"],
                "required_min": host_policy["min_version"],
            }
        )

    gpu_policy = policy["hardware_policies"]["gpu_vbios"]
    for gpu in inventory["gpus"]:
        expected = gpu_policy.get(gpu["name"])
        if expected is None:
            findings.append(
                {
                    "asset": gpu["serial"],
                    "control": "gpu_vbios",
                    "status": "unapproved_gpu_model",
                    "observed": gpu["name"],
                }
            )
            continue
        if gpu["vbios_version"] in expected.get("blocked_versions", []):
            findings.append(
                {
                    "asset": gpu["serial"],
                    "control": "gpu_vbios",
                    "status": "blocked_version",
                    "observed": gpu["vbios_version"],
                }
            )
            continue
        if Version(gpu["vbios_version"]) < Version(expected["min_version"]):
            findings.append(
                {
                    "asset": gpu["serial"],
                    "control": "gpu_vbios",
                    "status": "out_of_policy",
                    "observed": gpu["vbios_version"],
                    "required_min": expected["min_version"],
                }
            )

    return findings


if __name__ == "__main__":
    print(json.dumps(evaluate(), indent=2))

Step 3: Fence and drain out-of-policy nodes

For Kubernetes-backed AI clusters, taint and drain nodes immediately so no new workloads land on compromised or outdated hardware.

# Fence an out-of-policy node
kubectl taint nodes ai-gpu-node-17 aidefend/quarantine=true:NoSchedule
kubectl cordon ai-gpu-node-17
kubectl drain ai-gpu-node-17 --ignore-daemonsets --delete-emptydir-data

Action: Run firmware collection during hardware onboarding, re-admission after maintenance, and supplier-image refresh. Compare the result against a version-controlled admission policy, then automatically taint or quarantine nodes that fall below the approved firmware floor. Ongoing vulnerability-monitoring cadence and patch rollout remain owned by AID-H-009.002.

AID-H-003.004-G004

Concept:

This is an advanced, proactive defense for physically accessible edge devices. The device intentionally tries to attack itself with low-level techniques to see if its defenses are working. For example, it might briefly undervolt the CPU (fault injection) during a cryptographic calculation to see if the calculation produces an error or a correctable fault, rather than a silently incorrect (and potentially exploitable) result.

Develop Self-Test Routines

This requires deep hardware and embedded systems expertise. The code is highly specific to the System-on-a-Chip (SoC) and its capabilities. Conceptually, it involves using privileged kernel modules or direct hardware register access to manipulate clock speeds, voltage, or EM emissions while a sensitive operation is running.

Target-specific self-test contract

1. Use a vendor-approved fault-injection fixture and an immutable known-answer test vector for the exact SoC, firmware, voltage domain, and cryptographic implementation.
2. Run a clean baseline first; abort if the baseline output does not match.
3. Apply only the reviewed voltage, clock, or electromagnetic fault profile inside a maintenance window with an independent hardware safety interlock.
4. Require the device to reject the operation, reset into a known state, or return the exact known answer. A silently corrupted output is FAIL.
5. Restore nominal settings and independently read back the voltage, clock, boot, and tamper state before returning the device to service.

Evidence

Preserve the SoC and board revision, firmware and cryptographic-library digests, fixture calibration, signed fault profile, known-answer vector digest, raw oscilloscope or glitch-controller trace, device telemetry, restore readback, and explicit PASS/FAIL/ERROR verdict. This procedure is NOT_APPLICABLE when the approved board cannot expose a bounded fault-injection interface; do not substitute unreviewed register writes for a vendor-specific harness.

Action: For high-stakes edge AI devices where physical tampering is a credible threat, partner with hardware security experts to develop self-test routines. These routines should be scheduled to run automatically during maintenance windows or on-demand as part of a comprehensive device health check.

Keep the action safely bounded

quarantine the node and deny scheduling when evidence is stale, replayed, incomplete, unverifiable, or outside signed admission policy.

Executable vendor-adapter contract

# File: hardware_fault_selftest/runner.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class FaultFixture:
    fixture_id: str
    board_revision: str
    firmware_sha256: str
    known_answer: bytes


def run(
    fixture: FaultFixture,
    *,
    read_identity: Callable[[], tuple[str, str]],
    clean_operation: Callable[[], bytes],
    inject_reviewed_fault: Callable[[], bytes],
    restore_and_readback: Callable[[], bool],
) -> dict:
    board, firmware = read_identity()
    if (board, firmware) != (fixture.board_revision, fixture.firmware_sha256):
        raise PermissionError("device identity differs from signed fixture")
    if clean_operation() != fixture.known_answer:
        raise RuntimeError("clean known-answer baseline failed")
    observed = inject_reviewed_fault()
    restored = restore_and_readback()
    if not restored:
        raise RuntimeError("nominal hardware state was not restored")
    silently_corrupted = observed not in {fixture.known_answer, b"REJECTED", b"RESET"}
    return {
        "fixture_id": fixture.fixture_id,
        "silently_corrupted": silently_corrupted,
        "outcome": "FAIL" if silently_corrupted else "PASS",
    }

The four callables are implemented only by the reviewed board/fixture adapter; they must not expose arbitrary register writes. Run from a maintenance controller with a physical safety interlock, capture fixture telemetry independently, and quarantine on any exception or silent corruption.

Open-source tools

• U-Boot
• binwalk
• Intel SGX SDK
• Open Enclave SDK
• Keylime (nonce-bound TPM attestation and verifier policy enforcement)
• tpm2-tools (TPM quote generation and verification only; asset identity, measurement policy, nonce issuance, and cluster admission remain external)

Commercial / hosted tools

• NVIDIA Confidential Computing
• Azure Confidential Computing
• Google Cloud Confidential Computing

Threat mappings

MITRE ATLAS

• AML.T0010 AI Supply Chain Compromise (hardware is part of the broader AI supply chain)
• AML.T0010.000 AI Supply Chain Compromise: Hardware

MAESTRO

• Supply Chain Attacks (Cross-Layer)

OWASP LLM Top 10 2026

• LLM04:2026 Supply Chain

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

• AITech-14.1 Unauthorized Access (Attests the underlying host before admitting the AI workload)
• AISubtech-14.1.2 Insufficient Access Controls (fresh platform identity and measured-state admission before workload credential release)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• N/A