AID-H-005.001: Differential Privacy for AI

Version 1.20261015 · Actionable sub-technique

Implements differential privacy mechanisms to add calibrated noise to model training, outputs, or data queries, ensuring that individual data points cannot be identified while maintaining overall utility.

Pillars: data, model

Lifecycle: building

Scope boundary

Owns quantified differential-privacy mechanisms, clipping and noise calibration, privacy-budget composition, and utility validation for training, outputs, queries, or local collection.

AID-H-005.003: AID-H-005.001 provides a quantified differential-privacy guarantee under an explicit accountant and threat model; AID-H-005.003 provides empirical non-DP memorization and confidence-leakage reduction through regularization. Empirical attack improvement must not be described as a formal DP guarantee.

Warning

High on Training Time & Model Utility

This technique adds computational overhead to each training step by adding noise and clipping gradients.

Training Time: Can slow down the training process by 2x to 5x.

Accuracy: May moderately reduce the final accuracy of the model, which is a direct trade-off for the privacy guarantees provided.

Implementation guidance

AID-H-005.001-G001

Implement DP-SGD with tuned gradient clipping and noise parameters.
Before you begin

Owned jointly by privacy engineering and the training platform for model training that requires a formal central differential-privacy guarantee at the policy-defined privacy unit.

Production implementation

Build the training loop with a maintained DP library that performs true per-sample clipping before calibrated noise addition, reject unsupported layers or non-Poisson/fixed-batch assumptions that conflict with the accountant, reserve the run's privacy budget before reading protected data, record realized sampling and steps, query the accountant after every policy interval, and terminate before the authorized budget can be exceeded. Never continue with a non-DP optimizer after library or accountant failure.

Independent verification

A separately credentialed evaluator replays the accountant from immutable sample-rate, step and noise/clipping evidence; runs the library's clipping/noise and unsupported-layer tests; checks the exact trained artifact and dataset population; and reproduces policy-required privacy and utility metrics on held-out data.

Executable DP-SGD integration
# File: privacy/train_private.py
# Dependencies: torch, opacus
from opacus import PrivacyEngine


def make_private(
    model, optimizer, data_loader, *,
    target_epsilon: float,
    target_delta: float,
    epochs: int,
    max_grad_norm: float,
):
    if not 0 < target_epsilon or not 0 < target_delta < 1:
        raise ValueError("invalid privacy target")
    if epochs < 1 or max_grad_norm <= 0:
        raise ValueError("invalid training or clipping bound")
    engine = PrivacyEngine(accountant="prv")
    private_model, private_optimizer, private_loader = (
        engine.make_private_with_epsilon(
            module=model,
            optimizer=optimizer,
            data_loader=data_loader,
            epochs=epochs,
            target_epsilon=target_epsilon,
            target_delta=target_delta,
            max_grad_norm=max_grad_norm,
        )
    )
    return private_model, private_optimizer, private_loader, engine


def privacy_receipt(engine, delta: float) -> dict:
    epsilon = float(engine.get_epsilon(delta))
    if epsilon < 0:
        raise RuntimeError("privacy accountant returned an invalid epsilon")
    return {"epsilon": epsilon, "delta": delta, "accountant": "prv"}

Load every bound from a signed release policy, retain the Opacus/library lock, dataset and sampling manifest, clipping/noise configuration, accountant state, and final receipt, and independently replay one fixed-seed fixture before promotion.

AID-H-005.001-G002

Apply output perturbation techniques adding calibrated noise to model predictions.
Production implementation

Canonicalize the approved query, enforce contribution and domain bounds before evaluation, atomically reserve budget, compute the aggregate inside the trusted service, invoke the vetted Laplace/Gaussian or policy-selected library mechanism with cryptographic randomness, apply only privacy-safe post-processing, commit the budget and return the noised value. Never log or return the true aggregate, seed, raw contributions or intermediate unnoised output; abort on unsupported sensitivity or ledger failure.

Independent verification

An independent test service recomputes the sensitivity proof and budget charge, executes signed neighboring-dataset and distribution fixtures, verifies no raw aggregate crosses the release boundary, reconciles every query with the authoritative request/budget ledger, and statistically tests the pinned mechanism implementation within policy tolerances.

