AID-H-010: Classifier-Free Guidance Hardening

Version 1.20261015 · Actionable technique

A set of techniques focused on hardening the Classifier-Free Guidance (CFG) mechanism in diffusion models.

What CFG does: CFG is a core component that steers image generation towards a text prompt, but adversaries can exploit high guidance scale values to force the model to generate harmful, unsafe, or out-of-distribution content.

The threat: High CFG scales can overpower built-in safety conditioning and negative prompts, effectively steering the diffusion model toward disallowed or high-risk generations.

Defense goal: These hardening techniques aim to control the CFG scale and its influence, preventing its misuse while preserving the model's creative capabilities.

Pillars: model

Lifecycle: building, operation

Scope boundary

Owns server-side policy enforcement and effective readback for Classifier-Free Guidance and related diffusion guider parameters when untrusted callers can influence them. It does not treat CFG, negative prompts, or risk heuristics as authoritative content filters.

AID-H-002.002: AID-H-010 enforces diffusion-specific CFG and guider control surfaces at sampling time; AID-H-002.002 validates general inference prompts and request inputs. A prompt that passes admission can still request an unsafe guidance configuration.

AID-H-033.006: AID-H-010 owns CFG-specific diffusion semantics and safety constraints; AID-H-033.006 resolves and enforces the complete effective invocation configuration after gateway defaults, aliases, provider translation, and retries. The gateway must preserve the specialized CFG policy but does not redefine it.

Implementation guidance

AID-H-010-G001

Implement strict server-side validation and clipping of the guidance scale parameter.

Apply this control only to a diffusion endpoint whose deployed sampler exposes classifier-free guidance. The serving team owns the enforcement wrapper; the caller may request a scale but cannot select the policy, model binding, or validation result. A signed policy binds the exact model artifact, allowed scale interval, default, inference-step interval, validity window, and policy version. An endpoint without CFG is NOT_APPLICABLE; an unavailable or invalid policy is ERROR and inference remains closed.

Verify the release policy and enforce finite bounded values
# File: serving/cfg_guard.py
# Dependencies: pydantic>=2.8,<3
from __future__ import annotations

import hashlib
import json
import math
import os
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from pydantic import BaseModel, ConfigDict, Field

# aidefend.cfg-request.v1 defines an empty prompt as structurally invalid.
CFG_REQUEST_NON_EMPTY_PROMPT_LENGTH = 1

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

GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H010_GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS")


POLICY = Path(os.environ["CFG_POLICY_PATH"])
POLICY_BUNDLE = Path(os.environ["CFG_POLICY_BUNDLE_PATH"])
VERIFY_KEY = os.environ["CFG_POLICY_VERIFY_KEY"]
REQUIRED = {
    "schema_version", "policy_version", "model_artifact_sha256",
    "minimum_scale", "maximum_scale", "default_scale",
    "minimum_inference_steps", "maximum_inference_steps",
    "issued_at", "expires_at",
}


def no_duplicate_keys(pairs: list[tuple[str, object]]) -> dict:
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate policy key: {key}")
        value[key] = item
    return value


def reject_nonfinite(value: str):
    raise ValueError(f"non-finite policy number: {value}")


def strict_json(raw: bytes) -> dict:
    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=no_duplicate_keys,
        parse_constant=reject_nonfinite,
    )
    if not isinstance(value, dict):
        raise ValueError("policy root must be an object")
    return value


