Scope boundary

Owns anomaly findings from AI accelerator power, temperature, utilization, and performance-counter telemetry against representative workload baselines, including cryptomining, denial-of-service, and side-channel-probing signals.

AID-D-002: AID-D-005.005 detects anomalies in GPU, TPU, NPU, or accelerator hardware telemetry; AID-D-002 detects drift and anomalies in model inputs, outputs, labeled performance, or model-internal signals. Hardware resource behavior and model behavioral drift may correlate but have different baselines and observation points.

Implementation guidance

• AID-D-005.005-G001: Establish statistical baselines (mean/std) under representative workloads; compare live metrics and alert on 3-sigma deviations.

AID-D-005.005-G001

Concept:

To detect what is abnormal, first define workload-specific normal. A single global baseline is too noisy for mixed inference, batch, and training clusters. Build a baseline per accelerator class and workload profile, persist mean/stddev values, then compare sustained live telemetry against that profile instead of against ad hoc dashboard intuition.

Step 1: Build a workload baseline from Prometheus/DCGM data

Collect accelerator telemetry through NVIDIA DCGM Exporter or equivalent and store it in Prometheus. Train baselines only from known-good windows that represent the workload you actually want to protect.

# File: accelerator_telemetry/baseline_policy.schema.yaml
schema_version: aidefend.accelerator-baseline-policy-schema.v1
signed_policy_required_fields:
  - schema_version
  - policy_version
  - prometheus_url
  - baseline_window_seconds
  - query_step_seconds
  - minimum_samples
  - upper_sigma
  - minimum_scale
  - alert_after_consecutive_breaches
  - request_timeout_seconds
  - maximum_sample_age_seconds
  - maximum_future_skew_seconds
  - response_max_bytes
  - response_chunk_bytes
  - metrics
numeric_rule: finite policy values with the documented positive or non-negative domain
evidence_bindings:
  - policy_version
  - policy_sha256
  - workload_label
  - metric_population
  - source_window
  - sample_counts

from __future__ import annotations

# File: accelerator_telemetry/build_baseline.py
import hashlib
import json
import math
import os
import subprocess
from datetime import datetime, timedelta, timezone
from pathlib import Path
import hmac
import re
import tempfile
from urllib.parse import urlparse