Executable bounded output perturbation
# File: privacy/output_perturbation.py
from __future__ import annotations

import math
import numpy as np


def privatize_scores(
    scores: np.ndarray,
    *,
    l1_sensitivity: float,
    epsilon: float,
    lower: float,
    upper: float,
    rng: np.random.Generator,
) -> np.ndarray:
    values = np.asarray(scores, dtype=np.float64)
    if (
        values.size == 0
        or not np.isfinite(values).all()
        or not math.isfinite(l1_sensitivity)
        or l1_sensitivity <= 0
        or not math.isfinite(epsilon)
        or epsilon <= 0
        or not math.isfinite(lower)
        or not math.isfinite(upper)
        or lower >= upper
    ):
        raise ValueError("invalid perturbation input or signed policy")
    clipped = np.clip(values, lower, upper)
    noise = rng.laplace(0.0, l1_sensitivity / epsilon, size=clipped.shape)
    released = clipped + noise
    if not np.isfinite(released).all():
        raise RuntimeError("perturbed output is non-finite")
    return released

Use an operating-system-seeded generator in production; inject a deterministic generator only in tests. The sensitivity claim must match the exact clipped query, and the budget ledger in AID-H-005.001-G003 must authorize the release before these values leave the trusted boundary.

AID-H-005.001-G003

Manage privacy budget (epsilon, delta) across multiple queries or training epochs.
Delivery level: reusable single-writer ledger

Put this ledger behind the authenticated DP release service. The request supplies the mechanism cost and idempotency key; it must never supply its own cap. A separately administered policy adapter verifies the signed organizational privacy profile and constructs the immutable BudgetPolicy, including its finite positive SQLite writer-lock timeout. This module then performs exact basic composition under that policy-owned lock deadline.

Use exact finite decimal values and policy-owned caps
# File: privacy/budget_ledger.py
from __future__ import annotations

import math
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation


@dataclass(frozen=True)
class BudgetPolicy:
    policy_version: str
    subject: str
    purpose: str
    epsilon_cap: Decimal
    delta_cap: Decimal
    ledger_lock_timeout_seconds: float


def finite_decimal(name: str, value: str, *, allow_zero: bool) -> Decimal:
    if not isinstance(value, str) or not value:
        raise ValueError(f"{name} must be a decimal string")
    try:
        parsed = Decimal(value)
    except InvalidOperation as error:
        raise ValueError(f"{name} is not a decimal") from error
    if not parsed.is_finite() or parsed < 0 or (not allow_zero and parsed == 0):
        raise ValueError(f"{name} must be finite and {'non-negative' if allow_zero else 'positive'}")
    return parsed


def validate_policy(policy: BudgetPolicy) -> None:
    if not all(
        isinstance(value, str) and value
        for value in (policy.policy_version, policy.subject, policy.purpose)
    ):
        raise ValueError("verified budget policy identity is incomplete")
    if (
        not isinstance(policy.epsilon_cap, Decimal)
        or not isinstance(policy.delta_cap, Decimal)
        or not policy.epsilon_cap.is_finite()
        or not policy.delta_cap.is_finite()
        or policy.epsilon_cap <= 0
        or not Decimal("0") < policy.delta_cap < Decimal("1")
        or isinstance(policy.ledger_lock_timeout_seconds, bool)
        or not isinstance(policy.ledger_lock_timeout_seconds, (int, float))
        or not math.isfinite(float(policy.ledger_lock_timeout_seconds))
        or float(policy.ledger_lock_timeout_seconds) <= 0
    ):
        raise ValueError("verified budget caps are invalid")


