AID-H-007.002: Runtime Training Job Monitoring & Auditing

Version 1.20261015 · Actionable sub-technique

Focuses on instrumenting the training script itself to continuously monitor for behavioral anomalies and to create a detailed, immutable audit log. This involves real-time tracking of key training metrics (e.g., loss, gradient norms) to detect signs of instability or poisoning, and systematically logging all parameters, code versions, and data versions to ensure any training run is fully auditable and reproducible.

Pillars: infra

Lifecycle: building, operation

Scope boundary

Owns runtime training-job instrumentation, anomaly and stability checks, controlled termination, and immutable audit records for metrics, code, data, parameters, environment, and produced artifacts.

AID-H-007.001: AID-H-007.002 observes and controls the running training job; AID-H-007.001 establishes the isolated infrastructure and least-privilege execution environment. Runtime evidence and environment provisioning are separately verifiable controls.

AID-H-007.003: AID-H-007.002 records and responds to behavior during the original run; AID-H-007.003 captures the complete run manifest and independently replays it for reproducibility. A detailed audit trail is necessary but does not establish that another environment can reproduce the result.

Implementation guidance

AID-H-007.002-G001

Monitor training metrics in real time to detect anomalies that suggest poisoning, backdoor insertion, or destabilization.
Concept:

Poisoned data, injected backdoors, or malicious changes to the training loop often show up as unusual behavior during training: sudden loss spikes, abnormal gradient norms, unexpected divergence after previously stable convergence, etc. You want to continuously collect these signals and treat them as security telemetry, not just ML telemetry.

Step 1: Log Key Training Signals Every Epoch

Use an experiment tracking system (e.g. MLflow) to log metrics such as loss, accuracy, and gradient norm at each epoch. This creates a historical baseline for normal training behavior.

# File: hardening/training_monitor.py
import mlflow
import torch
import numpy as np

# Example training loop (simplified)
# with mlflow.start_run():
#     for epoch in range(num_epochs):
#         model.train()
#         optimizer.zero_grad()
#
#         outputs = model(inputs)
#         loss = criterion(outputs, labels)
#         loss.backward()
#
#         # Calculate total gradient L2 norm for anomaly tracking
#         total_norm_sq = 0.0
#         for p in model.parameters():
#             if p.grad is not None:
#                 param_norm = p.grad.data.norm(2).item()
#                 total_norm_sq += param_norm ** 2
#         gradient_norm = total_norm_sq ** 0.5
#
#         # Optimizer step
#         optimizer.step()
#
#         # Compute batch accuracy
#         accuracy = (outputs.argmax(dim=1) == labels).float().mean().item()
#
#         # Log metrics to MLflow for this epoch
#         mlflow.log_metric("train_loss", loss.item(), step=epoch)
#         mlflow.log_metric("accuracy", accuracy, step=epoch)
#         mlflow.log_metric("gradient_norm", gradient_norm, step=epoch)
Step 2: Run a Post-Training Anomaly Check

After the run finishes, compare collected metrics against historical "known good" runs. Alert if you see suspicious instability, such as loss doubling in one step or gradient norms exploding.

# File: hardening/check_metrics.py
import mlflow

def check_run_metrics(run_id: str, loss_spike_factor: float) -> bool:
    """
    Returns False if suspicious behavior is detected, True otherwise.
    Example heuristic: flag if training loss jumps by more than `loss_spike_factor`
    between consecutive logged steps.
    """
    if loss_spike_factor <= 0:
        raise ValueError("signed loss-spike factor must be positive")
    client = mlflow.tracking.MlflowClient()
    loss_history = client.get_metric_history(run_id, "train_loss")

    # Sort by step just in case
    loss_history = sorted(loss_history, key=lambda m: m.step)

    for i in range(1, len(loss_history)):
        prev_val = loss_history[i-1].value
        curr_val = loss_history[i].value
        if prev_val > 0 and curr_val > prev_val * loss_spike_factor:
            print(f"🚨 ALERT: Sudden loss spike at step {loss_history[i].step} ({prev_val} -> {curr_val})")
            return False

    return True

# Example usage after training:
# is_clean = check_run_metrics(run.info.run_id)
# if not is_clean:
#     raise RuntimeError("Training run flagged as anomalous. Investigate for poisoning or instability.")

Action: Treat your training loop like production telemetry. Always log loss, accuracy, and gradient norm to a central place. After every run, automatically compare against historical norms and block promotion of any model produced by a suspicious run.

Verify safely

A separately credentialed verifier independently reads back or replays complete step/epoch/job telemetry for each eligible training run under a signed baseline and anomaly policy, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-007.002-G002