def timestamp(value: object) -> datetime:
    if not isinstance(value, str):
        raise ValueError("policy timestamp is absent")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("policy timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def load_policy(model_artifact_sha256: str) -> tuple[dict, str]:
    raw = POLICY.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h010-policy-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", VERIFY_KEY,
                "--bundle", str(POLICY_BUNDLE), str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS,
        )
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)
    policy = strict_json(raw)
    if (
        not isinstance(policy, dict)
        or set(policy) != REQUIRED
        or policy["schema_version"] != "aidefend.cfg-policy.v1"
        or policy["model_artifact_sha256"] != model_artifact_sha256
    ):
        raise ValueError("CFG policy schema or model binding differs")
    for field in ("minimum_scale", "maximum_scale", "default_scale"):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            raise ValueError(f"{field} is not numeric")
        policy[field] = float(value)
        if not math.isfinite(policy[field]):
            raise ValueError(f"{field} is not finite")
    if not (
        0.0 < policy["minimum_scale"]
        <= policy["default_scale"]
        <= policy["maximum_scale"]
    ):
        raise ValueError("CFG scale interval is invalid")
    for field in ("minimum_inference_steps", "maximum_inference_steps"):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise ValueError(f"{field} is invalid")
    if policy["minimum_inference_steps"] > policy["maximum_inference_steps"]:
        raise ValueError("inference-step interval is invalid")
    now = datetime.now(timezone.utc)
    issued, expires = timestamp(policy["issued_at"]), timestamp(policy["expires_at"])
    if issued > now or expires <= issued or now >= expires:
        raise ValueError("CFG policy is not currently valid")
    if not isinstance(policy["policy_version"], str) or not policy["policy_version"]:
        raise ValueError("policy version is absent")
    return policy, hashlib.sha256(raw).hexdigest()


class GenerateRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", allow_inf_nan=False, strict=True)
    prompt: str = Field(min_length=CFG_REQUEST_NON_EMPTY_PROMPT_LENGTH)
    guidance_scale: float | None = None
    num_inference_steps: int


def guarded_generate(
    pipeline: Any,
    request: GenerateRequest,
    request_id: str,
    model_artifact_sha256: str,
) -> tuple[Any, dict]:
    if not request_id:
        raise ValueError("authenticated request ID is absent")
    policy, policy_sha256 = load_policy(model_artifact_sha256)
    scale = (
        policy["default_scale"]
        if request.guidance_scale is None
        else request.guidance_scale
    )
    if not policy["minimum_scale"] <= scale <= policy["maximum_scale"]:
        raise ValueError("guidance scale is outside signed policy")
    if not (
        policy["minimum_inference_steps"]
        <= request.num_inference_steps
        <= policy["maximum_inference_steps"]
    ):
        raise ValueError("inference steps are outside signed policy")
    image = pipeline(
        prompt=request.prompt,
        guidance_scale=scale,
        num_inference_steps=request.num_inference_steps,
    ).images[0]
    decision = {
        "schema_version": "aidefend.cfg-decision.v1",
        "request_id": request_id,
        "model_artifact_sha256": model_artifact_sha256,
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
        "requested_scale": request.guidance_scale,
        "applied_scale": scale,
        "inference_steps": request.num_inference_steps,
        "decision": "ALLOWED",
        "decided_at": datetime.now(timezone.utc).isoformat(),
    }
    return image, decision
Verification, evidence, and outcomes

Route every production endpoint through this wrapper and write allow and reject decisions to an append-only store using an identity that the model-serving workload cannot alter. Retain the signed policy and bundle, exact model digest, endpoint inventory, request correlation ID, requested/applied values, decision, and test-run identity. An independent verifier enumerates every in-scope route, re-verifies the policy, and replays below-minimum, above-maximum, boundary, missing, NaN, infinity, Boolean, string, unknown-field, and invalid-step cases against the deployed endpoint.

Action: Deploy the guard at the last trusted hop before the sampler, export the exact signed decision population, and block release unless independent replay returns PASS.

AID-H-010-G002

Apply adaptive or per-prompt guidance scaling based on prompt risk analysis.
Applicability, dependency, and scope

Use adaptive CFG only when the deployment has an independently operated prompt-risk service and a diffusion endpoint with CFG. This guidance depends on the finite base bounds from AID-H-010-G001; its signed risk policy may only narrow that range. It is defense-in-depth and does not replace input sanitization, output moderation, or model-safety evaluation. If CFG is absent, return NOT_APPLICABLE. If the risk service or its evidence is unavailable, return ERROR and do not sample.

Before startup, a bootstrap identity must signature-verify a separate versioned transport profile and export the exact risk-service URL, trust bundle, client certificate/key paths, positive HTTP deadlines, and decoded response-byte cap used below. Retain that profile version and digest with each decision; do not take these bounds from an unverified environment or from the policy being verified.

Consume a signed, prompt-bound risk receipt
# File: serving/adaptive_cfg.py
# Dependencies: httpx>=0.27,<1, PyJWT>=2.9,<3, cryptography>=43,<46
from __future__ import annotations

