AID-H-005.003: Training-Time Privacy Regularization
Reduce record-level memorization and confidence leakage by applying one governed, non-differentially-private regularization mechanism during training and requiring the defended candidate to improve the policy-selected membership-inference evaluation without exceeding the signed utility budget for the model family. The signed profile selects mixup, calibrated input noise, or label smoothing and binds that mechanism to the candidate and evaluation population.
Pillars: model
Lifecycle: building, validation
Scope boundary
Owns one governed non-differentially-private training regularization profile and empirical evidence that it reduces policy-selected membership-inference risk within the utility budget. It makes no formal privacy guarantee.
AID-H-005.001: AID-H-005.003 provides empirical privacy hardening through mixup, input noise, or label smoothing; AID-H-005.001 provides a quantified differential-privacy guarantee. Use the DP control when a formal epsilon and delta claim is required.
AID-H-005.004: AID-H-005.003 changes model training to reduce memorization risk; AID-H-005.004 removes duplicate and near-duplicate corpus content before training. Data-centric deduplication and model-training regularization address different causes of memorization.
Implementation guidance
- AID-H-005.003-G001: Apply one versioned non-DP training-time privacy regularization profile using mixup, calibrated input noise, or label smoothing.
- AID-H-005.003-G002: Independently compare untreated and regularized models on the same membership-inference split and utility policy.
AID-H-005.003-G001
Apply one versioned non-DP training-time privacy regularization profile using mixup, calibrated input noise, or label smoothing.
Before you begin
Use this control when a model family shows unacceptable membership-inference evidence but does not require a formal differential-privacy claim.
Step 1: Load one signed regularization profile
The training-policy owner supplies a concrete JSON profile with one selected mechanism, all three typed parameter slots, a seed, owner, baseline run, and required evidence set. The loader below verifies an immutable snapshot and rejects an inactive parameter that is nonzero, so a deployment cannot silently combine mechanisms.
Step 2: Verify the profile and apply only its selected mechanism
# File: hardening/privacy_regularized_step.py
from __future__ import annotations
import json, math, os, subprocess, tempfile
from dataclasses import dataclass
from pathlib import Path
import numpy as np
import torch
import torch.nn.functional as F
PROFILE_KEY = Path("/opt/aidefend/trust/privacy-regularization-profile.pub")
def verify_timeout() -> float:
value = float(os.environ["VERIFIED_H005_PROFILE_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(value) or value <= 0: raise ValueError("profile verification timeout must be finite and positive")
return value
@dataclass(frozen=True)
class RegularizationConfig:
mechanism: str
seed: int
mixup_alpha: float
input_noise_stddev: float
label_smoothing: float
def validate(self) -> None:
if self.mechanism not in {"mixup", "input_noise", "label_smoothing"}: raise ValueError("unsupported mechanism")
values = (self.mixup_alpha, self.input_noise_stddev, self.label_smoothing)
if any(isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) for value in values): raise ValueError("regularization parameters must be finite numbers")
if self.mechanism == "mixup" and not (self.mixup_alpha > 0 and self.input_noise_stddev == 0 and self.label_smoothing == 0): raise ValueError("mixup profile must enable only mixup_alpha")
if self.mechanism == "input_noise" and not (self.input_noise_stddev > 0 and self.mixup_alpha == 0 and self.label_smoothing == 0): raise ValueError("input-noise profile must enable only input_noise_stddev")
if self.mechanism == "label_smoothing" and not (0 < self.label_smoothing < 1 and self.mixup_alpha == 0 and self.input_noise_stddev == 0): raise ValueError("label-smoothing profile must enable only label_smoothing")
if isinstance(self.seed, bool) or not isinstance(self.seed, int) or self.seed < 0: raise ValueError("seed must be a non-negative integer")
def load_config(path: str, signature: str) -> RegularizationConfig:
payload = Path(path).read_bytes()
with tempfile.NamedTemporaryFile(prefix="aidefend-privacy-profile-", delete=False) as handle:
handle.write(payload); snapshot = Path(handle.name)
try:
subprocess.run(["cosign", "verify-blob", "--key", str(PROFILE_KEY), "--bundle", signature, str(snapshot)], check=True, capture_output=True, text=True, timeout=verify_timeout())
finally:
snapshot.unlink(missing_ok=True)
raw = json.loads(payload)
fields = {"schema_version", "profile_version", "mechanism", "parameters", "seed", "owner", "baseline_run_id", "required_verification_artifacts"}
if not isinstance(raw, dict) or set(raw) != fields: raise ValueError("regularization profile schema mismatch")
if raw["schema_version"] != "aidefend.training-privacy-regularization.v1" or not isinstance(raw["profile_version"], str) or not raw["profile_version"]: raise ValueError("regularization profile identity is invalid")
if not isinstance(raw["parameters"], dict) or set(raw["parameters"]) != {"mixup_alpha", "input_noise_stddev", "label_smoothing"}: raise ValueError("regularization parameter schema mismatch")
if any(not isinstance(raw[field], str) or not raw[field] for field in ("owner", "baseline_run_id")): raise ValueError("owner and baseline run are required")
expected_evidence = {"task_utility_report", "membership_inference_report", "training_config_digest"}
if not isinstance(raw["required_verification_artifacts"], list) or set(raw["required_verification_artifacts"]) != expected_evidence or len(raw["required_verification_artifacts"]) != len(expected_evidence): raise ValueError("required evidence population differs")
config = RegularizationConfig(mechanism=raw["mechanism"], seed=raw["seed"], **raw["parameters"]); config.validate(); return config
def train_step(model, optimizer, inputs, targets, cfg: RegularizationConfig, step: int) -> float:
cfg.validate(); optimizer.zero_grad(set_to_none=True)
generator = torch.Generator(device=inputs.device).manual_seed(cfg.seed + step)
if cfg.mechanism == "mixup":
rng = np.random.default_rng(cfg.seed + step); lam = float(rng.beta(cfg.mixup_alpha, cfg.mixup_alpha))
permutation = torch.randperm(inputs.size(0), generator=generator, device=inputs.device)
mixed = lam * inputs + (1.0 - lam) * inputs[permutation]; logits = model(mixed)
loss = lam * F.cross_entropy(logits, targets) + (1.0 - lam) * F.cross_entropy(logits, targets[permutation])
elif cfg.mechanism == "input_noise":
noise = torch.randn(inputs.shape, dtype=inputs.dtype, device=inputs.device, generator=generator) * cfg.input_noise_stddev
logits = model(inputs + noise); loss = F.cross_entropy(logits, targets)
else:
logits = model(inputs); loss = F.cross_entropy(logits, targets, label_smoothing=cfg.label_smoothing)
if not torch.isfinite(loss): raise RuntimeError("privacy regularization produced a non-finite loss")
loss.backward(); optimizer.step(); return float(loss.detach().cpu())Step 3: Preserve replay inputs
Store the exact configuration, code commit, data-split digest, model initialization digest, seed policy, and resulting model digest with the training run. If the baseline and candidate do not share the declared comparison inputs, mark the result insufficient_data rather than passed.
Action: Approve one mechanism only after an independent evaluator can replay the baseline/candidate comparison and reproduce both the privacy-attack and utility results. Do not describe mixup, noise, or label smoothing as differential privacy.
Production action and outcome
Action: train an otherwise identical baseline/candidate pair with one signed non-DP regularizer and promote only when every privacy-attack and utility gate passes.
Boundary: Dataset and attack-harness evidence are prerequisites only; H-005.001 owns formal DP, and no prerequisite PASS is inherited.
AID-H-005.003-G002
Independently compare untreated and regularized models on the same membership-inference split and utility policy.
# File: evaluation/evaluate_privacy_regularization.py
from __future__ import annotations
import hashlib
import hmac
import io
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score, roc_curve
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
BOOTSTRAP_MAX_ITERATIONS = _required_positive_int("VERIFIED_H005_BOOTSTRAP_MAX_ITERATIONS")
MEMBERSHIP_ATTACK_MAX_ITERATIONS = _required_positive_int("VERIFIED_H005_MEMBERSHIP_ATTACK_MAX_ITERATIONS")
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
PRIVACY_ARTIFACT_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H005_PRIVACY_ARTIFACT_VERIFY_TIMEOUT_SECONDS")
PRIVACY_VERDICT_SIGN_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H005_PRIVACY_VERDICT_SIGN_TIMEOUT_SECONDS")
PRIVACY_VERDICT_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H005_PRIVACY_VERDICT_VERIFY_TIMEOUT_SECONDS")
POLICY_IDENTITY = "https://github.com/example-org/model-policy/.github/workflows/privacy.yml@refs/heads/main"
SCORE_EVALUATOR_IDENTITY = "https://github.com/example-org/model-eval/.github/workflows/privacy-scores.yml@refs/heads/main"
UTILITY_EVALUATOR_IDENTITY = "https://github.com/example-org/model-eval/.github/workflows/utility.yml@refs/heads/main"
OIDC_ISSUER = "https://token.actions.githubusercontent.com"
VERDICT_VERIFY_KEY = Path("/opt/aidefend/trust/privacy-release-gate.pub")
HEX64 = re.compile(r"^[0-9a-f]{64}$")
POLICY_FIELDS = {
"schema_version", "policy_version", "split_salt", "test_fraction",
"max_candidate_attack_auc", "max_candidate_attack_advantage",
"minimum_attack_auc_improvement", "maximum_utility_drop",
"minimum_train_samples_per_class", "minimum_test_samples_per_class",
"bootstrap_iterations", "bootstrap_confidence_level", "bootstrap_seed",
"utility_metric_id", "utility_value_min", "utility_value_max",
"utility_higher_is_better", "training_code_sha256",
"data_split_sha256", "seed_policy_sha256",
"baseline_profile_sha256", "candidate_profile_sha256",
"evaluation_corpus_sha256", "attack_implementation_sha256",
"score_evaluator_image_sha256", "utility_evaluator_image_sha256",
}
BINDING_COLUMNS = {
"training_code_sha256", "data_split_sha256", "seed_policy_sha256",
"regularization_profile_sha256", "evaluation_corpus_sha256",
"attack_implementation_sha256", "score_evaluator_image_sha256",
}
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-privacy-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=PRIVACY_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 canonical_json_bytes(value: dict) -> bytes:
return (
json.dumps(
value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False,
)
+ "\n"
).encode("utf-8")
def load_signed_canonical_json(
path: Path,
bundle: Path,
identity: str,
expected_fields: set[str],
label: str,
) -> tuple[dict, bytes, bytes]:
raw, verified_bundle_bytes = verified_blob_bytes(path, bundle, identity)
try:
value = json.loads(raw.decode("utf-8", errors="strict"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
if not isinstance(value, dict) or set(value) != expected_fields:
raise ValueError(f"{label} schema differs")
if not hmac.compare_digest(raw, canonical_json_bytes(value)):
raise ValueError(f"{label} is not canonical JSON bytes")
return value, raw, verified_bundle_bytes
def finite_number(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]:
policy, raw, verified_bundle_bytes = load_signed_canonical_json(
path, bundle, POLICY_IDENTITY, POLICY_FIELDS, "privacy policy"
)
if policy["schema_version"] != "aidefend.training-privacy-policy.v3":
raise ValueError("privacy policy version is unsupported")
if not isinstance(policy["policy_version"], str) or not policy["policy_version"].strip():
raise ValueError("policy_version is required")
if not isinstance(policy["split_salt"], str) or len(policy["split_salt"]) < 16:
raise ValueError("split_salt is insufficient")
for field in (
"training_code_sha256", "data_split_sha256", "seed_policy_sha256",
"baseline_profile_sha256", "candidate_profile_sha256",
"evaluation_corpus_sha256", "attack_implementation_sha256",
"score_evaluator_image_sha256", "utility_evaluator_image_sha256",
):
if not isinstance(policy[field], str) or not HEX64.fullmatch(policy[field]):
raise ValueError(f"{field} is not a SHA-256 value")
test_fraction = finite_number(policy["test_fraction"], "test_fraction")
max_auc = finite_number(policy["max_candidate_attack_auc"], "max_candidate_attack_auc")
max_advantage = finite_number(
policy["max_candidate_attack_advantage"], "max_candidate_attack_advantage"
)
minimum_improvement = finite_number(
policy["minimum_attack_auc_improvement"], "minimum_attack_auc_improvement"
)
maximum_utility_drop = finite_number(
policy["maximum_utility_drop"], "maximum_utility_drop"
)
utility_min = finite_number(policy["utility_value_min"], "utility_value_min")
utility_max = finite_number(policy["utility_value_max"], "utility_value_max")
for field in (
"minimum_train_samples_per_class", "minimum_test_samples_per_class",
):
value = policy[field]
if isinstance(value, bool) or not isinstance(value, int) or value < 2:
raise ValueError(f"{field} must be an integer of at least 2")
iterations = policy["bootstrap_iterations"]
if (
isinstance(iterations, bool) or not isinstance(iterations, int)
or not 1 <= iterations <= BOOTSTRAP_MAX_ITERATIONS
):
raise ValueError("bootstrap_iterations must be within [1000,100000]")
confidence = finite_number(
policy["bootstrap_confidence_level"], "bootstrap_confidence_level"
)
if not 0.0 < confidence < 1.0:
raise ValueError("bootstrap_confidence_level must be within [0.90,1)")
seed = policy["bootstrap_seed"]
if isinstance(seed, bool) or not isinstance(seed, int) or seed < 0:
raise ValueError("bootstrap_seed must be a non-negative integer")
if not 0.0 < test_fraction < 1.0:
raise ValueError("test_fraction must be in (0, 1)")
if not 0.5 <= max_auc <= 1.0:
raise ValueError("max_candidate_attack_auc must be in [0.5, 1]")
if not 0.0 <= max_advantage <= 1.0:
raise ValueError("max_candidate_attack_advantage must be in [0, 1]")
if not 0.0 <= minimum_improvement <= 0.5:
raise ValueError("minimum_attack_auc_improvement must be in [0, 0.5]")
if maximum_utility_drop < 0.0 or not utility_min < utility_max:
raise ValueError("utility policy range is invalid")
if maximum_utility_drop > utility_max - utility_min:
raise ValueError("maximum_utility_drop exceeds the metric range")
if not isinstance(policy["utility_metric_id"], str) or not policy["utility_metric_id"]:
raise ValueError("utility_metric_id is required")
if policy["utility_higher_is_better"] is not True:
raise ValueError("implemented utility-drop direction must be signed as higher-is-better")
return (
policy,
hashlib.sha256(raw).hexdigest(),
hashlib.sha256(verified_bundle_bytes).hexdigest(),
)
def expected_bindings(policy: dict, profile_field: str) -> dict[str, str]:
return {
"training_code_sha256": policy["training_code_sha256"],
"data_split_sha256": policy["data_split_sha256"],
"seed_policy_sha256": policy["seed_policy_sha256"],
"regularization_profile_sha256": policy[profile_field],
"evaluation_corpus_sha256": policy["evaluation_corpus_sha256"],
"attack_implementation_sha256": policy["attack_implementation_sha256"],
"score_evaluator_image_sha256": policy["score_evaluator_image_sha256"],
}
def load_scores(
path: Path,
bundle: Path,
expected_model_sha256: str,
expected: dict[str, str],
) -> tuple[pd.DataFrame, str, str]:
raw, verified_bundle_bytes = verified_blob_bytes(
path, bundle, SCORE_EVALUATOR_IDENTITY
)
if (
raw.startswith(b"\xef\xbb\xbf")
or b"\r" in raw
or b"\x00" in raw
or not raw.endswith(b"\n")
):
raise InsufficientData(
f"{path} is not canonical UTF-8/LF score evidence"
)
try:
text = raw.decode("utf-8", errors="strict")
except UnicodeDecodeError as exc:
raise InsufficientData(f"{path} is not strict UTF-8") from exc
frame = pd.read_csv(io.StringIO(text), dtype=str, keep_default_na=False)
required = {
"sample_id", "is_member", "model_sha256", *BINDING_COLUMNS,
}
features = sorted(column for column in frame if column.startswith("score_"))
expected_columns = [
"sample_id", "is_member", "model_sha256",
*sorted(BINDING_COLUMNS), *features,
]
if set(frame.columns) != required | set(features) or list(frame.columns) != expected_columns:
raise InsufficientData(f"{path} score schema or canonical column order differs")
if not features:
raise InsufficientData(f"{path} score columns are missing")
sample_ids = frame["sample_id"].tolist()
if (
any(not value for value in sample_ids)
or sample_ids != sorted(set(sample_ids))
):
raise InsufficientData(
f"{path} sample IDs must be sorted, unique, and non-empty"
)
if set(frame["is_member"]) != {"0", "1"}:
raise InsufficientData(f"{path} must contain member and non-member observations")
frame.loc[:, "is_member"] = frame["is_member"].astype(int)
if set(frame["model_sha256"]) != {expected_model_sha256}:
raise InsufficientData(f"{path} model binding differs")
for field, value in expected.items():
if set(frame[field]) != {value}:
raise InsufficientData(f"{path} {field} binding differs")
numeric = frame[features].apply(pd.to_numeric, errors="coerce").to_numpy(dtype=float)
if not np.isfinite(numeric).all():
raise InsufficientData(f"{path} contains non-finite attack features")
frame.loc[:, features] = numeric
return (
frame.reset_index(drop=True),
hashlib.sha256(raw).hexdigest(),
hashlib.sha256(verified_bundle_bytes).hexdigest(),
)
def split_mask(sample_ids: pd.Series, salt: str, test_fraction: float) -> np.ndarray:
values = []
for sample_id in sample_ids.astype(str):
digest = hashlib.sha256(f"{salt}:{sample_id}".encode()).digest()
values.append(int.from_bytes(digest[:8], "big") / 2**64 < test_fraction)
mask = np.asarray(values, dtype=bool)
if not mask.any() or mask.all():
raise InsufficientData("deterministic attack split has an empty partition")
return mask
def summarize_attack(
y: np.ndarray, probability: np.ndarray, *, choose_orientation: bool = True,
) -> dict:
raw_auc = float(roc_auc_score(y, probability))
oriented = (
probability
if not choose_orientation or raw_auc >= 0.5
else 1.0 - probability
)
effective_auc = float(roc_auc_score(y, oriented))
fpr, tpr, _ = roc_curve(y, oriented)
return {
"raw_attack_auc": raw_auc,
"effective_attack_auc": effective_auc,
"attack_advantage": float(np.max(tpr - fpr)),
"test_samples": int(len(y)),
}
def attack_metrics(
frame: pd.DataFrame, mask: np.ndarray, policy: dict,
) -> tuple[dict, np.ndarray, np.ndarray, dict]:
features = sorted(column for column in frame if column.startswith("score_"))
y = frame["is_member"].to_numpy(dtype=np.int64)
coverage = {}
for name, part, minimum in (
("train", ~mask, policy["minimum_train_samples_per_class"]),
("test", mask, policy["minimum_test_samples_per_class"]),
):
counts = np.bincount(y[part], minlength=2)
coverage[f"{name}_nonmember"] = int(counts[0])
coverage[f"{name}_member"] = int(counts[1])
if int(counts.min()) < minimum:
raise InsufficientData(
f"attack {name} split is below the signed per-class minimum"
)
attack = LogisticRegression(
max_iter=MEMBERSHIP_ATTACK_MAX_ITERATIONS, class_weight="balanced", random_state=policy["bootstrap_seed"]
)
attack.fit(frame.loc[~mask, features], y[~mask])
probability = attack.predict_proba(frame.loc[mask, features])[:, 1]
if not np.isfinite(probability).all():
raise InsufficientData("attack emitted non-finite probabilities")
metrics = summarize_attack(y[mask], probability)
if metrics["raw_attack_auc"] < 0.5:
probability = 1.0 - probability
return metrics, y[mask], probability, coverage
def bootstrap_intervals(
baseline_y: np.ndarray,
baseline_probability: np.ndarray,
candidate_y: np.ndarray,
candidate_probability: np.ndarray,
policy: dict,
) -> dict:
if not np.array_equal(baseline_y, candidate_y):
raise InsufficientData("baseline and candidate test labels differ")
members = np.flatnonzero(baseline_y == 1)
nonmembers = np.flatnonzero(baseline_y == 0)
rng = np.random.default_rng(policy["bootstrap_seed"])
candidate_auc, candidate_advantage, improvement = [], [], []
for _ in range(policy["bootstrap_iterations"]):
indices = np.concatenate([
rng.choice(nonmembers, size=len(nonmembers), replace=True),
rng.choice(members, size=len(members), replace=True),
])
baseline_metrics = summarize_attack(
baseline_y[indices], baseline_probability[indices],
choose_orientation=False,
)
candidate_metrics = summarize_attack(
candidate_y[indices], candidate_probability[indices],
choose_orientation=False,
)
candidate_auc.append(candidate_metrics["effective_attack_auc"])
candidate_advantage.append(candidate_metrics["attack_advantage"])
improvement.append(
baseline_metrics["effective_attack_auc"]
- candidate_metrics["effective_attack_auc"]
)
tail = (1.0 - policy["bootstrap_confidence_level"]) / 2.0
def interval(values: list[float]) -> dict:
return {
"lower": float(np.quantile(values, tail)),
"upper": float(np.quantile(values, 1.0 - tail)),
}
return {
"method": "paired_stratified_percentile_bootstrap",
"confidence_level": policy["bootstrap_confidence_level"],
"iterations": policy["bootstrap_iterations"],
"seed": policy["bootstrap_seed"],
"candidate_attack_auc": interval(candidate_auc),
"candidate_attack_advantage": interval(candidate_advantage),
"attack_auc_improvement": interval(improvement),
}
def load_utility(
path: Path,
bundle: Path,
policy: dict,
baseline_digest: str,
candidate_digest: str,
) -> tuple[dict, float, str, str]:
fields = {
"schema_version", "metric_id", "baseline_utility", "candidate_utility",
"baseline_model_sha256", "candidate_model_sha256",
"evaluation_corpus_sha256", "training_code_sha256",
"data_split_sha256", "seed_policy_sha256",
"baseline_profile_sha256", "candidate_profile_sha256",
"utility_evaluator_image_sha256",
}
report, raw, verified_bundle_bytes = load_signed_canonical_json(
path,
bundle,
UTILITY_EVALUATOR_IDENTITY,
fields,
"utility comparison",
)
if report["schema_version"] != "aidefend.utility-comparison.v2":
raise InsufficientData("utility report version differs")
expected = {
"metric_id": policy["utility_metric_id"],
"baseline_model_sha256": baseline_digest,
"candidate_model_sha256": candidate_digest,
"evaluation_corpus_sha256": policy["evaluation_corpus_sha256"],
"training_code_sha256": policy["training_code_sha256"],
"data_split_sha256": policy["data_split_sha256"],
"seed_policy_sha256": policy["seed_policy_sha256"],
"baseline_profile_sha256": policy["baseline_profile_sha256"],
"candidate_profile_sha256": policy["candidate_profile_sha256"],
"utility_evaluator_image_sha256": policy[
"utility_evaluator_image_sha256"
],
}
if any(report.get(field) != value for field, value in expected.items()):
raise InsufficientData("utility report binding differs")
baseline_value = finite_number(report["baseline_utility"], "baseline_utility")
candidate_value = finite_number(report["candidate_utility"], "candidate_utility")
low = finite_number(policy["utility_value_min"], "utility_value_min")
high = finite_number(policy["utility_value_max"], "utility_value_max")
if not low <= baseline_value <= high or not low <= candidate_value <= high:
raise InsufficientData("utility value is outside the signed metric range")
return (
report,
baseline_value - candidate_value,
hashlib.sha256(raw).hexdigest(),
hashlib.sha256(verified_bundle_bytes).hexdigest(),
)
def evaluate(
baseline_scores: Path,
baseline_scores_bundle: Path,
candidate_scores: Path,
candidate_scores_bundle: Path,
baseline_model: Path,
candidate_model: Path,
utility_report: Path,
utility_report_bundle: Path,
policy_path: Path,
policy_bundle: Path,
) -> dict:
policy, policy_sha256, policy_bundle_sha256 = load_policy(
policy_path, policy_bundle
)
baseline_digest = sha256_file(baseline_model)
candidate_digest = sha256_file(candidate_model)
(
baseline,
baseline_scores_sha256,
baseline_scores_bundle_sha256,
) = load_scores(
baseline_scores,
baseline_scores_bundle,
baseline_digest,
expected_bindings(policy, "baseline_profile_sha256"),
)
(
candidate,
candidate_scores_sha256,
candidate_scores_bundle_sha256,
) = load_scores(
candidate_scores,
candidate_scores_bundle,
candidate_digest,
expected_bindings(policy, "candidate_profile_sha256"),
)
identity = [
"sample_id", "is_member", "evaluation_corpus_sha256",
"attack_implementation_sha256", "seed_policy_sha256",
]
if not baseline[identity].equals(candidate[identity]):
raise InsufficientData("baseline and candidate populations or bindings differ")
if sorted(c for c in baseline if c.startswith("score_")) != sorted(
c for c in candidate if c.startswith("score_")
):
raise InsufficientData("baseline and candidate score schemas differ")
mask = split_mask(
baseline["sample_id"], policy["split_salt"], float(policy["test_fraction"])
)
baseline_metrics, baseline_y, baseline_probability, sample_counts = attack_metrics(
baseline, mask, policy
)
candidate_metrics, candidate_y, candidate_probability, candidate_counts = attack_metrics(
candidate, mask, policy
)
if sample_counts != candidate_counts:
raise InsufficientData("baseline and candidate split coverage differs")
uncertainty = bootstrap_intervals(
baseline_y, baseline_probability, candidate_y, candidate_probability, policy
)
(
utility,
utility_drop,
utility_report_sha256,
utility_report_bundle_sha256,
) = load_utility(
utility_report,
utility_report_bundle,
policy,
baseline_digest,
candidate_digest,
)
improvement = (
baseline_metrics["effective_attack_auc"]
- candidate_metrics["effective_attack_auc"]
)
failures = []
if (
candidate_metrics["effective_attack_auc"]
> float(policy["max_candidate_attack_auc"])
or uncertainty["candidate_attack_auc"]["upper"]
> float(policy["max_candidate_attack_auc"])
):
failures.append("candidate_attack_auc")
if (
candidate_metrics["attack_advantage"]
> float(policy["max_candidate_attack_advantage"])
or uncertainty["candidate_attack_advantage"]["upper"]
> float(policy["max_candidate_attack_advantage"])
):
failures.append("candidate_attack_advantage")
if (
improvement < float(policy["minimum_attack_auc_improvement"])
or uncertainty["attack_auc_improvement"]["lower"]
< float(policy["minimum_attack_auc_improvement"])
):
failures.append("privacy_improvement")
if utility_drop > float(policy["maximum_utility_drop"]):
failures.append("utility_budget")
return {
"schema_version": "aidefend.training-privacy-evaluation.v4",
"status": "fail" if failures else "pass",
"failures": failures,
"policy_version": policy["policy_version"],
"policy_sha256": policy_sha256,
"policy_bundle_sha256": policy_bundle_sha256,
"baseline_model_sha256": baseline_digest,
"candidate_model_sha256": candidate_digest,
"baseline_scores_sha256": baseline_scores_sha256,
"baseline_scores_bundle_sha256": baseline_scores_bundle_sha256,
"candidate_scores_sha256": candidate_scores_sha256,
"candidate_scores_bundle_sha256": candidate_scores_bundle_sha256,
"utility_report_sha256": utility_report_sha256,
"utility_report_bundle_sha256": utility_report_bundle_sha256,
"comparison_bindings": {
field: policy[field] for field in (
"training_code_sha256", "data_split_sha256",
"seed_policy_sha256", "baseline_profile_sha256",
"candidate_profile_sha256", "evaluation_corpus_sha256",
"attack_implementation_sha256",
"score_evaluator_image_sha256",
"utility_evaluator_image_sha256",
)
},
"baseline_attack": baseline_metrics,
"candidate_attack": candidate_metrics,
"attack_auc_improvement": improvement,
"sample_counts": sample_counts,
"uncertainty": uncertainty,
"utility_drop": utility_drop,
"utility_metric_id": utility["metric_id"],
}
def evaluate_with_status(**kwargs) -> dict:
try:
outcome = evaluate(**kwargs)
outcome["input_evidence_state"] = "verified"
return outcome
except InsufficientData as exc:
return {
"schema_version": "aidefend.training-privacy-evaluation.v4",
"status": "insufficient_data",
"failures": [],
"reason": str(exc),
"input_evidence_state": "unverified",
"verified_input_digests": None,
}
except Exception as exc:
return {
"schema_version": "aidefend.training-privacy-evaluation.v4",
"status": "error",
"failures": [],
"reason": "privacy 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 evaluate_and_sign(
*,
verdict_path: Path,
verdict_signature_path: Path,
**evaluation_inputs,
) -> dict:
if verdict_path.exists() or verdict_signature_path.exists():
raise FileExistsError("verdict publish path already exists")
outcome = evaluate_with_status(**evaluation_inputs)
payload = canonical_json_bytes(outcome)
expected_digest = hashlib.sha256(payload).hexdigest()
with tempfile.TemporaryDirectory(
prefix="aidefend-privacy-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://PRIVACY_GATE_SIGNING_KEY",
"--bundle", str(signature_snapshot),
str(payload_snapshot),
], check=True, capture_output=True, text=True, timeout=PRIVACY_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=PRIVACY_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")
# Publish the signature first and the payload last; payload creation is
# the commit marker consumers use before attempting 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
Production action and outcome
Action: compare the signed baseline and candidate on the identical population and block promotion unless every privacy-attack and utility threshold passes.
Boundary: This empirical non-DP test never inherits dataset or attack-harness PASS and never establishes formal DP.
Open-source tools
- PyTorch
- TensorFlow
- Albumentations (data augmentation)
- torchvision.transforms (data augmentation)
- Adversarial Robustness Toolbox (ART)
- MLflow
- Privacy Meter (membership-inference auditing for regularization validation; does not implement regularization)
- tensorflow-empirical-privacy (membership-inference auditing for regularization validation; does not implement regularization)
Commercial / hosted tools
- Amazon SageMaker AI
- Google Vertex AI
- Databricks
- Azure Machine Learning
Threat mappings
MITRE ATLAS
- AML.T0024 Exfiltration via AI Inference API
- AML.T0024.000 Exfiltration via AI Inference API: Infer Training Data Membership
MAESTRO
- Membership Inference Attacks (L1)
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure
OWASP ML Top 10 2023
- ML04:2023 Membership Inference Attack
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- NISTAML.033 Membership Inference
Cisco Integrated AI Security and Safety Framework
- AISubtech-8.1.1 Presence Detection
- AITech-8.1 Membership Inference
Google Secure AI Framework 2.0 - Risks
- SDD: Sensitive Data Disclosure (training regularization is independently checked for reduced membership disclosure)
Databricks AI Security Framework 3.0
- Model Serving - Inference requests 9.5: Infer training data membership