def _aidefend_write_private(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 _aidefend_strict_json(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 _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str,
    timeout_seconds: float,
    proof_flag: str,
) -> bytes:
    if payload_path == proof_path or proof_flag != "--bundle":
        raise RuntimeError("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-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "proof"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(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"
            _aidefend_write_private(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 RuntimeError("private signature-verification snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)
    return payload_raw


import requests


def load_signed_accelerator_policy() -> tuple[dict, str]:
    policy_path = Path(os.environ["D005005_ACCELERATOR_POLICY_PATH"])
    bundle_path = Path(os.environ["D005005_ACCELERATOR_POLICY_SIGSTORE_BUNDLE"])
    verify_timeout = float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("signed bootstrap-profile verification timeout is invalid")
    raw = _aidefend_capture_signed_bytes(
        policy_path, bundle_path, os.environ["D005005_ACCELERATOR_POLICY_VERIFY_KEY"],
        verify_timeout, "--bundle",
    )
    policy = _aidefend_strict_json(raw)
    required = {
        "schema_version", "policy_version", "prometheus_url",
        "baseline_window_seconds", "query_step_seconds", "minimum_samples",
        "upper_sigma", "minimum_scale", "alert_after_consecutive_breaches",
        "request_timeout_seconds", "maximum_sample_age_seconds",
        "maximum_future_skew_seconds", "response_max_bytes",
        "response_chunk_bytes", "metrics",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise RuntimeError("signed accelerator policy fields differ")
    if (
        policy["schema_version"] != "aidefend.accelerator-baseline-policy.v1"
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
        or not isinstance(policy["prometheus_url"], str)
        or urlparse(policy["prometheus_url"]).scheme != "https"
    ):
        raise RuntimeError("signed accelerator policy identity or endpoint is invalid")
    for field in (
        "baseline_window_seconds", "query_step_seconds", "minimum_samples",
        "alert_after_consecutive_breaches", "response_max_bytes",
        "response_chunk_bytes",
    ):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise RuntimeError("signed accelerator integer is invalid: " + field)
    for field in (
        "upper_sigma", "minimum_scale", "request_timeout_seconds",
        "maximum_sample_age_seconds",
    ):
        value = policy[field]
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
            or float(value) <= 0
        ):
            raise RuntimeError("signed accelerator bound is invalid: " + field)
    if (policy["response_max_bytes"] < 2
            or policy["response_chunk_bytes"] > policy["response_max_bytes"]):
        raise RuntimeError("signed Prometheus response bounds are invalid")
    future_skew = policy["maximum_future_skew_seconds"]
    if (
        isinstance(future_skew, bool)
        or not isinstance(future_skew, (int, float))
        or not math.isfinite(float(future_skew))
        or float(future_skew) < 0
    ):
        raise RuntimeError("signed accelerator future-skew bound is invalid")
    metrics = policy["metrics"]
    if (
        not isinstance(metrics, list)
        or not metrics
        or any(not isinstance(metric, str) or not metric for metric in metrics)
        or len(metrics) != len(set(metrics))
    ):
        raise RuntimeError("signed accelerator metric population is invalid")
    return policy, "sha256:" + hashlib.sha256(raw).hexdigest()


POLICY, POLICY_SHA256 = load_signed_accelerator_policy()


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


def read_prometheus_json(response):
    try:
        if response.headers.get("Content-Encoding", "identity").lower() != "identity":
            raise RuntimeError("encoded Prometheus response is forbidden")
        declared = response.headers.get("Content-Length")
        maximum = policy_maximum = POLICY["response_max_bytes"]
        if declared is not None and (not declared.isdigit() or int(declared) > maximum):
            raise RuntimeError("Prometheus Content-Length is invalid")
        chunks, observed = [], 0
        for chunk in response.iter_content(chunk_size=POLICY["response_chunk_bytes"]):
            observed += len(chunk)
            if observed > policy_maximum:
                raise RuntimeError("Prometheus response exceeds signed bound")
            chunks.append(chunk)
        if declared is not None and int(declared) != observed:
            raise RuntimeError("Prometheus Content-Length differs from observed bytes")
        return 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)))
    finally:
        response.close()


def query_range(query: str, start: datetime, end: datetime) -> list[float]:
    response = requests.get(
        POLICY["prometheus_url"].rstrip("/") + "/api/v1/query_range",
        params={
            "query": query,
            "start": start.isoformat(),
            "end": end.isoformat(),
            "step": str(POLICY["query_step_seconds"]) + "s",
        },
        timeout=float(POLICY["request_timeout_seconds"]),
        allow_redirects=False,
        stream=True,
        headers={"Accept-Encoding": "identity"},
    )
    response.raise_for_status()
    payload = read_prometheus_json(response)
    if payload.get("status") != "success":
        raise RuntimeError("Prometheus range query did not succeed")
    series = payload.get("data", {}).get("result")
    if not isinstance(series, list):
        raise RuntimeError("Prometheus range result schema differs")
    values: list[float] = []
    for item in series:
        points = item.get("values") if isinstance(item, dict) else None
        if not isinstance(points, list):
            raise RuntimeError("Prometheus range series schema differs")
        for point in points:
            if not isinstance(point, list) or len(point) != 2:
                raise RuntimeError("Prometheus sample schema differs")
            value = float(point[1])
            if not math.isfinite(value):
                raise RuntimeError("Prometheus sample is non-finite")
            values.append(value)
    return values


def build_baseline(workload_label: str, output_path: str) -> dict:
    if not isinstance(workload_label, str) or not workload_label:
        raise ValueError("trusted workload label is required")
    end = datetime.now(timezone.utc)
    start = end - timedelta(seconds=POLICY["baseline_window_seconds"])
    baseline = {
        "schema_version": "aidefend.accelerator-baseline.v1",
        "policy_version": POLICY["policy_version"],
        "policy_sha256": POLICY_SHA256,
        "workload_label": workload_label,
        "window_start": start.isoformat(),
        "window_end": end.isoformat(),
        "metrics": {},
    }
    for metric in POLICY["metrics"]:
        values = query_range(
            metric + '{workload="' + workload_label + '"}',
            start,
            end,
        )
        if len(values) < POLICY["minimum_samples"]:
            raise ValueError("metric sample population is below signed policy: " + metric)
        mean = math.fsum(values) / len(values)
        variance = math.fsum((value - mean) ** 2 for value in values) / len(values)
        baseline["metrics"][metric] = {
            "mean": mean,
            "stdev": math.sqrt(variance),
            "sample_count": len(values),
        }
    target = Path(output_path)
    target.write_text(
        json.dumps(baseline, sort_keys=True, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )
    if json.loads(target.read_text(encoding="utf-8")) != baseline:
        raise RuntimeError("accelerator baseline readback differs")
    return baseline

Step 2: Alert on sustained policy-threshold deviations

Evaluate live values against the saved baseline and require multiple consecutive breaches before alerting. This reduces false positives from bursty but normal accelerator usage.

# Continuing in accelerator_telemetry/build_baseline.py
def latest_value(query: str) -> tuple[float, float]:
    response = requests.get(
        POLICY["prometheus_url"].rstrip("/") + "/api/v1/query",
        params={"query": query},
        timeout=float(POLICY["request_timeout_seconds"]),
        allow_redirects=False,
        stream=True,
        headers={"Accept-Encoding": "identity"},
    )
    response.raise_for_status()
    payload = read_prometheus_json(response)
    result = payload.get("data", {}).get("result")
    if payload.get("status") != "success" or not isinstance(result, list) or len(result) != 1:
        raise ValueError("instant query must return exactly one series")
    point = result[0].get("value")
    if not isinstance(point, list) or len(point) != 2:
        raise ValueError("instant-query sample schema differs")
    observed_at = float(point[0])
    value = float(point[1])
    age = datetime.now(timezone.utc).timestamp() - observed_at
    if (
        not math.isfinite(value)
        or not math.isfinite(age)
        or age > float(POLICY["maximum_sample_age_seconds"])
        or age < -float(POLICY["maximum_future_skew_seconds"])
    ):
        raise ValueError("instant-query sample is non-finite or outside signed freshness")
    return value, observed_at


def detect_gpu_anomalies(
    workload_label: str,
    baseline: dict,
    consecutive_breaches: dict[str, int],
) -> dict:
    if (
        baseline.get("policy_version") != POLICY["policy_version"]
        or baseline.get("policy_sha256") != POLICY_SHA256
        or baseline.get("workload_label") != workload_label
        or set(baseline.get("metrics", {})) != set(POLICY["metrics"])
    ):
        raise RuntimeError("accelerator baseline binding or metric population differs")
    findings = []
    observations = []
    for metric in POLICY["metrics"]:
        live, observed_at = latest_value(
            metric + '{workload="' + workload_label + '"}'
        )
        stats = baseline["metrics"][metric]
        scale = max(float(stats["stdev"]), float(POLICY["minimum_scale"]))
        threshold = float(stats["mean"]) + float(POLICY["upper_sigma"]) * scale
        if live > threshold:
            consecutive_breaches[metric] = consecutive_breaches.get(metric, 0) + 1
        else:
            consecutive_breaches[metric] = 0
        observation = {
            "metric": metric,
            "live": live,
            "observed_at_unix": observed_at,
            "threshold": threshold,
            "consecutive_breaches": consecutive_breaches[metric],
        }
        observations.append(observation)
        if (
            consecutive_breaches[metric]
            >= POLICY["alert_after_consecutive_breaches"]
        ):
            findings.append(observation)
    return {
        "schema_version": "aidefend.accelerator-detection-result.v1",
        "control": "AID-D-005.005",
        "policy_version": POLICY["policy_version"],
        "policy_sha256": POLICY_SHA256,
        "workload_label": workload_label,
        "expected_metric_count": len(POLICY["metrics"]),
        "evaluated_metric_count": len(observations),
        "observations": observations,
        "findings": findings,
        "finding_status": "FINDING" if findings else "NO_FINDING",
    }

Step 3: Integrate with the response path

Alert to SIEM, but keep any automated kill or throttle action behind an explicit policy threshold because accelerator spikes can still be workload-driven. Log the workload label, node identity, and breached metric so responders can distinguish cryptomining, DoS, and side-channel probing.

Action: Baseline accelerator metrics under representative workloads, persist those statistics by workload class, and alert on sustained policy-threshold deviations rather than on one-off raw spikes.

Verify safely

An independent verifier must query fleet and scheduler inventories separately, replay raw metric samples with the documented aggregation method, and compare device coverage, missing intervals, baseline lookup, scores, and findings, read evidence back, verify signatures, digests, and immutability, and compare the canonical result digest.

Open-source tools

• NVIDIA DCGM Exporter
• AMD Device Metrics Exporter
• Prometheus
• Grafana

Commercial / hosted tools

• Datadog
• New Relic
• Splunk Observability Cloud

Threat mappings

MITRE ATLAS

• AML.T0029 Denial of AI Service
• AML.T0034 Cost Harvesting

MAESTRO

• Resource Hijacking (L4)
• Denial of Service (DoS) Attacks (L1) (DoS on foundation models manifests as GPU/TPU telemetry spikes)
• Denial of Service (DoS) Attacks (L4) (infrastructure-level DoS is directly detectable via accelerator telemetry)

OWASP LLM Top 10 2026

• LLM02:2026 Sensitive Information Disclosure (accelerator anomalies expose cache and performance-counter side-channel probing)
• LLM06:2026 Unbounded Consumption

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI08:2026 Cascading Failures (telemetry anomalies indicate cascading resource exhaustion)

NIST Adversarial Machine Learning 2025

• NISTAML.014 Energy-latency

Cisco Integrated AI Security and Safety Framework

• AISubtech-13.1.1 Compute Exhaustion (sustained GPU-utilization anomalies within the measured workload)
• AISubtech-13.2.1 Service Misuse for Cost Inflation (sustained GPU-utilization anomalies within the measured workload)
• AITech-13.1 Disruption of Availability (sustained GPU-utilization anomalies within the measured workload)
• AITech-13.2 Cost Harvesting / Repurposing (sustained GPU-utilization anomalies within the measured workload)

Google Secure AI Framework 2.0 - Risks

• DMS: Denial of ML Service

Databricks AI Security Framework 3.0

• Model Serving - Inference requests 9.7: Denial of Service (DoS)
• Agents - Core 13.4: Resource Overload