import hashlib
import json
import math
import os
import ssl
import subprocess
import tempfile
import time
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit

import httpx
import jwt

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

ADAPTIVE_GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H010_ADAPTIVE_GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS")
PROMPT_RISK_TOTAL_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H010_PROMPT_RISK_TOTAL_TIMEOUT_SECONDS")
PROMPT_RISK_CONNECT_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H010_PROMPT_RISK_CONNECT_TIMEOUT_SECONDS")


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


PROMPT_RISK_RESPONSE_MAX_BYTES = _required_positive_int(
    "VERIFIED_H010_PROMPT_RISK_RESPONSE_MAX_BYTES"
)


def canonical_https_url(value: str) -> str:
    if not isinstance(value, str) or not value:
        raise RuntimeError("risk service URL is empty")
    parsed = urlsplit(value)
    if (
        parsed.scheme != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.query
        or parsed.fragment
        or not parsed.path.startswith("/")
    ):
        raise RuntimeError("risk service URL must be an exact HTTPS endpoint")
    canonical = urlunsplit(("https", parsed.netloc.lower(), parsed.path, "", ""))
    if value != canonical:
        raise RuntimeError("risk service URL is not canonical")
    return value


POLICY = Path(os.environ["CFG_RISK_POLICY_PATH"])
POLICY_BUNDLE = Path(os.environ["CFG_RISK_POLICY_BUNDLE_PATH"])
POLICY_KEY = os.environ["CFG_RISK_POLICY_VERIFY_KEY"]
RISK_PUBLIC_KEY = Path(os.environ["CFG_RISK_RECEIPT_PUBLIC_KEY_PATH"])
RISK_URL = canonical_https_url(os.environ["CFG_RISK_URL"])
RISK_CA = os.environ["CFG_RISK_CA_BUNDLE"]
RISK_CERT = os.environ["CFG_RISK_CLIENT_CERT"]
RISK_KEY = os.environ["CFG_RISK_CLIENT_KEY"]
TIERS = {"allow", "clamp", "block"}


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def no_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


def strict_json(raw: bytes) -> dict:
    value = json.loads(raw, object_pairs_hook=no_duplicates, parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")))
    if not isinstance(value, dict):
        raise ValueError("JSON root must be an object")
    return value


def bounded_json_response(response: httpx.Response) -> dict:
    if response.status_code != 200:
        raise RuntimeError(f"risk service returned HTTP {response.status_code}")
    media_type = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
    if media_type != "application/json":
        raise RuntimeError("risk service did not return application/json")
    declared = response.headers.get("Content-Length")
    if declared is not None:
        try:
            if int(declared) > PROMPT_RISK_RESPONSE_MAX_BYTES:
                raise RuntimeError("risk response exceeds signed byte cap")
        except ValueError as exc:
            raise RuntimeError("risk response Content-Length is invalid") from exc
    body = bytearray()
    for chunk in response.iter_bytes():
        body.extend(chunk)
        if len(body) > PROMPT_RISK_RESPONSE_MAX_BYTES:
            raise RuntimeError("risk response exceeds signed decoded-byte cap")
    return strict_json(bytes(body))


def load_policy() -> tuple[dict, str]:
    raw = POLICY.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h010-policy-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", POLICY_KEY,
                "--bundle", str(POLICY_BUNDLE), str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=ADAPTIVE_GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS,
        )
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)
    policy = strict_json(raw)
    required = {
        "schema_version", "policy_version", "receipt_issuer",
        "receipt_audience", "receipt_public_key_sha256",
        "maximum_receipt_age_seconds", "tier_actions",
    }
    if (
        not isinstance(policy, dict)
        or set(policy) != required
        or policy["schema_version"] != "aidefend.cfg-risk-policy.v1"
        or set(policy["tier_actions"]) != TIERS
        or policy["receipt_public_key_sha256"]
        != sha256_bytes(RISK_PUBLIC_KEY.read_bytes())
    ):
        raise ValueError("adaptive CFG policy schema or key binding differs")
    age = policy["maximum_receipt_age_seconds"]
    if isinstance(age, bool) or not isinstance(age, int) or age < 1:
        raise ValueError("receipt age policy is invalid")
    for tier, rule in policy["tier_actions"].items():
        if not isinstance(rule, dict) or set(rule) != {"action", "maximum_scale"}:
            raise ValueError(f"{tier} rule schema differs")
        if tier == "block":
            if rule != {"action": "BLOCK", "maximum_scale": None}:
                raise ValueError("block tier must deny")
        else:
            value = rule["maximum_scale"]
            if (
                rule["action"] not in {"ALLOW", "CLAMP"}
                or isinstance(value, bool)
                or not isinstance(value, (int, float))
                or not math.isfinite(float(value))
                or float(value) <= 0
            ):
                raise ValueError(f"{tier} scale ceiling is invalid")
            rule["maximum_scale"] = float(value)
    return policy, sha256_bytes(raw)


