Scope boundary

This technique owns training-time changes that make a model measurably more resistant to declared adversarial-input and model-level evasion threats, including adversarial-example training, robust objectives, and architecture-specific regularization, together with threat-specific pass criteria verified by independent replay. Architecture or representation choices such as relative position bias or rotary positional embeddings do not satisfy this technique unless they are independently shown to improve the declared robustness outcome. It does not own evaluation-data integrity, instruction-hierarchy protection against prompt injection, or runtime authorization of tools and side effects.

AID-H-007.004: AID-H-001 changes the model to improve adversarial robustness; AID-H-007.004 protects the evaluation evidence used to substantiate that robustness. AID-H-007.004 covers the integrity, independence, sufficiency, and promotion use of datasets, holdouts, benchmarks, and test evidence. It applies whether or not adversarial training was used, and AID-H-001 cannot compensate for manipulated or insufficient evaluation evidence.

AID-H-016: AID-H-001 targets adversarial examples and model-evasion threats; AID-H-016 protects instruction precedence against prompt injection. Both can involve model training, but AID-H-016 separates trusted instructions from untrusted data and teaches instruction-following systems to preserve the instruction hierarchy. Neither technique substitutes for the other.

AID-H-018: AID-H-001 improves model robustness; AID-H-018 authorizes and constrains runtime tool actions and side effects. AID-H-001 does not authorize tool calls, constrain parameters, or make side effects transaction-safe. AID-H-018 remains necessary regardless of adversarial-training results.

Implementation guidance

• AID-H-001-G001: Implement general adversarial example training using a PGD-style, Friendly, or Free variant appropriate to the model and compute budget.
• AID-H-001-G002: For Reinforcement Learning, use a two-player, zero-sum game setup (RARL).
• AID-H-001-G003: Implement patch-based adversarial training for vision models.
• AID-H-001-G004: Employ masking or occlusion-based training as data augmentation.
• AID-H-001-G005: Apply a signed Transformer attention-regularization profile and independently gate the artifact on replayed robustness and task-quality evidence.

AID-H-001-G001

Concept:

For standard classification or sequence models facing evasion attacks, teams usually choose one primary adversarial-example training variant rather than implementing several in parallel. PGD-style training, Friendly Adversarial Training, and Free Adversarial Training all serve the same control objective: expose the model to adversarially perturbed inputs during training so it learns decision boundaries that are harder to exploit.

Choose a variant

• PGD-based training: strong default when robustness is the top priority and extra training cost is acceptable.
• Friendly Adversarial Training: use when you need a better balance between clean accuracy and adversarial robustness.
• Free Adversarial Training: use when compute budget is tight and you need a lower-cost approximation.

Load the verified model-family training profile

Before running any variant, a bootstrap identity must signature-verify the model family's versioned training profile and export these nine required fields: VERIFIED_H001_PGD_LEARNING_RATE, VERIFIED_H001_PGD_WEIGHT_DECAY, VERIFIED_H001_PGD_EPSILON, VERIFIED_H001_PGD_STEP_SIZE, VERIFIED_H001_PGD_MAX_ITERATIONS, VERIFIED_H001_FRIENDLY_KL_WEIGHT, VERIFIED_H001_FREE_REPLAY_COUNT, VERIFIED_H001_FREE_STEP_SIZE, and VERIFIED_H001_FREE_EPSILON. Resolve these model-family thresholds and compute budgets from that approved profile; AIDEFEND defines no universal values. Refuse to start when profile signature, version, digest, or any required export is missing. The runnable Python below validates every consumed value and rejects invalid relationships.

Variant A - PGD-based adversarial training

Generate strong iterative adversarial examples inside an Linf ball and train directly on them. The optimizer is defined once and reused by both the ART wrapper and the PyTorch training loop so the example can run as written.

# File: hardening/adversarial_training.py
import math
import os
import torch
import torch.optim as optim
from art.attacks.evasion import ProjectedGradientDescent
from art.estimators.classification import PyTorchClassifier

