AID-D-002.003: White-Box Transformer Internal-Signal Anomaly Telemetry

Version 1.20261015 · Actionable sub-technique

Instrument self-hosted or open-weight Transformer inference stacks to monitor internal-signal anomalies such as abnormal attention concentration, activation shifts, entropy collapse, and attention isolation. Bind every probe and baseline to the exact model digest, layer or hook, runtime image, framework, kernel path, numeric precision, quantization, and compilation mode. This sub-technique applies only when those internals are observable at runtime.

Pillars: model

Lifecycle: operation

Scope boundary

Owns white-box runtime anomaly telemetry from self-hosted or open-weight Transformer internals such as attention tensors, logits, entropy, and activations, including configuration-bound calibration and transfer limits. Provider-returned top-k log probabilities remain output-distribution telemetry under AID-D-002.001.

AID-D-002.004: AID-D-002.003 observes model-internal tensors and scores inside a white-box inference stack; AID-D-002.004 observes provider or API-visible reasoning metadata, trace summaries, and thinking-token accounting. Internal attention or logit telemetry and output-side reasoning telemetry are different observability surfaces.

Implementation guidance

AID-D-002.003-G001

Detect anomalous attention patterns in Transformer-based models.
Reusable attention-telemetry evaluator

Instrument a pinned self-hosted runtime to emit the exact declared layer and head tensors plus the valid-key mask. Do not infer a clean result from a missing layer, unsupported attention implementation, or comment-only hook. Compare each measured head against a signed layer/head baseline.

from __future__ import annotations

# File: detection/attention_entropy.py

import math

import torch


def masked_entropy(attention: torch.Tensor, valid_key_mask: torch.Tensor) -> torch.Tensor:
    if attention.ndim != 4:
        raise ValueError("attention must have [batch, heads, query, key] shape")
    if valid_key_mask.ndim != 2 or valid_key_mask.shape != (attention.shape[0], attention.shape[3]):
        raise ValueError("valid-key mask does not match attention tensor")
    mask = valid_key_mask[:, None, None, :].to(dtype=torch.bool, device=attention.device)
    probabilities = attention.to(dtype=torch.float64).masked_fill(~mask, 0.0)
    totals = probabilities.sum(dim=-1, keepdim=True)
    if torch.any(totals <= 0):
        raise ValueError("attention row has no valid probability mass")
    probabilities = probabilities / totals
    entropy = -(probabilities * torch.log2(probabilities.clamp_min(1e-12))).sum(dim=-1)
    return entropy.mean(dim=(0, 2))


def evaluate_attention_capture(
    request_id: str,
    attention_by_layer: dict[int, torch.Tensor],
    valid_key_mask: torch.Tensor,
    policy: dict,
) -> dict:
    required = {
        "policy_version", "expected_layers", "expected_heads_by_layer",
        "baseline_entropy_by_layer_head", "minimum_entropy_ratio",
    }
    if set(policy) != required:
        raise ValueError("attention policy schema mismatch")
    expected_layers = [int(value) for value in policy["expected_layers"]]
    missing_layers = sorted(set(expected_layers) - set(attention_by_layer))
    extra_layers = sorted(set(attention_by_layer) - set(expected_layers))
    if missing_layers or extra_layers:
        return {
            "control_status": "INSUFFICIENT_DATA",
            "finding_status": "UNKNOWN",
            "request_id": request_id,
            "missing_layers": missing_layers,
            "unexpected_layers": extra_layers,
            "policy_version": policy["policy_version"],
        }
    measurements = []
    findings = []
    for layer in expected_layers:
        tensor = attention_by_layer[layer]
        expected_heads = int(policy["expected_heads_by_layer"][str(layer)])
        if tensor.shape[1] != expected_heads:
            return {
                "control_status": "INSUFFICIENT_DATA",
                "finding_status": "UNKNOWN",
                "request_id": request_id,
                "reason": "head count mismatch",
                "layer": layer,
                "observed_heads": int(tensor.shape[1]),
                "expected_heads": expected_heads,
                "policy_version": policy["policy_version"],
            }
        entropies = masked_entropy(tensor, valid_key_mask)
        for head, value in enumerate(entropies.tolist()):
            key = f"{layer}:{head}"
            baseline = float(policy["baseline_entropy_by_layer_head"][key])
            if not math.isfinite(value) or baseline <= 0:
                raise ValueError("attention entropy or baseline is invalid")
            ratio = value / baseline
            finding = ratio < float(policy["minimum_entropy_ratio"])
            measurements.append({
                "layer": layer, "head": head,
                "entropy": round(value, 8),
                "baseline_entropy": baseline,
                "entropy_ratio": round(ratio, 8),
                "finding": finding,
            })
            if finding:
                findings.append(key)
    return {
        "schema_version": "aidefend.attention_entropy.v1",
        "control_status": "PASS",
        "finding_status": "FINDING" if findings else "NO_FINDING",
        "request_id": request_id,
        "policy_version": policy["policy_version"],
        "finding_layer_heads": findings,
        "measurements": measurements,
    }