def classify(prompt: str, request_id: str, policy: dict) -> tuple[dict, str]:
    prompt_sha256 = sha256_bytes(prompt.encode("utf-8"))
    tls_context = ssl.create_default_context(cafile=RISK_CA)
    tls_context.minimum_version = ssl.TLSVersion.TLSv1_2
    tls_context.load_cert_chain(certfile=RISK_CERT, keyfile=RISK_KEY)
    with httpx.Client(
        verify=tls_context,
        timeout=httpx.Timeout(PROMPT_RISK_TOTAL_TIMEOUT_SECONDS, connect=PROMPT_RISK_CONNECT_TIMEOUT_SECONDS),
        follow_redirects=False,
    ) as client:
        with client.stream(
            "POST",
            RISK_URL,
            json={
                "request_id": request_id,
                "prompt": prompt,
                "prompt_sha256": prompt_sha256,
                "policy_version": policy["policy_version"],
            },
        ) as response:
            body = bounded_json_response(response)
    if not isinstance(body, dict) or set(body) != {"receipt"}:
        raise ValueError("risk service response schema differs")
    token = body["receipt"]
    if not isinstance(token, str) or not token:
        raise ValueError("risk receipt must be a non-empty compact JWT")
    try:
        token.encode("ascii")
    except UnicodeEncodeError as exc:
        raise ValueError("risk receipt must be ASCII") from exc
    claims = jwt.decode(
        token,
        RISK_PUBLIC_KEY.read_bytes(),
        algorithms=["EdDSA"],
        issuer=policy["receipt_issuer"],
        audience=policy["receipt_audience"],
        options={"require": [
            "iss", "aud", "iat", "exp", "jti", "request_id",
            "prompt_sha256", "policy_version", "tier",
        ]},
    )
    now = int(time.time())
    if (
        claims["request_id"] != request_id
        or claims["prompt_sha256"] != prompt_sha256
        or claims["policy_version"] != policy["policy_version"]
        or claims["tier"] not in TIERS
        or now - int(claims["iat"]) > policy["maximum_receipt_age_seconds"]
    ):
        raise ValueError("risk receipt is stale or misbound")
    return claims, sha256_bytes(token.encode("utf-8"))


def adaptive_scale(prompt: str, requested_scale: float, request_id: str) -> dict:
    if (
        isinstance(requested_scale, bool)
        or not isinstance(requested_scale, (int, float))
        or not math.isfinite(float(requested_scale))
    ):
        raise ValueError("requested scale is invalid")
    policy, policy_sha256 = load_policy()
    claims, receipt_sha256 = classify(prompt, request_id, policy)
    rule = policy["tier_actions"][claims["tier"]]
    if rule["action"] == "BLOCK":
        return {
            "decision": "BLOCK",
            "request_id": request_id,
            "tier": claims["tier"],
            "risk_receipt_jti": claims["jti"],
            "risk_receipt_sha256": receipt_sha256,
            "policy_version": policy["policy_version"],
            "policy_sha256": policy_sha256,
        }
    applied = min(float(requested_scale), rule["maximum_scale"])
    return {
        "decision": "ALLOW",
        "request_id": request_id,
        "tier": claims["tier"],
        "requested_scale": float(requested_scale),
        "applied_scale": applied,
        "risk_receipt_jti": claims["jti"],
        "risk_receipt_sha256": receipt_sha256,
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
    }
Enforcement and evidence