def _required_positive_float(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number")
    return value

PGD_LEARNING_RATE = _required_positive_float("VERIFIED_H001_PGD_LEARNING_RATE")
PGD_WEIGHT_DECAY = _required_positive_float("VERIFIED_H001_PGD_WEIGHT_DECAY")
PGD_EPSILON = _required_positive_float("VERIFIED_H001_PGD_EPSILON")
PGD_STEP_SIZE = _required_positive_float("VERIFIED_H001_PGD_STEP_SIZE")


def _required_positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be a positive integer")
    return value

PGD_MAX_ITERATIONS = _required_positive_int("VERIFIED_H001_PGD_MAX_ITERATIONS")


if PGD_STEP_SIZE > PGD_EPSILON:
    raise RuntimeError("verified PGD step size exceeds epsilon")


def train_with_pgd(*, model, device, loss_fn, input_shape, class_count, clip_values, train_loader):
    model = model.to(device)
    optimizer = optim.AdamW(model.parameters(), lr=PGD_LEARNING_RATE, weight_decay=PGD_WEIGHT_DECAY)
    
    art_classifier = PyTorchClassifier(
        model=model,
        loss=loss_fn,
        optimizer=optimizer,
        input_shape=input_shape,
        nb_classes=class_count,
        clip_values=clip_values,
        device_type="gpu" if device.type == "cuda" else "cpu",
    )
    
    pgd_attack = ProjectedGradientDescent(
        estimator=art_classifier,
        eps=PGD_EPSILON,
        eps_step=PGD_STEP_SIZE,
        max_iter=PGD_MAX_ITERATIONS,
        targeted=False,
    )
    
    for data, target in train_loader:
        model.train()
        data = data.to(device)
        target = target.to(device)
    
        adv_np = pgd_attack.generate(x=data.detach().cpu().numpy())
        adv_tensor = torch.from_numpy(adv_np).to(device=device, dtype=data.dtype)
    
        optimizer.zero_grad(set_to_none=True)
        logits = model(adv_tensor)
        loss = loss_fn(logits, target)
        loss.backward()
        optimizer.step()
    return model

Variant B - Friendly Adversarial Training

Add a KL-divergence term so predictions on adversarial inputs stay close to predictions on clean inputs, which helps preserve benign accuracy.

# File: hardening/friendly_adversarial_training.py
import math
import os
import torch.nn.functional as F

def _required_positive_float(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number")
    return value

FRIENDLY_KL_WEIGHT = _required_positive_float("VERIFIED_H001_FRIENDLY_KL_WEIGHT")


def friendly_adversarial_step(*, model, optimizer, clean_input, adversarial_input, true_label):
    clean_logits = model(clean_input)
    adv_logits = model(adversarial_input)
    
    clean_probs = F.softmax(clean_logits, dim=1)
    adv_log_probs = F.log_softmax(adv_logits, dim=1)
    
    loss_ce = F.cross_entropy(adv_logits, true_label)
    loss_kl = F.kl_div(adv_log_probs, clean_probs.detach(), reduction="batchmean")
    
    total_loss = loss_ce + FRIENDLY_KL_WEIGHT * loss_kl
    optimizer.zero_grad()
    total_loss.backward()
    optimizer.step()
    return total_loss.detach()

Variant C - Free Adversarial Training

Replay the same minibatch several times while updating both the model and an in-batch perturbation to approximate multi-step adversarial training at lower cost.

# File: hardening/free_adversarial_training.py
import math
import os
import torch

def _required_positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be a positive integer")
    return value

FREE_REPLAY_COUNT = _required_positive_int("VERIFIED_H001_FREE_REPLAY_COUNT")

def _required_positive_float(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number")
    return value

FREE_STEP_SIZE = _required_positive_float("VERIFIED_H001_FREE_STEP_SIZE")
FREE_EPSILON = _required_positive_float("VERIFIED_H001_FREE_EPSILON")

if FREE_STEP_SIZE > FREE_EPSILON:
    raise RuntimeError("verified Free adversarial step exceeds epsilon")


def train_free_adversarial(*, model, device, optimizer, criterion, dataloader, clip_min: float, clip_max: float):
    if not math.isfinite(clip_min) or not math.isfinite(clip_max) or clip_min >= clip_max:
        raise ValueError("signed input clip interval is invalid")
    for data, target in dataloader:
        data = data.to(device)
        target = target.to(device)
        delta = torch.zeros_like(data, requires_grad=True)
    
        for _ in range(FREE_REPLAY_COUNT):
            optimizer.zero_grad()
            adv = torch.clamp(data + delta, min=clip_min, max=clip_max)
            logits = model(adv)
            loss = criterion(logits, target)
            loss.backward()
    
            with torch.no_grad():
                delta += FREE_STEP_SIZE * delta.grad.sign()
                delta.clamp_(-FREE_EPSILON, FREE_EPSILON)
                delta.grad = None
    
            optimizer.step()
    return model

Validation and evidence

• Report clean accuracy and adversarial accuracy from the same validation suite.
• Version the chosen attack parameters (eps, eps_step, max_iter, m, lambda_kl) with the training run.
• Do not count this control as complete unless the model was re-evaluated against the same attack family used during training.

Action: Pick one primary general-purpose adversarial example training variant per model family, document why it was chosen, and tie the training configuration to the resulting robustness report.

Before you begin

Apply AID-H-001-G001 only when the model can be retrained and its signed threat model identifies an evasion, perturbation, patch, occlusion, or model-level attention-manipulation threat addressed by the selected variant.

Keep the action safely bounded

block promotion unless the bound candidate actually used this variant and independent replay meets robustness and task-quality policy.

AID-H-001-G002

Production implementation: bounded alternating robust RL

RARL is applicable only when the environment exposes a physically meaningful disturbance channel. The signed profile defines action bounds, disturbance cost, alternating updates, seeds, and the robust-return gate; the adversary may not alter observations, rewards, termination flags, or simulator state outside that channel.

# File: robust_rl/alternating_rarl.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable
import numpy as np


@dataclass(frozen=True)
class Policy:
    act_fn: Callable[[np.ndarray, bool], np.ndarray]
    update_fn: Callable[[list[dict]], dict]

    def act(self, observation: np.ndarray, deterministic: bool) -> np.ndarray:
        value = np.asarray(self.act_fn(observation, deterministic), dtype=np.float32)
        if value.size == 0 or not np.isfinite(value).all():
            raise RuntimeError("policy emitted an empty or non-finite action")
        return value

    def update(self, transitions: list[dict]) -> dict:
        receipt = self.update_fn(transitions)
        if not isinstance(receipt, dict) or not receipt:
            raise RuntimeError("policy update did not return an evidence receipt")
        return receipt


@dataclass(frozen=True)
class DisturbanceEnv:
    reset_fn: Callable[[int], tuple[np.ndarray, dict]]
    step_fn: Callable[[np.ndarray, np.ndarray], tuple[np.ndarray, float, bool, bool, dict]]

    def reset(self, *, seed: int) -> tuple[np.ndarray, dict]:
        observation, receipt = self.reset_fn(seed)
        value = np.asarray(observation, dtype=np.float32)
        if value.size == 0 or not np.isfinite(value).all() or not isinstance(receipt, dict):
            raise RuntimeError("environment reset contract failed")
        return value, receipt

    def step(self, protagonist_action: np.ndarray, disturbance: np.ndarray) -> tuple[np.ndarray, float, bool, bool, dict]:
        result = self.step_fn(protagonist_action, disturbance)
        if not isinstance(result, tuple) or len(result) != 5:
            raise RuntimeError("environment step contract failed")
        observation, reward, terminated, truncated, receipt = result
        value = np.asarray(observation, dtype=np.float32)
        if value.size == 0 or not np.isfinite(value).all() or type(terminated) is not bool or type(truncated) is not bool or not isinstance(receipt, dict):
            raise RuntimeError("environment step returned invalid evidence")
        return value, float(reward), terminated, truncated, receipt

@dataclass(frozen=True)
class RarlProfile:
    seed: int
    total_rounds: int
    episodes_per_round: int
    maximum_steps: int
    disturbance_l2_maximum: float
    disturbance_cost: float

def bounded_disturbance(raw: np.ndarray, maximum: float) -> np.ndarray:
    value = np.asarray(raw, dtype=np.float32)
    norm = float(np.linalg.norm(value))
    if not np.isfinite(norm):
        raise ValueError("adversary emitted non-finite disturbance")
    return value if norm <= maximum else value * (maximum / norm)

def collect_episode(env: DisturbanceEnv, protagonist: Policy, adversary: Policy, profile: RarlProfile, seed: int) -> tuple[list[dict], float]:
    observation, _ = env.reset(seed=seed)
    transitions, episode_return = [], 0.0
    for step in range(profile.maximum_steps):
        action = protagonist.act(observation, deterministic=False)
        disturbance = bounded_disturbance(adversary.act(observation, deterministic=False), profile.disturbance_l2_maximum)
        next_observation, reward, terminated, truncated, info = env.step(action, disturbance)
        if not np.isfinite(reward):
            raise RuntimeError("environment emitted non-finite reward")
        adversary_reward = -float(reward) - profile.disturbance_cost * float(np.square(disturbance).sum())
        transitions.append({"observation": observation, "protagonist_action": action, "disturbance": disturbance, "protagonist_reward": float(reward), "adversary_reward": adversary_reward, "next_observation": next_observation, "done": bool(terminated or truncated), "step": step, "environment_evidence": info})
        episode_return += float(reward)
        observation = next_observation
        if terminated or truncated:
            break
    return transitions, episode_return

def train(env: DisturbanceEnv, protagonist: Policy, adversary: Policy, profile: RarlProfile) -> list[dict]:
    if profile.total_rounds < 1 or profile.episodes_per_round < 1 or profile.maximum_steps < 1 or profile.disturbance_l2_maximum <= 0 or profile.disturbance_cost < 0:
        raise ValueError("invalid signature-verified RARL profile")
    evidence = []
    for round_index in range(profile.total_rounds):
        batch, returns = [], []
        for episode in range(profile.episodes_per_round):
            transitions, robust_return = collect_episode(env, protagonist, adversary, profile, profile.seed + round_index * profile.episodes_per_round + episode)
            batch.extend(transitions)
            returns.append(robust_return)
        if round_index % 2 == 0:
            update, updated = protagonist.update(batch), "protagonist"
        else:
            update, updated = adversary.update(batch), "adversary"
        evidence.append({"round": round_index, "updated_policy": updated, "episodes": len(returns), "mean_robust_return": float(np.mean(returns)), "update": update})
    return evidence

Independent gate: Freeze both policies and replay the complete signed seed/disturbance matrix against baseline and candidate. Retain per-episode returns, clean return, constraint violations, policy digests, simulator image digest, and profile digest. Promotion requires the policy-defined robust-return improvement without exceeding clean-return or safety budgets.

Keep the action safely bounded

block promotion unless the bound candidate actually used this variant and independent replay meets robustness and task-quality policy.

AID-H-001-G003

Concept:

Expose the model to localized, high-saliency adversarial patches during training so it learns to discount them.

Step 1: Generate and apply patches with ART

import math
import os
import json
import subprocess
import tempfile
from pathlib import Path

import torch
from art.attacks.evasion import AdversarialPatch

def _required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number of seconds")
    return value

PATCH_POLICY_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H001_PATCH_POLICY_VERIFY_TIMEOUT_SECONDS")


POLICY_PATH = Path("hardening/patch_training_policy.json")
policy_bytes = POLICY_PATH.read_bytes()
snapshot_path: Path | None = None
try:
    with tempfile.NamedTemporaryFile(prefix="patch-training-policy-", suffix=".json", delete=False) as handle:
        handle.write(policy_bytes)
        handle.flush()
        os.fsync(handle.fileno())
        snapshot_path = Path(handle.name)
    subprocess.run([
        "cosign", "verify-blob", "--key", "keys/patch-training-policy.pub",
        "--bundle", "hardening/patch_training_policy.sig", str(snapshot_path),
    ], check=True, capture_output=True, text=True, timeout=PATCH_POLICY_VERIFY_TIMEOUT_SECONDS)
finally:
    if snapshot_path is not None:
        snapshot_path.unlink(missing_ok=True)

def unique_pairs(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError("duplicate patch-policy key: " + key)
        value[key] = item
    return value

def reject_constant(value):
    raise ValueError("non-finite patch-policy number: " + value)

signed_training_policy = json.loads(
    policy_bytes.decode("utf-8", errors="strict"),
    object_pairs_hook=unique_pairs, parse_constant=reject_constant,
)
if set(signed_training_policy) != {
    "schema_version", "maximum_attack_iterations", "rotation_max_degrees",
    "scale_minimum", "scale_maximum", "attack_batch_size",
    "patched_loss_weight", "patch_scale",
} or signed_training_policy["schema_version"] != "aidefend.patch-training-policy.v1":
    raise ValueError("unsupported patch training policy schema")

attack_iterations = signed_training_policy.get("maximum_attack_iterations")
if isinstance(attack_iterations, bool) or not isinstance(attack_iterations, int) or attack_iterations < 1:
    raise ValueError("signed maximum_attack_iterations must be a positive integer")
rotation_max_degrees = float(signed_training_policy["rotation_max_degrees"])
scale_minimum = float(signed_training_policy["scale_minimum"])
scale_maximum = float(signed_training_policy["scale_maximum"])
attack_batch_size = signed_training_policy["attack_batch_size"]
patched_loss_weight = float(signed_training_policy["patched_loss_weight"])
if (
    not all(math.isfinite(value) for value in (
        rotation_max_degrees, scale_minimum, scale_maximum, patched_loss_weight
    ))
    or rotation_max_degrees < 0
    or not 0 < scale_minimum <= scale_maximum <= 1
    or isinstance(attack_batch_size, bool)
    or not isinstance(attack_batch_size, int)
    or attack_batch_size < 1
    or not 0 < patched_loss_weight < 1
):
    raise ValueError("signed adversarial-patch profile is invalid")

patch_attack = AdversarialPatch(
    classifier=art_classifier,
    rotation_max=rotation_max_degrees,
    scale_min=scale_minimum,
    scale_max=scale_maximum,
    max_iter=attack_iterations,
    batch_size=attack_batch_size
)

for images, labels in train_loader:
    images = images.to(device)
    labels = labels.to(device)

    imgs_np = images.detach().cpu().numpy()
    patch_np, mask_np = patch_attack.generate(x=imgs_np)
    patch_scale = float(signed_training_policy["patch_scale"])
    if not patch_attack.scale_min <= patch_scale <= patch_attack.scale_max:
        raise ValueError("patch_scale is outside the signed attack profile")
    patched_np = patch_attack.apply_patch(
        imgs_np, scale=patch_scale, patch_external=patch_np, mask=mask_np
    )
    patched = torch.from_numpy(patched_np).to(device)

    optimizer.zero_grad()
    clean_outputs = model(images)
    patched_outputs = model(patched)
    loss = (
        (1.0 - patched_loss_weight) * criterion(clean_outputs, labels)
        + patched_loss_weight * criterion(patched_outputs, labels)
    )
    loss.backward()
    optimizer.step()

Step 2: Keep clean and patched objectives in the same loop

patched_loss_weight comes from the verified training policy and must remain strictly between zero and one, so every admitted batch contributes both clean-task and patch-robustness loss. Promotion still requires a separate clean and patch-augmented validation population.

Action: Use the clean/patched loss weight from the verified training policy and evaluate robustness on separate clean and patch-augmented validation populations.

Before you begin

Apply AID-H-001-G003 only when the model can be retrained and its signed threat model identifies an evasion, perturbation, patch, occlusion, or model-level attention-manipulation threat addressed by the selected variant.

Keep the action safely bounded

block promotion unless the bound candidate actually used this variant and independent replay meets robustness and task-quality policy.

AID-H-001-G004

Before you begin

Apply AID-H-001-G004 only when the model can be retrained and its signed threat model identifies an evasion, perturbation, patch, occlusion, or model-level attention-manipulation threat addressed by the selected variant.

Concept:

Random Erasing / Cutout reduces over-reliance on small regions and improves robustness to localized attacks.

Step 1: Add RandomErasing to your transforms

from torchvision import transforms


def build_occlusion_training_transforms(
    *,
    image_size: tuple[int, int],
    normalize_mean: tuple[float, ...],
    normalize_std: tuple[float, ...],
    horizontal_flip_probability: float,
    erase_probability: float,
    erase_scale: tuple[float, float],
    erase_ratio: tuple[float, float],
    erase_value: float,
):
    # Every argument comes from the signature-verified preprocessing/augmentation profile.
    return transforms.Compose([
        transforms.Resize(image_size),
        transforms.RandomHorizontalFlip(p=horizontal_flip_probability),
        transforms.ToTensor(),
        transforms.Normalize(mean=normalize_mean, std=normalize_std),
        transforms.RandomErasing(
            p=erase_probability,
            scale=erase_scale,
            ratio=erase_ratio,
            value=erase_value,
        ),
    ])

Step 2: Feed the transform through the real training loader

from pathlib import Path

from torch.utils.data import DataLoader
from torchvision.datasets import ImageFolder


def build_occlusion_training_loader(
    *,
    dataset_path: Path,
    training_transforms,
    batch_size: int,
    worker_count: int,
) -> DataLoader:
    if batch_size < 1 or worker_count < 0:
        raise ValueError("signed loader profile is invalid")
    train_dataset = ImageFolder(str(dataset_path), transform=training_transforms)
    return DataLoader(
        train_dataset,
        batch_size=batch_size,
        shuffle=True,
        num_workers=worker_count,
        pin_memory=True,
    )

Verification: Compare the exact baseline and candidate model digests on a versioned clean set and a versioned occlusion/patch set; retain per-case metrics, seeds, transform policy, and signed release result.

Action: Track accuracy under occlusions (synthetic patches/masks) to quantify gains.

Keep the action safely bounded

block promotion unless the bound candidate actually used this variant and independent replay meets robustness and task-quality policy.

AID-H-001-G005

Transformer attention-regularization variant

This is one architecture-specific method under AID-H-001, not a separate control objective. AID-H-007.004 owns the integrity and independence of generic evaluation artifacts; this guidance owns the Transformer-specific robustness metric and pass criteria.

Runnable training regularizer

# File: training/attention_entropy_regularization.py
from __future__ import annotations

import math
import torch


def regularized_loss(
    main_loss: torch.Tensor,
    attention_weights: torch.Tensor,
    coefficient: float,
) -> tuple[torch.Tensor, dict]:
    if main_loss.ndim != 0 or not torch.isfinite(main_loss):
        raise ValueError("main_loss must be one finite scalar")
    if (
        not attention_weights.is_floating_point()
        or attention_weights.ndim < 3
        or not torch.isfinite(attention_weights).all()
        or (attention_weights < 0).any()
    ):
        raise ValueError("attention weights must be finite, non-negative floating values")
    if isinstance(coefficient, bool) or not math.isfinite(float(coefficient)) or coefficient <= 0:
        raise ValueError("coefficient must be a finite positive signed-profile value")
    probabilities = attention_weights.clamp_min(
        torch.finfo(attention_weights.dtype).tiny
    )
    row_sums = attention_weights.sum(dim=-1)
    if not torch.allclose(
        row_sums, torch.ones_like(row_sums), atol=1e-5, rtol=1e-5
    ):
        raise ValueError("attention rows must be normalized")
    entropy = -(probabilities * probabilities.log()).sum(dim=-1).mean()
    total = main_loss - float(coefficient) * entropy
    if not torch.isfinite(total):
        raise RuntimeError("regularized loss is not finite")
    return total, {
        "main_loss": float(main_loss.detach().cpu()),
        "attention_entropy": float(entropy.detach().cpu()),
        "coefficient": float(coefficient),
    }

The training workflow verifies the detached signature and exact schema of the versioned regularization profile before constructing coefficient, records the profile digest with every run, and treats unsupported attention tensors as not applicable rather than silently coercing them.

Independent release gate

# File: evaluation/gate_attention_robustness.py
from __future__ import annotations

import hashlib
import hmac
import json
import math
import os
import subprocess
import tempfile
from collections import Counter
from pathlib import Path

def _required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number of seconds")
    return value

ATTENTION_ARTIFACT_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H001_ATTENTION_ARTIFACT_VERIFY_TIMEOUT_SECONDS")
ATTENTION_VERDICT_SIGN_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H001_ATTENTION_VERDICT_SIGN_TIMEOUT_SECONDS")
ATTENTION_VERDICT_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H001_ATTENTION_VERDICT_VERIFY_TIMEOUT_SECONDS")



POLICY_IDENTITY = Path("/opt/aidefend/trust/attention-policy.identity").read_text(encoding="utf-8").strip()
EVALUATOR_IDENTITY = Path("/opt/aidefend/trust/attention-evaluator.identity").read_text(encoding="utf-8").strip()
OIDC_ISSUER = Path("/opt/aidefend/trust/github-actions.issuer").read_text(encoding="utf-8").strip()
VERDICT_VERIFY_KEY = Path("/opt/aidefend/trust/attention-release-gate.pub")
POLICY_FIELDS = {
    "schema_version", "policy_version", "required_threat_contexts",
    "minimum_cases_per_context", "minimum_robustness_improvement",
    "task_quality_minimums", "maximum_task_quality_drop",
    "metric_definition_sha256", "baseline_profile_sha256",
    "candidate_profile_sha256", "evaluation_corpus_sha256",
    "attack_implementation_sha256", "seed_policy_sha256",
}
METRIC_FIELDS = {
    "schema_version", "definition_id", "robustness_formula",
    "attack_success_field", "task_metrics",
}
REPORT_FIELDS = {
    "schema_version", "model_sha256", "regularization_profile_sha256",
    "evaluation_corpus_sha256", "attack_implementation_sha256",
    "seed_policy_sha256", "metric_definition_sha256", "observations",
}
OBSERVATION_FIELDS = {
    "case_id", "threat_context", "attack_success", "task_quality",
}


class InsufficientData(RuntimeError):
    pass


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def verified_blob_bytes(
    path: Path, bundle: Path, identity: str,
) -> tuple[bytes, bytes]:
    artifact_bytes = path.read_bytes()
    bundle_bytes = bundle.read_bytes()
    if not artifact_bytes or not bundle_bytes:
        raise ValueError("signed artifact or bundle is empty")
    with tempfile.TemporaryDirectory(prefix="aidefend-attention-verify-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        artifact_snapshot = root / "artifact"
        bundle_snapshot = root / "bundle.json"
        artifact_snapshot.write_bytes(artifact_bytes)
        bundle_snapshot.write_bytes(bundle_bytes)
        os.chmod(artifact_snapshot, 0o400)
        os.chmod(bundle_snapshot, 0o400)
        subprocess.run([
            "cosign", "verify-blob", "--bundle", str(bundle_snapshot),
            "--certificate-identity", identity,
            "--certificate-oidc-issuer", OIDC_ISSUER, str(artifact_snapshot),
        ], check=True, capture_output=True, text=True, timeout=ATTENTION_ARTIFACT_VERIFY_TIMEOUT_SECONDS)
        if (
            not hmac.compare_digest(artifact_snapshot.read_bytes(), artifact_bytes)
            or not hmac.compare_digest(bundle_snapshot.read_bytes(), bundle_bytes)
        ):
            raise ValueError("verification snapshot changed")
    return artifact_bytes, bundle_bytes


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

    try:
        value = json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=reject_duplicates,
            parse_constant=lambda value: (_ for _ in ()).throw(
                ValueError(f"non-finite JSON number: {value}")
            ),
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
    if not isinstance(value, dict):
        raise ValueError(f"{label} must be an object")
    return value


def finite(value: object, field: str) -> float:
    if isinstance(value, bool):
        raise ValueError(f"{field} must be numeric")
    number = float(value)
    if not math.isfinite(number):
        raise ValueError(f"{field} must be finite")
    return number


def load_policy(path: Path, bundle: Path) -> tuple[dict, str, str]:
    raw, verified_bundle_bytes = verified_blob_bytes(
        path, bundle, POLICY_IDENTITY
    )
    policy = strict_json(raw, "attention policy")
    if set(policy) != POLICY_FIELDS:
        raise ValueError("attention policy schema differs")
    if policy["schema_version"] != "aidefend.attention-robustness-policy.v2":
        raise ValueError("attention policy version is unsupported")
    if not isinstance(policy["policy_version"], str) or not policy["policy_version"]:
        raise ValueError("policy_version is required")
    contexts = policy["required_threat_contexts"]
    if not isinstance(contexts, list) or not contexts or len(contexts) != len(set(contexts)):
        raise ValueError("required_threat_contexts must be unique and non-empty")
    minimum_cases = policy["minimum_cases_per_context"]
    if isinstance(minimum_cases, bool) or not isinstance(minimum_cases, int) or minimum_cases <= 0:
        raise ValueError("minimum_cases_per_context must be a positive integer")
    improvement = finite(
        policy["minimum_robustness_improvement"],
        "minimum_robustness_improvement",
    )
    if not 0.0 <= improvement <= 1.0:
        raise ValueError("minimum_robustness_improvement must be in [0, 1]")
    for field in (
        "metric_definition_sha256", "baseline_profile_sha256",
        "candidate_profile_sha256", "evaluation_corpus_sha256",
        "attack_implementation_sha256", "seed_policy_sha256",
    ):
        value = policy[field]
        if not isinstance(value, str) or len(value) != 64:
            raise ValueError(f"{field} is not a SHA-256 value")
        int(value, 16)
    return (
        policy,
        hashlib.sha256(raw).hexdigest(),
        hashlib.sha256(verified_bundle_bytes).hexdigest(),
    )


def load_metric_definition(
    path: Path,
    bundle: Path,
    policy: dict,
) -> tuple[dict, str, str]:
    raw, verified_bundle_bytes = verified_blob_bytes(
        path, bundle, POLICY_IDENTITY
    )
    raw_sha256 = hashlib.sha256(raw).hexdigest()
    if raw_sha256 != policy["metric_definition_sha256"]:
        raise ValueError("metric definition digest differs from signed policy")
    metric = strict_json(raw, "attention metric definition")
    if set(metric) != METRIC_FIELDS:
        raise ValueError("metric definition schema differs")
    if (
        metric["schema_version"] != "aidefend.attention-metric-definition.v1"
        or metric["robustness_formula"] != "one_minus_attack_success_rate"
        or metric["attack_success_field"] != "attack_success"
    ):
        raise ValueError("metric definition is unsupported")
    task_metrics = metric["task_metrics"]
    if not isinstance(task_metrics, dict) or not task_metrics:
        raise ValueError("task metric definitions are required")
    for name, spec in task_metrics.items():
        if (
            not isinstance(name, str)
            or not name
            or not isinstance(spec, dict)
            or set(spec) != {"aggregation", "value_min", "value_max"}
        ):
            raise ValueError("task metric definition differs")
        if spec["aggregation"] != "mean":
            raise ValueError("only declared mean aggregation is implemented")
        low = finite(spec["value_min"], f"{name}.value_min")
        high = finite(spec["value_max"], f"{name}.value_max")
        if not low < high:
            raise ValueError(f"{name} range is invalid")
    if set(policy["task_quality_minimums"]) != set(task_metrics):
        raise ValueError("task-quality minimum keys differ from metric definition")
    if set(policy["maximum_task_quality_drop"]) != set(task_metrics):
        raise ValueError("task-quality drop keys differ from metric definition")
    for name, spec in task_metrics.items():
        low = finite(spec["value_min"], f"{name}.value_min")
        high = finite(spec["value_max"], f"{name}.value_max")
        minimum = finite(policy["task_quality_minimums"][name], f"{name}.minimum")
        maximum_drop = finite(
            policy["maximum_task_quality_drop"][name], f"{name}.maximum_drop"
        )
        if not low <= minimum <= high or not 0.0 <= maximum_drop <= high - low:
            raise ValueError(f"{name} policy limit is outside its signed range")
    return (
        metric,
        raw_sha256,
        hashlib.sha256(verified_bundle_bytes).hexdigest(),
    )

def load_report(
    path: Path,
    bundle: Path,
    expected_model_sha256: str,
    expected_profile_sha256: str,
    policy: dict,
    metric: dict,
) -> tuple[dict, str, str]:
    raw, verified_bundle_bytes = verified_blob_bytes(
        path, bundle, EVALUATOR_IDENTITY
    )
    report = strict_json(raw, "attention raw observations")
    if set(report) != REPORT_FIELDS:
        raise InsufficientData("raw evaluation report schema differs")
    if report["schema_version"] != "aidefend.attention-raw-observations.v1":
        raise InsufficientData("raw evaluation report version is unsupported")
    if report["model_sha256"] != expected_model_sha256:
        raise InsufficientData("raw report is not bound to actual model bytes")
    if report["regularization_profile_sha256"] != expected_profile_sha256:
        raise InsufficientData("regularization profile binding differs")
    if report["metric_definition_sha256"] != policy["metric_definition_sha256"]:
        raise InsufficientData("metric definition binding differs")
    for field in (
        "evaluation_corpus_sha256",
        "attack_implementation_sha256",
        "seed_policy_sha256",
    ):
        if report[field] != policy[field]:
            raise InsufficientData(
                f"raw report {field} differs from the signed policy"
            )
    observations = report["observations"]
    if not isinstance(observations, list) or not observations:
        raise InsufficientData("raw observations are missing")
    case_ids = []
    for row in observations:
        if not isinstance(row, dict) or set(row) != OBSERVATION_FIELDS:
            raise InsufficientData("raw observation schema differs")
        if not isinstance(row["case_id"], str) or not row["case_id"]:
            raise InsufficientData("case_id is missing")
        if row["threat_context"] not in policy["required_threat_contexts"]:
            raise InsufficientData("undeclared threat context is present")
        if not isinstance(row["attack_success"], bool):
            raise InsufficientData("attack_success must be a raw boolean")
        if (
            not isinstance(row["task_quality"], dict)
            or set(row["task_quality"]) != set(metric["task_metrics"])
        ):
            raise InsufficientData("task-quality observation schema differs")
        for name, value in row["task_quality"].items():
            number = finite(value, f"{name}.observation")
            spec = metric["task_metrics"][name]
            if not finite(spec["value_min"], name) <= number <= finite(
                spec["value_max"], name
            ):
                raise InsufficientData(
                    f"{name} observation is outside its metric range"
                )
        case_ids.append(row["case_id"])
    if len(case_ids) != len(set(case_ids)):
        raise InsufficientData("case IDs are duplicated")
    counts = Counter(row["threat_context"] for row in observations)
    expected_contexts = set(policy["required_threat_contexts"])
    if set(counts) != expected_contexts:
        raise InsufficientData("required threat-context coverage differs")
    if any(
        counts[name] < policy["minimum_cases_per_context"]
        for name in expected_contexts
    ):
        raise InsufficientData("threat-context case coverage is insufficient")
    return (
        report,
        hashlib.sha256(raw).hexdigest(),
        hashlib.sha256(verified_bundle_bytes).hexdigest(),
    )

def summarize(report: dict, metric: dict) -> tuple[float, dict[str, float]]:
    rows = report["observations"]
    robustness = 1.0 - sum(row["attack_success"] for row in rows) / len(rows)
    task_quality = {
        name: sum(float(row["task_quality"][name]) for row in rows) / len(rows)
        for name in metric["task_metrics"]
    }
    return robustness, task_quality


def gate(
    policy_path: Path,
    policy_bundle: Path,
    metric_path: Path,
    metric_bundle: Path,
    baseline_report_path: Path,
    baseline_report_bundle: Path,
    candidate_report_path: Path,
    candidate_report_bundle: Path,
    baseline_model_path: Path,
    candidate_model_path: Path,
) -> dict:
    baseline_model_bytes = baseline_model_path.read_bytes()
    candidate_model_bytes = candidate_model_path.read_bytes()
    if not baseline_model_bytes or not candidate_model_bytes:
        raise InsufficientData("baseline or candidate model artifact is empty")
    baseline_model_sha256 = hashlib.sha256(baseline_model_bytes).hexdigest()
    candidate_model_sha256 = hashlib.sha256(candidate_model_bytes).hexdigest()

    (
        policy,
        policy_sha256,
        policy_bundle_sha256,
    ) = load_policy(policy_path, policy_bundle)
    (
        metric,
        metric_sha256,
        metric_bundle_sha256,
    ) = load_metric_definition(metric_path, metric_bundle, policy)
    (
        baseline,
        baseline_report_sha256,
        baseline_report_bundle_sha256,
    ) = load_report(
        baseline_report_path,
        baseline_report_bundle,
        baseline_model_sha256,
        policy["baseline_profile_sha256"],
        policy,
        metric,
    )
    (
        candidate,
        candidate_report_sha256,
        candidate_report_bundle_sha256,
    ) = load_report(
        candidate_report_path,
        candidate_report_bundle,
        candidate_model_sha256,
        policy["candidate_profile_sha256"],
        policy,
        metric,
    )
    bindings = (
        "evaluation_corpus_sha256", "attack_implementation_sha256",
        "seed_policy_sha256", "metric_definition_sha256",
    )
    if any(baseline[field] != candidate[field] for field in bindings):
        raise InsufficientData("baseline and candidate evaluation bindings differ")
    baseline_population = [
        (row["case_id"], row["threat_context"]) for row in baseline["observations"]
    ]
    candidate_population = [
        (row["case_id"], row["threat_context"]) for row in candidate["observations"]
    ]
    if baseline_population != candidate_population:
        raise InsufficientData("baseline and candidate case/context populations differ")

    baseline_robustness, baseline_quality = summarize(baseline, metric)
    candidate_robustness, candidate_quality = summarize(candidate, metric)
    improvement = candidate_robustness - baseline_robustness
    failures = []
    if improvement < float(policy["minimum_robustness_improvement"]):
        failures.append("robustness_improvement")
    for name in metric["task_metrics"]:
        if candidate_quality[name] < float(policy["task_quality_minimums"][name]):
            failures.append(f"task_quality_minimum:{name}")
        if baseline_quality[name] - candidate_quality[name] > float(
            policy["maximum_task_quality_drop"][name]
        ):
            failures.append(f"task_quality_drop:{name}")

    return {
        "schema_version": "aidefend.attention-robustness-evaluation.v2",
        "status": "fail" if failures else "pass",
        "failures": failures,
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
        "policy_bundle_sha256": policy_bundle_sha256,
        "metric_definition_sha256": metric_sha256,
        "metric_definition_bundle_sha256": metric_bundle_sha256,
        "baseline_model_sha256": baseline_model_sha256,
        "candidate_model_sha256": candidate_model_sha256,
        "baseline_report_sha256": baseline_report_sha256,
        "baseline_report_bundle_sha256": baseline_report_bundle_sha256,
        "candidate_report_sha256": candidate_report_sha256,
        "candidate_report_bundle_sha256": candidate_report_bundle_sha256,
        "bindings": {field: candidate[field] for field in bindings},
        "baseline_robustness": baseline_robustness,
        "candidate_robustness": candidate_robustness,
        "robustness_improvement": improvement,
        "baseline_task_quality": baseline_quality,
        "candidate_task_quality": candidate_quality,
    }


def canonical(value: dict) -> bytes:
    return json.dumps(
        value, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode()


def gate_with_status(**kwargs) -> dict:
    try:
        outcome = gate(**kwargs)
        outcome["input_evidence_state"] = "verified"
        return outcome
    except InsufficientData as exc:
        return {
            "schema_version": "aidefend.attention-robustness-evaluation.v2",
            "status": "insufficient_data",
            "failures": [],
            "reason": str(exc),
            "input_evidence_state": "unverified",
            "verified_input_digests": None,
        }
    except Exception as exc:
        return {
            "schema_version": "aidefend.attention-robustness-evaluation.v2",
            "status": "error",
            "failures": [],
            "reason": "attention gate input or dependency verification failed",
            "error_class": type(exc).__name__,
            "input_evidence_state": "unverified",
            "verified_input_digests": None,
        }


def write_exclusive(path: Path, payload: bytes) -> None:
    with path.open("xb") as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())


def gate_and_sign(
    *, verdict_path: Path, verdict_signature_path: Path, **gate_inputs,
) -> dict:
    if verdict_path.exists() or verdict_signature_path.exists():
        raise FileExistsError("verdict publish path already exists")
    outcome = gate_with_status(**gate_inputs)
    payload = canonical(outcome) + b"\n"
    expected_digest = hashlib.sha256(payload).hexdigest()

    with tempfile.TemporaryDirectory(
        prefix="aidefend-attention-verdict-"
    ) as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "verdict.json"
        signature_snapshot = root / "verdict.sig"
        write_exclusive(payload_snapshot, payload)
        os.chmod(payload_snapshot, 0o400)
        if sha256_file(payload_snapshot) != expected_digest:
            raise RuntimeError("private canonical verdict snapshot differs")
        subprocess.run([
            "cosign", "sign-blob", "--yes",
            "--key", "env://ATTENTION_GATE_SIGNING_KEY",
            "--bundle", str(signature_snapshot),
            str(payload_snapshot),
        ], check=True, capture_output=True, text=True, timeout=ATTENTION_VERDICT_SIGN_TIMEOUT_SECONDS)
        signature_bytes = signature_snapshot.read_bytes()
        if not signature_bytes or sha256_file(payload_snapshot) != expected_digest:
            raise RuntimeError("private verdict changed during signing")
        subprocess.run([
            "cosign", "verify-blob", "--key", str(VERDICT_VERIFY_KEY),
            "--bundle", str(signature_snapshot), str(payload_snapshot),
        ], check=True, capture_output=True, text=True, timeout=ATTENTION_VERDICT_VERIFY_TIMEOUT_SECONDS)
        if (
            sha256_file(payload_snapshot) != expected_digest
            or not hmac.compare_digest(
                signature_snapshot.read_bytes(), signature_bytes
            )
        ):
            raise RuntimeError("private verdict or signature changed during verification")
        write_exclusive(verdict_signature_path, signature_bytes)
        write_exclusive(verdict_path, payload)
    if (
        sha256_file(verdict_path) != expected_digest
        or not hmac.compare_digest(
            verdict_signature_path.read_bytes(), signature_bytes
        )
    ):
        raise RuntimeError("published verdict pair differs from verified snapshots")
    return outcome

Evidence: The gate captures policy, metric definition, and raw reports into private read-only snapshots, verifies signatures over those snapshots, and parses only the verified snapshot bytes; hashes the actual baseline and candidate model bytes; rejects non-finite values; and derives robustness from signed per-case observations rather than trusting a report-supplied score. Both models must cover the exact same case/context population and all policy-required contexts. The signed policy, rather than either report, pins the exact evaluation-corpus, attack-implementation, and seed-policy digests; equality between two reports is never sufficient by itself.

Action: Release automation must call gate_and_sign, not the pure gate, so pass, fail, insufficient_data, and error all become canonical verdicts signed under an identity separate from training. Missing signatures, artifacts, cases, contexts, metrics, or digest bindings produces a signed negative verdict and blocks promotion. The presence of entropy regularization, noisy attention, RoPE, or an architecture label is never evidence by itself.

Open-source tools

• Adversarial Robustness Toolbox (ART)
• CleverHans
• Foolbox
• Torchattacks
• TextAttack
• Stable-Baselines3
• Ray RLlib
• PettingZoo
• PyTorch
• TensorFlow
• Albumentations (data augmentation)
• torchvision.transforms (data augmentation)
• MLflow
• RobustBench (AutoAttack evaluation)

Commercial / hosted tools

• Cisco AI Defense AI Model & Application Validation
• HiddenLayer AI Attack Simulation
• Prisma AIRS AI Red Teaming
• Adversa AI Red Teaming Platform
• Amazon SageMaker AI
• Google Vertex AI
• Databricks Runtime for Machine Learning
• Azure Machine Learning
• Weights & Biases

Threat mappings

MITRE ATLAS

• AML.T0015 Evade AI Model
• AML.T0043 Craft Adversarial Data
• AML.T0043.000 Craft Adversarial Data: White-Box Optimization
• AML.T0043.001 Craft Adversarial Data: Black-Box Optimization
• AML.T0043.002 Craft Adversarial Data: Black-Box Transfer
• AML.T0043.003 Craft Adversarial Data: Manual Modification

MAESTRO

• Adversarial Examples (L1)
• Evasion of Security AI Agents (L6)
• Reprogramming Attacks (L1)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• ML01:2023 Input Manipulation Attack

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.022 Evasion
• NISTAML.025 Black-box Evasion

Cisco Integrated AI Security and Safety Framework

• AISubtech-1.3.2 Image Manipulation
• AITech-1.3 Multimodal Injection and Manipulation
• AITech-9.1 Model or Agentic System Manipulation
• AITech-9.2 Detection Evasion

Google Secure AI Framework 2.0 - Risks

• MEV: Model Evasion

Databricks AI Security Framework 3.0

• Model Serving - Inference response 10.5: Black-box attacks