Verification: Replay clean, entropy-collapse, masked-padding, missing-layer, wrong-head-count, empty-mask, unsupported-runtime, and malformed-tensor fixtures through the actual hook. Compare every required layer and head against the signed baseline and retain capture digests.

Action: Emit authenticated tensor reductions from the pinned runtime and evaluate all declared layers and heads; a missing hook or tensor is INSUFFICIENT_DATA, never NO_FINDING.

AID-D-002.003-G002

Monitor backdoor-specific internal telemetry (entropy collapse and attention isolation) in sampled/high-audit runtime flows.
Concept:

Some sleeper-agent backdoors exhibit distinctive internal signatures when a trigger activates: outputs become unusually certain (entropy collapse) and attention concentrates disproportionately inside a small trigger span (attention isolation). Treat these as corroborating telemetry, not as standalone proof.

Prerequisite:

This strategy requires a white-box or semi-white-box inference stack that can emit generation scores and, optionally, attention tensors. If attention tensors are unavailable, run the entropy-collapse path only.

Step 1: Baseline internal signals per model version

Store known-good output entropy and attention-isolation baselines for each production model version. Thresholds must be calibrated per version, not globally.

Step 2: Sample and score high-risk flows
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")
BACKDOOR_TELEMETRY_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D002003_BACKDOOR_TELEMETRY_COMMAND_TIMEOUT_SECONDS")

# File: detection/backdoor_internal_telemetry.py

import json
import math
import os
import subprocess
from dataclasses import dataclass
from pathlib import Path
import hmac
import re
import tempfile
from typing import Dict, Optional, Tuple

import numpy as np
import torch


@dataclass(frozen=True)
class Thresholds:
    min_entropy_ratio: float
    max_isolation_multiplier: float


@dataclass(frozen=True)
class GenerationSample:
    scores: Tuple[torch.Tensor, ...]
    attentions: Optional[torch.Tensor] = None


def mean_token_entropy(scores: Tuple[torch.Tensor, ...]) -> float:
    entropies = []
    for step_logits in scores:
        probs = torch.softmax(step_logits[0], dim=-1)
        entropies.append(float((-torch.sum(probs * torch.log(probs + 1e-12))).item()))
    return float(np.mean(entropies)) if entropies else 0.0


def attention_isolation(attn: torch.Tensor, trigger_span: Tuple[int, int]) -> float:
    if attn.dim() == 5:
        attn = attn[-1]
    averaged = attn[0].mean(dim=0)
    start, end = trigger_span
    within_trigger = averaged[start:end, start:end].mean().item()
    prompt_to_trigger = averaged[:start, start:end].mean().item() if start > 0 else 0.0
    return float(within_trigger / (prompt_to_trigger + 1e-8))


def load_generation_sample(scores_path: Path, attentions_path: Optional[Path]) -> GenerationSample:
    scores = tuple(torch.load(scores_path, map_location="cpu", weights_only=True))
    attentions = None
    if attentions_path and attentions_path.exists():
        attentions = torch.load(attentions_path, map_location="cpu", weights_only=True)
    return GenerationSample(scores=scores, attentions=attentions)