The trusted sampler wrapper calls adaptive_scale after H-010#1 validation and before denoising. It rejects BLOCK; otherwise it passes only applied_scale, never the caller value. Store the decision, signed receipt, policy and bundle, exact model/config digest, request ID, prompt digest, requested/applied scale, service identity, and timestamp in append-only storage. Do not retain raw prompts in the assurance package unless a separate data-handling policy authorizes it.

Action: Make the receipt verifier the only producer of the sampler scale, fail closed on dependency failure, and gate deployment on independent negative replay.

AID-H-010-G003

Enforce non-removable safety conditioning / negative prompts in the guidance branch.

Use this control only when the deployed diffusion pipeline supports a negative-conditioning branch. Negative conditioning is defense-in-depth: it can reduce unsafe generations but is not an authoritative content-safety decision and must not replace input/output policy enforcement. The mandatory terms come from a signed release policy bound to the exact model and tokenizer; callers cannot delete or reorder them. A pipeline without this branch is NOT_APPLICABLE.

Verify, compose, and prove the effective conditioning
# File: serving/negative_conditioning.py
from __future__ import annotations
import math

import hashlib
import json
import os
import subprocess
import tempfile
import unicodedata
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

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

SAFETY_CONDITIONING_POLICY_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H010_SAFETY_CONDITIONING_POLICY_VERIFY_TIMEOUT_SECONDS")


POLICY = Path(os.environ["NEGATIVE_CONDITIONING_POLICY_PATH"])
BUNDLE = Path(os.environ["NEGATIVE_CONDITIONING_POLICY_BUNDLE_PATH"])
VERIFY_KEY = os.environ["NEGATIVE_CONDITIONING_POLICY_VERIFY_KEY"]
REQUIRED = {
    "schema_version", "policy_version", "model_artifact_sha256",
    "tokenizer_artifact_sha256", "mandatory_negative_terms",
    "maximum_user_negative_bytes", "issued_at", "expires_at",
}


def no_duplicate_keys(pairs: list[tuple[str, object]]) -> dict:
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate policy key: {key}")
        value[key] = item
    return value


def reject_nonfinite(value: str):
    raise ValueError(f"non-finite policy number: {value}")


def strict_json(raw: bytes) -> dict:
    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=no_duplicate_keys,
        parse_constant=reject_nonfinite,
    )
    if not isinstance(value, dict):
        raise ValueError("policy root must be an object")
    return value


def timestamp(value: object) -> datetime:
    if not isinstance(value, str):
        raise ValueError("policy timestamp is absent")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("policy timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def load_policy(model_sha256: str, tokenizer_sha256: str) -> tuple[dict, str]:
    raw = POLICY.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h010-policy-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", VERIFY_KEY,
                "--bundle", str(BUNDLE), str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=SAFETY_CONDITIONING_POLICY_VERIFY_TIMEOUT_SECONDS,
        )
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)
    policy = strict_json(raw)
    if (
        not isinstance(policy, dict)
        or set(policy) != REQUIRED
        or policy["schema_version"] != "aidefend.negative-conditioning.v1"
        or policy["model_artifact_sha256"] != model_sha256
        or policy["tokenizer_artifact_sha256"] != tokenizer_sha256
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
    ):
        raise ValueError("conditioning policy schema or artifact binding differs")
    terms = policy["mandatory_negative_terms"]
    if (
        not isinstance(terms, list)
        or not terms
        or terms != list(dict.fromkeys(terms))
        or any(not isinstance(term, str) or not term.strip() for term in terms)
    ):
        raise ValueError("mandatory terms are empty, duplicated, or invalid")
    limit = policy["maximum_user_negative_bytes"]
    if isinstance(limit, bool) or not isinstance(limit, int) or limit < 0:
        raise ValueError("user-negative byte limit is invalid")
    now = datetime.now(timezone.utc)
    issued, expires = timestamp(policy["issued_at"]), timestamp(policy["expires_at"])
    if issued > now or expires <= issued or now >= expires:
        raise ValueError("conditioning policy is not currently valid")
    return policy, hashlib.sha256(raw).hexdigest()