def consume(
    database: str,
    *,
    policy: BudgetPolicy,
    release_id: str,
    epsilon_text: str,
    delta_text: str,
) -> dict[str, str]:
    validate_policy(policy)
    if not isinstance(release_id, str) or not release_id:
        raise ValueError("release_id is required")
    epsilon = finite_decimal("epsilon", epsilon_text, allow_zero=True)
    delta = finite_decimal("delta", delta_text, allow_zero=True)
    if delta >= Decimal("1"):
        raise ValueError("delta must be less than one")

    with sqlite3.connect(
        database, isolation_level=None,
        timeout=float(policy.ledger_lock_timeout_seconds),
    ) as connection:
        connection.execute("PRAGMA journal_mode=WAL")
        connection.execute("PRAGMA synchronous=FULL")
        connection.execute("""CREATE TABLE IF NOT EXISTS privacy_releases_v2(
          release_id TEXT PRIMARY KEY,
          subject TEXT NOT NULL,
          purpose TEXT NOT NULL,
          epsilon_text TEXT NOT NULL,
          delta_text TEXT NOT NULL,
          policy_version TEXT NOT NULL,
          recorded_at TEXT NOT NULL
        )""")
        connection.execute("""CREATE TABLE IF NOT EXISTS privacy_release_receipts_v1(
          release_id TEXT PRIMARY KEY,
          composed_epsilon_text TEXT NOT NULL,
          composed_delta_text TEXT NOT NULL,
          FOREIGN KEY(release_id) REFERENCES privacy_releases_v2(release_id)
        )""")
        connection.execute("BEGIN IMMEDIATE")
        try:
            existing = connection.execute(
                "SELECT subject, purpose, epsilon_text, delta_text, policy_version "
                "FROM privacy_releases_v2 WHERE release_id=?",
                (release_id,),
            ).fetchone()
            if existing is not None:
                stored_epsilon = finite_decimal(
                    "stored epsilon", existing[2], allow_zero=True
                )
                stored_delta = finite_decimal(
                    "stored delta", existing[3], allow_zero=True
                )
                if (
                    existing[0] != policy.subject
                    or existing[1] != policy.purpose
                    or stored_epsilon != epsilon
                    or stored_delta != delta
                    or existing[4] != policy.policy_version
                ):
                    raise PermissionError(
                        "release_id reused with a different request"
                    )
                receipt = connection.execute(
                    "SELECT composed_epsilon_text, composed_delta_text "
                    "FROM privacy_release_receipts_v1 WHERE release_id=?",
                    (release_id,),
                ).fetchone()
                if receipt is None:
                    raise RuntimeError(
                        "existing release lacks a reconciled idempotency receipt"
                    )
                composed_epsilon = finite_decimal(
                    "stored composed epsilon", receipt[0], allow_zero=True
                )
                composed_delta = finite_decimal(
                    "stored composed delta", receipt[1], allow_zero=True
                )
                if (
                    composed_epsilon > policy.epsilon_cap
                    or composed_delta > policy.delta_cap
                ):
                    raise PermissionError(
                        "stored reservation exceeds the verified policy cap"
                    )
                connection.execute("COMMIT")
                return {
                    "release_id": release_id,
                    "policy_version": policy.policy_version,
                    "composed_epsilon": str(composed_epsilon),
                    "composed_delta": str(composed_delta),
                }

            rows = connection.execute(
                "SELECT epsilon_text, delta_text FROM privacy_releases_v2 "
                "WHERE subject=? AND purpose=?",
                (policy.subject, policy.purpose),
            ).fetchall()
            used_epsilon = sum((finite_decimal("stored epsilon", row[0], allow_zero=True) for row in rows), Decimal("0"))
            used_delta = sum((finite_decimal("stored delta", row[1], allow_zero=True) for row in rows), Decimal("0"))
            composed_epsilon = used_epsilon + epsilon
            composed_delta = used_delta + delta
            if composed_epsilon > policy.epsilon_cap or composed_delta > policy.delta_cap:
                raise PermissionError("privacy budget exhausted")
            connection.execute(
                "INSERT INTO privacy_releases_v2 VALUES(?,?,?,?,?,?,?)",
                (
                    release_id, policy.subject, policy.purpose,
                    str(epsilon), str(delta), policy.policy_version,
                    datetime.now(timezone.utc).isoformat(),
                ),
            )
            connection.execute(
                "INSERT INTO privacy_release_receipts_v1 VALUES(?,?,?)",
                (release_id, str(composed_epsilon), str(composed_delta)),
            )
            connection.execute("COMMIT")
        except BaseException:
            connection.execute("ROLLBACK")
            raise
    return {
        "release_id": release_id,
        "policy_version": policy.policy_version,
        "composed_epsilon": str(composed_epsilon),
        "composed_delta": str(composed_delta),
    }
Enforcement and readback