def score_request(
    sample: GenerationSample,
    baselines: Dict[str, float],
    thresholds: Thresholds,
    trigger_span: Optional[Tuple[int, int]],
) -> Dict[str, object]:
    output_entropy = mean_token_entropy(sample.scores)
    flags = []

    if output_entropy < baselines["output_entropy"] * thresholds.min_entropy_ratio:
        flags.append("entropy_collapse")

    isolation_score = None
    if trigger_span and sample.attentions is not None:
        isolation_score = attention_isolation(sample.attentions, trigger_span)
        if isolation_score >= baselines.get("attention_isolation", 1.0) * thresholds.max_isolation_multiplier:
            flags.append("attention_isolation")

    return {
        "flags": flags,
        "metrics": {
            "output_entropy": round(output_entropy, 4),
            "attention_isolation": None if isolation_score is None else round(isolation_score, 4),
        },
    }


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(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 load_signed_json(path: Path, key_environment_variable: str) -> dict:
    signature_path = Path(str(path) + ".sig")
    payload_raw = path.read_bytes()
    signature_raw = signature_path.read_bytes()
    key_reference = os.environ[key_environment_variable]
    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-d002003-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        signature_snapshot = root / "signature"
        _write_private_snapshot(payload_snapshot, payload_raw)
        _write_private_snapshot(signature_snapshot, signature_raw)
        captured = [(payload_snapshot, payload_raw), (signature_snapshot, signature_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(signature_snapshot), str(payload_snapshot),
                ],
                check=True, capture_output=True, text=True,
                timeout=BACKDOOR_TELEMETRY_COMMAND_TIMEOUT_SECONDS,
            )
            if any(
                not hmac.compare_digest(candidate.read_bytes(), expected)
                for candidate, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
            value = _strict_json(payload_raw)
        finally:
            for candidate, _ in captured:
                os.chmod(candidate, 0o600)
    if not isinstance(value, dict):
        raise ValueError(f"{path} must contain a JSON object")
    return value


def positive_finite(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 positive and finite")
    return result


def main() -> None:
    try:
        policy = load_signed_json(
            Path("policy/backdoor-internal-signal-policy.json"),
            "D002003_POLICY_VERIFY_KEY",
        )
        baselines = load_signed_json(
            Path("baselines/internal-signal-baselines.json"),
            "D002003_BASELINE_VERIFY_KEY",
        )
        required_policy_fields = {
            "schema_version", "policy_version", "model_version",
            "baseline_version", "min_entropy_ratio",
            "max_isolation_multiplier", "scores_path",
            "attentions_path", "trigger_span",
        }
        if set(policy) != required_policy_fields or policy["schema_version"] != "aidefend.d002003-policy.v1":
            raise ValueError("internal-signal policy schema mismatch")
        if (
            set(baselines) != {"schema_version", "baseline_version", "models"}
            or baselines["schema_version"] != "aidefend.d002003-baseline.v1"
            or baselines["baseline_version"] != policy["baseline_version"]
            or not isinstance(baselines["models"], dict)
        ):
            raise ValueError("baseline schema or policy binding mismatch")
        model_version = policy["model_version"]
        if not isinstance(model_version, str) or model_version not in baselines["models"]:
            raise ValueError("model version is absent from signed baseline")
        model_baseline = baselines["models"][model_version]
        if (
            not isinstance(model_baseline, dict)
            or set(model_baseline) != {"output_entropy", "attention_isolation"}
        ):
            raise ValueError("model baseline fields differ")
        normalized_baseline = {
            key: positive_finite(value, key)
            for key, value in model_baseline.items()
        }
        thresholds = Thresholds(
            min_entropy_ratio=positive_finite(
                policy["min_entropy_ratio"], "min_entropy_ratio"
            ),
            max_isolation_multiplier=positive_finite(
                policy["max_isolation_multiplier"],
                "max_isolation_multiplier",
            ),
        )
        trigger_span_value = policy["trigger_span"]
        if trigger_span_value is None:
            trigger_span = None
        elif (
            not isinstance(trigger_span_value, list)
            or len(trigger_span_value) != 2
            or any(isinstance(item, bool) or not isinstance(item, int) for item in trigger_span_value)
            or trigger_span_value[0] < 0
            or trigger_span_value[1] <= trigger_span_value[0]
        ):
            raise ValueError("trigger_span is invalid")
        else:
            trigger_span = (trigger_span_value[0], trigger_span_value[1])
        scores_path = Path(policy["scores_path"])
        attentions_path = (
            None
            if policy["attentions_path"] is None
            else Path(policy["attentions_path"])
        )
        sample = load_generation_sample(scores_path, attentions_path)
        result = score_request(
            sample, normalized_baseline, thresholds, trigger_span
        )
        result = {
            "schema_version": "aidefend.d002003-result.v1",
            "control_status": "PASS",
            "finding_status": "FINDING" if result["flags"] else "NO_FINDING",
            "model_version": model_version,
            "policy_version": policy["policy_version"],
            "baseline_version": policy["baseline_version"],
            **result,
        }
    except Exception as error:
        result = {
            "schema_version": "aidefend.d002003-result.v1",
            "control_status": "ERROR",
            "finding_status": "UNKNOWN",
            "error_class": type(error).__name__,
            "reason": "signed policy, baseline, or measurement processing failed",
        }
    print(json.dumps(result, sort_keys=True, separators=(",", ":")))


if __name__ == "__main__":
    main()
Step 3: Treat hits as high-audit routing signals

If signals cross thresholds, mark the session as suspected backdoor activation, downgrade capabilities, and trigger offline model re-vetting. Keep the default response non-blocking unless multiple independent indicators agree.

Action: Use this telemetry only on sampled or high-risk flows where white-box introspection is operationally acceptable. Persist the model version, baseline version, threshold set, and raw metric values so responders can reproduce the decision later.

AID-D-002.003-G003

Calibrate each activation, attention, or logit probe for the exact model and inference stack that produces the measured tensor.
Make probe transfer an explicit compatibility decision

Create a signed probe manifest with validated digest fields plus structured tensor_contract (shape, semantic, reduction) and quantization (method, config_sha256) records. Bind the model artifact digest, tokenizer digest, layer and hook names, tensor shape and semantic meaning, serving-image digest, framework and kernel versions, device type, numeric precision, quantization method and parameters, compiler mode, adapter set, probe artifact digest, calibration corpus digest, and threshold policy. Admission must compare every field with live readback before enabling the probe.

# File: detection/verify_probe_binding.py
from __future__ import annotations

import hashlib
import json
import re
from collections.abc import Mapping

SHA256 = re.compile(r"^[a-f0-9]{64}$")
IMAGE = re.compile(r"^[^\s]+@sha256:[a-f0-9]{64}$")
SHA_FIELDS = {"probe_sha256", "model_sha256", "tokenizer_sha256",
              "adapter_set_sha256", "calibration_corpus_sha256"}
TEXT_FIELDS = {"probe_id", "layer_name", "hook_name", "framework_version",
               "kernel_version", "device_type", "dtype", "compiler_mode",
               "policy_version"}
FIELDS = {
    "schema_version", "probe_id", "probe_sha256", "model_sha256",
    "tokenizer_sha256", "layer_name", "hook_name", "tensor_contract",
    "serving_image_digest", "framework_version", "kernel_version",
    "device_type", "dtype", "quantization", "compiler_mode",
    "adapter_set_sha256", "calibration_corpus_sha256", "policy_version",
}


def validate_binding(value: dict) -> None:
    if not isinstance(value, dict) or set(value) != FIELDS:
        raise ValueError("probe binding schema differs")
    if value["schema_version"] != "aidefend.probe-binding.v1":
        raise ValueError("unsupported probe binding schema")
    if any(not isinstance(value[field], str) or not value[field].strip()
           for field in TEXT_FIELDS):
        raise ValueError("probe binding text field is empty")
    if any(not isinstance(value[field], str) or SHA256.fullmatch(value[field]) is None
           for field in SHA_FIELDS):
        raise ValueError("probe binding digest is invalid")
    if not isinstance(value["serving_image_digest"], str) or IMAGE.fullmatch(value["serving_image_digest"]) is None:
        raise ValueError("serving image is not digest-pinned")
    tensor = value["tensor_contract"]
    if (not isinstance(tensor, dict) or set(tensor) != {"shape", "semantic", "reduction"}
            or not isinstance(tensor["shape"], list) or not tensor["shape"]
            or any((type(item) is not int or item < 1) and
                   (not isinstance(item, str) or re.fullmatch(r"[a-z][a-z0-9_]{0,31}", item) is None)
                   for item in tensor["shape"])
            or any(not isinstance(tensor[field], str) or not tensor[field]
                   for field in ("semantic", "reduction"))):
        raise ValueError("tensor contract is invalid")
    quantization = value["quantization"]
    if (not isinstance(quantization, dict)
            or set(quantization) != {"method", "config_sha256"}
            or not isinstance(quantization["method"], str)
            or not quantization["method"]):
        raise ValueError("quantization binding is invalid")
    config_sha256 = quantization["config_sha256"]
    if ((quantization["method"] == "none" and config_sha256 is not None)
            or (quantization["method"] != "none" and
                (not isinstance(config_sha256, str) or SHA256.fullmatch(config_sha256) is None))):
        raise ValueError("quantization configuration digest is invalid")


def canonical_sha256(document: Mapping) -> str:
    raw = json.dumps(document, sort_keys=True, separators=(",", ":"),
                     ensure_ascii=False, allow_nan=False).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def verify_probe_binding(approved: dict, observed: dict) -> dict:
    validate_binding(approved)
    validate_binding(observed)
    mismatches = sorted(field for field in FIELDS if approved[field] != observed[field])
    return {"schema_version": "aidefend.probe-binding-result.v1",
            "control_status": "PASS",
            "finding_status": "FINDING" if mismatches else "NO_FINDING",
            "approved_binding_sha256": canonical_sha256(approved),
            "observed_binding_sha256": canonical_sha256(observed),
            "mismatched_fields": mismatches}

A probe trained or calibrated on FP16 is not valid evidence for an AWQ, GPTQ, GGUF, FP8, or differently compiled deployment unless a separate, versioned transfer evaluation proves equivalence for that exact configuration. A mismatch disables the probe or produces a finding under policy; absent live readback, an unknown hook, or incomplete tensor coverage is INSUFFICIENT_DATA or ERROR, never PASS. Retain signed manifests, calibration results, complete sampled population, and independently replayed feature and threshold outputs.

Action: Reject probe evidence whenever any approved binding differs, and recalibrate after a model, tokenizer, adapter, runtime, compiler, dtype, tensor, or quantization change.

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-D-002.003: White-Box Transformer Internal-Signal Anomaly Telemetry | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service