def compose_negative_conditioning(
    tokenizer: Any,
    user_negative_prompt: str | None,
    request_id: str,
    model_sha256: str,
    tokenizer_sha256: str,
) -> tuple[str, dict]:
    if not request_id:
        raise ValueError("request ID is absent")
    policy, policy_sha256 = load_policy(model_sha256, tokenizer_sha256)
    user_text = unicodedata.normalize("NFKC", user_negative_prompt or "").strip()
    if len(user_text.encode("utf-8")) > policy["maximum_user_negative_bytes"]:
        raise ValueError("user negative prompt exceeds signed byte budget")
    mandatory = ", ".join(
        unicodedata.normalize("NFKC", term).strip()
        for term in policy["mandatory_negative_terms"]
    )
    final_prompt = mandatory if not user_text else mandatory + ", " + user_text
    encoded = tokenizer(
        final_prompt,
        add_special_tokens=True,
        truncation=False,
        return_attention_mask=False,
    )["input_ids"]
    maximum = getattr(tokenizer, "model_max_length", None)
    if isinstance(maximum, bool) or not isinstance(maximum, int) or maximum < 1:
        raise ValueError("tokenizer maximum length is unavailable")
    if len(encoded) > maximum:
        raise ValueError("effective negative conditioning would be truncated")
    receipt = {
        "schema_version": "aidefend.negative-conditioning-receipt.v1",
        "request_id": request_id,
        "model_artifact_sha256": model_sha256,
        "tokenizer_artifact_sha256": tokenizer_sha256,
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
        "mandatory_conditioning_sha256": hashlib.sha256(
            mandatory.encode("utf-8")
        ).hexdigest(),
        "effective_conditioning_sha256": hashlib.sha256(
            final_prompt.encode("utf-8")
        ).hexdigest(),
        "effective_token_count": len(encoded),
        "applied_at": datetime.now(timezone.utc).isoformat(),
    }
    return final_prompt, receipt
Integration, evidence, and outcomes

The last trusted sampler wrapper calls this function and passes only the returned string as negative_prompt. Retain the signed policy and bundle, exact model/tokenizer digests, endpoint population, receipt, sampler request correlation, and effective prompt digest in an append-only store; protect raw user text under the applicable data policy. An independent test uses no user prompt, an adversarial user prompt, Unicode variants, the maximum allowed size, an oversize value, wrong artifact bindings, an expired policy, and a token-overflow case, then reads back the sampler arguments.

Action: Place the composer immediately before the sampler, keep mandatory conditioning first, and gate the release on independent argument readback rather than a self-reported flag.

AID-H-010-G004

Implement alternative guidance formulations that are inherently more robust (e.g. Dynamic CFG / decaying guidance).

Use this control for a diffusion release that exposes a documented public guider component. The example uses the public Hugging Face Diffusers ClassifierFreeGuidance interface with policy-owned start, stop, scale, and rescale values; it does not mutate private pipeline attributes. Pin and validate the tested library version. A pipeline with no supported step-bounded guider is NOT_APPLICABLE, not silently patched with an unstable hook.

Load a signed guider policy and apply it through the public component API
# File: serving/configure_cfg_guider.py
# Pin the exact tested Diffusers release in the deployment lock file.
from __future__ import annotations

import hashlib
import json
import math
import os
import subprocess
import tempfile
from datetime import datetime, timezone
from importlib.metadata import version
from pathlib import Path
from typing import Any

from diffusers import ClassifierFreeGuidance

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

ROBUST_GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H010_ROBUST_GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS")


POLICY = Path(os.environ["CFG_GUIDER_POLICY_PATH"])
BUNDLE = Path(os.environ["CFG_GUIDER_POLICY_BUNDLE_PATH"])
VERIFY_KEY = os.environ["CFG_GUIDER_POLICY_VERIFY_KEY"]
REQUIRED = {
    "schema_version", "policy_version", "diffusers_version",
    "pipeline_config_sha256", "model_artifact_sha256",
    "guidance_scale", "guidance_rescale", "start", "stop",
    "issued_at", "expires_at",
}


def no_duplicate_keys(pairs: list[tuple[str, object]]) -> dict:
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate policy key: {key}")
        value[key] = item
    return value


def reject_nonfinite(value: str):
    raise ValueError(f"non-finite policy number: {value}")


def strict_json(raw: bytes) -> dict:
    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=no_duplicate_keys,
        parse_constant=reject_nonfinite,
    )
    if not isinstance(value, dict):
        raise ValueError("policy root must be an object")
    return value