Enforce runtime stability checks (loss sanity, gradient clipping) and kill the job if it becomes unstable.
Concept:

Attackers can intentionally destabilize training (e.g. exploding gradients, NaN loss) to waste GPU hours or to corrupt the model so it can't be safely deployed. You should automatically detect unstable conditions and stop the run before it burns more resources or writes a poisoned checkpoint.

Add Gradient Clipping and Loss Sanity Checks to the Training Loop

The logic below shows how to (1) halt on invalid loss and (2) clip gradients to keep training numerically stable.

# File: hardening/training_stability_guard.py
import math
import os
import numpy as np
import torch

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

MAX_GRAD_NORM = _required_positive_float("VERIFIED_H007_MAX_GRAD_NORM")



def training_step(model, optimizer, criterion, inputs, labels):
    model.train()
    optimizer.zero_grad()

    outputs = model(inputs)
    loss = criterion(outputs, labels)

    # Check for invalid loss (NaN / Inf)
    loss_value = loss.item()
    if not np.isfinite(loss_value):
        raise RuntimeError("🔥 TRAINING UNSTABLE: Loss is NaN or Inf. Aborting run.")

    loss.backward()

    # Clip gradient norm to prevent runaway updates
    torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=MAX_GRAD_NORM)

    optimizer.step()
    return loss_value

# Example usage inside the main loop:
# for epoch in range(num_epochs):
#     batch_loss = training_step(model, optimizer, criterion, batch_inputs, batch_labels)
#     print(f"Epoch {epoch} loss: {batch_loss}")

Action: Make gradient clipping and loss validity checks mandatory in your standard training template. If loss becomes non-finite or gradients explode, immediately fail the run and alert. Treat this as both reliability control and security control (potential poisoning / malicious batch).

Keep the action safely bounded

Evaluate authenticated telemetry at every policy interval, pause or terminate on breach, revoke job credentials, preserve a bounded forensic checkpoint, deny artifact promotion, and require signed exception/recovery evidence before restart.

Verify safely

A separately credentialed verifier independently reads back or replays each active training run and signed stability/resource/security stop policy, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-007.002-G003

Attach a mandatory audit trail to every training run: code version, dataset version, container image, and hyperparameters.
Concept:

If you cannot answer "what exact code/data/container produced this model?" then you cannot (a) prove integrity, (b) reproduce the model after an incident, or (c) investigate suspected poisoning. Every run must be logged with immutable identifiers that tie it back to source control, data lineage, and the runtime environment.

Collect Immutable Metadata at Job Start

The CI/CD or orchestrator that launches training should inject metadata (Git commit SHA, dataset hash from DVC or internal catalog, container image URI) into environment variables. The training script then records them in an experiment tracker.

# File: hardening/auditable_training.py
import json
import mlflow
import os

# These environment variables should be populated by your pipeline
# before the training container starts.
GIT_COMMIT_SHA = os.environ["GIT_COMMIT_SHA"]
DVC_DATA_HASH = os.environ["DVC_DATA_HASH"]
TRAINING_IMAGE_URI = os.environ["TRAINING_IMAGE_URI"]
HYPERPARAMS = json.loads(os.environ["VERIFIED_H007_TRAINING_HYPERPARAMS_JSON"])
if not isinstance(HYPERPARAMS, dict) or not HYPERPARAMS:
    raise RuntimeError("verified training hyperparameter profile is empty or invalid")

with mlflow.start_run() as run:
    # Record provenance as tags (immutable context)
    if GIT_COMMIT_SHA:
        mlflow.set_tag("git_commit", GIT_COMMIT_SHA)
    if DVC_DATA_HASH:
        mlflow.set_tag("data_hash", DVC_DATA_HASH)
    if TRAINING_IMAGE_URI:
        mlflow.set_tag("docker_image", TRAINING_IMAGE_URI)

    # Record hyperparameters
    mlflow.log_params(HYPERPARAMS)

    # The signed production training entry point executes between these audit events.
    # During training, also log metrics like loss/accuracy/gradient_norm

    print(f"Completed auditable run. Run ID: {run.info.run_id}")

Action: Make this metadata logging non-optional. Your promotion pipeline should refuse to register or deploy any model artifact that does not have: commit hash, dataset hash, container image reference, and hyperparameters logged. This turns every training run into a forensically usable record.

Verify safely

A separately credentialed verifier independently reads back or replays the complete event population for each training run from admission through artifact publication or termination, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

Open-source tools

Source-available / open-weight tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-007.002: Runtime Training Job Monitoring & Auditing | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service