Only the privacy service identity may call consume; its signed-policy adapter must reject stale or invalid profiles, including a missing, non-finite, zero, or negative ledger-lock timeout, before constructing BudgetPolicy. An exact retry of the same release ID, subject, purpose, decimal costs, and policy version returns the original reservation without consuming budget again; conflicting reuse, malformed or non-finite cost, unavailable ledger, corrupt stored decimal, or exhausted cap raises and prevents protected-data release. Persist the returned reservation with the mechanism/run receipt, and have an independent reader recompute each subject-purpose composition from the complete ledger. A distributed deployment must replace SQLite with a serializable transactional store while keeping this exact request/idempotency/policy contract.

Action: Invoke the atomic ledger reservation before every privacy-bearing release and deny the release on missing policy, non-idempotent reuse, exhausted budget, transaction error, or failed readback.

AID-H-005.001-G004

Implement local differential privacy for distributed data collection scenarios.
Concept:

For k-ary randomized response, the truthful value is emitted with probability e^epsilon / (e^epsilon + k - 1); otherwise the mechanism chooses uniformly from the other k-1 values. Choosing from the full domain in the false branch adds extra probability to the truth and no longer implements the stated epsilon-LDP mechanism.

Validated local randomizer
# File: privacy_collection/local_dp.py
from __future__ import annotations

import math
import secrets
from collections.abc import Hashable, Sequence
from typing import TypeVar


T = TypeVar("T", bound=Hashable)
RNG = secrets.SystemRandom()


def randomized_response(
    true_value: T,
    possible_values: Sequence[T],
    epsilon: float,
    *,
    maximum_epsilon: float,
) -> T:
    domain = list(possible_values)
    if len(domain) < 2 or len(set(domain)) != len(domain):
        raise ValueError("possible_values must contain at least two unique values")
    if true_value not in domain:
        raise ValueError("true_value must be in possible_values")
    if (
        isinstance(maximum_epsilon, bool)
        or not isinstance(maximum_epsilon, (int, float))
        or not math.isfinite(float(maximum_epsilon))
        or float(maximum_epsilon) <= 0.0
    ):
        raise ValueError("maximum_epsilon must be a finite positive value from signed policy")
    if (
        isinstance(epsilon, bool)
        or not isinstance(epsilon, (int, float))
        or not math.isfinite(float(epsilon))
        or not 0.0 <= float(epsilon) <= float(maximum_epsilon)
    ):
        raise ValueError("epsilon must be within the signed-policy interval")

    epsilon_value = float(epsilon)
    exp_epsilon = math.exp(epsilon_value)
    truth_probability = exp_epsilon / (exp_epsilon + len(domain) - 1)
    if RNG.random() < truth_probability:
        return true_value
    alternatives = [value for value in domain if value != true_value]
    return RNG.choice(alternatives)


def verify_probability_contract(domain_size: int, epsilon: float) -> None:
    if domain_size < 2:
        raise ValueError("domain_size must be at least two")
    exp_epsilon = math.exp(epsilon)
    denominator = exp_epsilon + domain_size - 1
    p_truth = exp_epsilon / denominator
    p_other = 1.0 / denominator
    if not math.isclose(
        p_truth / p_other, exp_epsilon, rel_tol=1e-12, abs_tol=0.0
    ):
        raise AssertionError("randomized-response probability ratio differs")


if __name__ == "__main__":
    colors = ["Red", "Green", "Blue", "Yellow"]
    verify_probability_contract(len(colors), 1.0)
    print(randomized_response("Blue", colors, 1.0, maximum_epsilon=1.0))

Action: Take epsilon and the categorical domain from a versioned collection policy, run the randomizer on the client before transmission, and retain the implementation version, domain digest, epsilon, and probability-contract test. For unsupported epsilon ranges, continuous values, longitudinal composition, or repeated collection, use a reviewed LDP library and privacy accountant; do not extend this single-event example by assumption.

Verify safely

An independent build verifies the shipped artifact and probability ratio, runs domain/boundary/distribution and secure-randomness fixtures, inspects client-to-collector traffic for raw-value leakage, and reconciles eligible collection events with privacy-budget and ingestion ledgers.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-005.001: Differential Privacy for AI | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service