def timestamp(value: object) -> datetime:
    if not isinstance(value, str):
        raise ValueError("policy timestamp is absent")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("policy timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def load_policy(
    pipeline_config_sha256: str,
    model_artifact_sha256: str,
) -> tuple[dict, str]:
    raw = POLICY.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h010-policy-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", VERIFY_KEY,
                "--bundle", str(BUNDLE), str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=ROBUST_GUIDANCE_POLICY_VERIFY_TIMEOUT_SECONDS,
        )
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)
    policy = strict_json(raw)
    if (
        not isinstance(policy, dict)
        or set(policy) != REQUIRED
        or policy["schema_version"] != "aidefend.cfg-guider-policy.v1"
        or policy["pipeline_config_sha256"] != pipeline_config_sha256
        or policy["model_artifact_sha256"] != model_artifact_sha256
        or policy["diffusers_version"] != version("diffusers")
    ):
        raise ValueError("guider policy schema, artifact, or library binding differs")
    for field in ("guidance_scale", "guidance_rescale", "start", "stop"):
        value = policy[field]
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
        ):
            raise ValueError(f"{field} is invalid")
        policy[field] = float(value)
    if (
        policy["guidance_scale"] <= 0
        or not 0.0 <= policy["guidance_rescale"] <= 1.0
        or not 0.0 <= policy["start"] < policy["stop"] <= 1.0
    ):
        raise ValueError("guider interval or scale is invalid")
    now = datetime.now(timezone.utc)
    issued, expires = timestamp(policy["issued_at"]), timestamp(policy["expires_at"])
    if issued > now or expires <= issued or now >= expires:
        raise ValueError("guider policy is not currently valid")
    if not isinstance(policy["policy_version"], str) or not policy["policy_version"]:
        raise ValueError("policy version is absent")
    return policy, hashlib.sha256(raw).hexdigest()


class UnsupportedGuiderPipelineError(RuntimeError):
    """The selected immutable pipeline adapter cannot enforce and read back a guider."""


def configure_guider(
    pipeline: Any,
    pipeline_config_sha256: str,
    model_artifact_sha256: str,
) -> dict:
    if not callable(getattr(pipeline, "update_components", None)):
        raise UnsupportedGuiderPipelineError("pipeline has no supported guider component API")
    if not callable(getattr(pipeline, "get_component_spec", None)):
        raise UnsupportedGuiderPipelineError("pipeline has no guider readback API")
    policy, policy_sha256 = load_policy(
        pipeline_config_sha256, model_artifact_sha256
    )
    guider = ClassifierFreeGuidance(
        guidance_scale=policy["guidance_scale"],
        guidance_rescale=policy["guidance_rescale"],
        start=policy["start"],
        stop=policy["stop"],
    )
    pipeline.update_components(guider=guider)
    spec = pipeline.get_component_spec("guider")
    effective = dict(spec.config)
    expected = {
        field: policy[field]
        for field in ("guidance_scale", "guidance_rescale", "start", "stop")
    }
    if any(float(effective.get(field, -1)) != value for field, value in expected.items()):
        raise RuntimeError("effective guider differs from signed policy")
    return {
        "schema_version": "aidefend.cfg-guider-receipt.v1",
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
        "diffusers_version": version("diffusers"),
        "pipeline_config_sha256": pipeline_config_sha256,
        "model_artifact_sha256": model_artifact_sha256,
        "guider_class": type(guider).__name__,
        "effective_config": expected,
        "configured_at": datetime.now(timezone.utc).isoformat(),
    }
Evaluation dependency, evidence, and outcomes

H-010 owns the effective guider configuration and readback. Use the signed scenario population and promotion evidence governed by AID-H-007.004 to compare the candidate schedule with the approved baseline for the exact model release; that upstream PASS does not prove this runtime loaded it. Retain the policy/bundle, dependency lock, pipeline/model/config digests, component-spec readback, receipt, scenario manifest, run outputs, and evaluator identity in immutable storage.

Action: Configure only the public guider component, pin the tested runtime, and block promotion until independent runtime readback and evaluation both pass.

Open-source 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-010: Classifier-Free Guidance Hardening | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service