AID-H-002.001: Training, Fine-Tuning & Evaluation Data Sanitization
Pillars: data
Lifecycle: building, validation
Scope boundary
Owns content-level sanitization of the complete candidate training, fine-tuning, or evaluation dataset before admission. Evaluation-data preparation also applies when using an existing model without training or fine-tuning it; each method retains its own data and runtime prerequisites. It does not establish supplier provenance, cryptographic artifact integrity, feature-pipeline integrity, label integrity, or split isolation.
AID-H-002.004: AID-H-002.001 cleans and admits raw records; AID-H-002.004 verifies the feature transformations and stores that convert admitted records into model inputs. Sanitized source data can still be corrupted by transformation logic or training-serving skew.
AID-H-002.005: AID-H-002.001 inspects the content of candidate records; AID-H-002.005 verifies the provenance and integrity of their supervisory labels and preference signals. A record can look benign while its label has been flipped or its annotator evidence forged.
AID-H-003.003: AID-H-002.001 decides whether dataset contents are safe enough for training or evaluation use; AID-H-003.003 decides whether the external supplier, provenance, license, signature, and lineage satisfy supply-chain admission. Content sanitization and source admission are independent requirements.
AID-H-007.004: AID-H-002.001 sanitizes candidate evaluation records before use; AID-H-007.004 protects evaluation independence, corpus sufficiency, holdout isolation, and promotion evidence. Sanitized records do not establish representative test coverage or trustworthy evaluation results.
AID-H-002.009: AID-H-002.001 sanitizes the records that may influence training or evaluation; AID-H-002.009 admits the dataset container and parser result against hostile format features. A dataset can contain benign records yet exploit its parser, and a parser-safe artifact can still contain poisoned or privacy-sensitive records.
Implementation guidance
- AID-H-002.001-G001: Perform EDA to understand distributions and flag outliers for human review.
- AID-H-002.001-G002: Validate data against schema and constraints with policy-as-code.
- AID-H-002.001-G003: Fit reference-bound generic and multimodal anomaly detectors, then quarantine anomalous records or paired/grouped fusion evidence before training.
- AID-H-002.001-G004: Scan and anonymize PII or sensitive data before training, fine-tuning, or evaluation.
AID-H-002.001-G001
Before you begin
Apply AID-H-002.001-G001 only when a training, fine-tuning, alignment, or evaluation snapshot contains records or files not created wholly inside the trusted build boundary.
Verify safely
An independent verifier replays the same versioned policy against the complete applicable population and compares the local output, enforcement effect, and authoritative readback; an upstream PASS is dependency evidence, not this guidance's result.
Concept:
Start with a human-in-the-loop pass to understand data shape and obvious anomalies. Outliers may be poison, corruption, or OOD drift.
Step 1: Profile the dataset
# File: data_pipeline/scripts/01_profile_data.py
import pandas as pd
from ydata_profiling import ProfileReport
df = pd.read_csv("data/raw/user_feedback.csv")
profile = ProfileReport(df, title="User Feedback Dataset Profiling")
profile.to_file("data_quality_report.html")
print("Report: data_quality_report.html")
Step 2: Programmatically flag candidates
# File: data_pipeline/scripts/02_find_outliers.py
import math
import os
import pandas as pd
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
OUTLIER_Z_SCORE_THRESHOLD = _required_positive_float("VERIFIED_H002_OUTLIER_Z_SCORE_THRESHOLD")
def find_numerical_outliers(df, column, threshold: float):
z = (df[column] - df[column].mean()) / df[column].std(ddof=1)
return df[(z.abs() > threshold)]
df = pd.read_csv("data/processed/reviews.csv")
outliers = find_numerical_outliers(df, "review_length", threshold=OUTLIER_Z_SCORE_THRESHOLD)
outliers.to_csv("data/review_queue/length_outliers.csv", index=False)
print("Queued: data/review_queue/length_outliers.csv")
Action: Gate admission to the selected training or evaluation job on sign-off of the EDA report and reviewed outlier queue.
Keep the action safely bounded
deny snapshot admission on any unaccounted object, unsupported parser, failed dependency, or policy violation; quarantine only outside admitted training and evaluation namespaces.
AID-H-002.001-G002
Before you begin
Apply AID-H-002.001-G002 only when a training, fine-tuning, alignment, or evaluation snapshot contains records or files not created wholly inside the trusted build boundary.
Production implementation: Great Expectations 1.19 dataframe admission
Use the current Fluent Data Source API. Validate the entire immutable snapshot dataframe, not a sample, and persist the validation result beside the population manifest before creating an admitted snapshot.
# File: data_pipeline/validate_snapshot.py
from __future__ import annotations
import hashlib
import json
from pathlib import Path
import great_expectations as gx
import pandas as pd
REQUIRED_COLUMNS = ["user_id", "email", "signup_date", "country_code"]
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 validate_snapshot(csv_path: Path, result_path: Path) -> dict:
if csv_path.is_symlink() or not csv_path.is_file():
raise ValueError("candidate snapshot must be one regular non-symlink file")
snapshot_sha256 = sha256_file(csv_path)
frame = pd.read_csv(csv_path)
if frame.empty or frame.index.has_duplicates:
raise ValueError("snapshot is empty or dataframe index is duplicated")
context = gx.get_context(mode="ephemeral")
source = context.data_sources.add_pandas(name="training_candidate")
asset = source.add_dataframe_asset(name="candidate_records")
batch_definition = asset.add_batch_definition_whole_dataframe(
name="complete_snapshot"
)
suite = context.suites.add(gx.ExpectationSuite(name="training_snapshot_contract"))
suite.add_expectation(
gx.expectations.ExpectTableColumnsToMatchOrderedList(
column_list=REQUIRED_COLUMNS
)
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeUnique(column="user_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToMatchRegex(
column="email", regex=r"^[^@\s]+@[^@\s]+\.[^@\s]+$"
)
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeInSet(
column="country_code",
value_set=["US", "CA", "GB", "AU", "DE", "FR"],
)
)
validation = context.validation_definitions.add(gx.ValidationDefinition(
name="candidate_snapshot_admission",
data=batch_definition,
suite=suite,
))
result = validation.run(batch_parameters={"dataframe": frame})
receipt = {
"schema_version": "aidefend.dataset-schema-validation.v1",
"snapshot_sha256": snapshot_sha256,
"population_rows": int(len(frame)),
"evaluated_rows": int(len(frame)),
"success": bool(result.success),
"statistics": dict(result.statistics),
}
result_path.parent.mkdir(parents=True, exist_ok=True)
with result_path.open("x", encoding="utf-8") as handle:
json.dump(receipt, handle, sort_keys=True, separators=(",", ":"))
if not result.success:
raise SystemExit("complete snapshot failed schema admission")
return receipt
Verification: A separate admission worker re-hashes the source file, verifies the signed schema-policy and receipt, confirms population_rows == evaluated_rows, and reruns the validation before granting the selected training or evaluation job read access.
Keep the action safely bounded
deny snapshot admission on any unaccounted object, unsupported parser, failed dependency, or policy violation; quarantine only outside admitted training and evaluation namespaces.
AID-H-002.001-G003
Prepare a trusted feature bundle
Use the AID-I-001 broker for bounded artifact capture, admitted parser execution, secretless feature extraction, network restrictions, execution deadlines and teardown. Consume its signed run receipt; this content control does not reimplement or re-score sandbox isolation. AID-H-002.009 owns hostile dataset-format parsing. The feature producer uses a digest-pinned approved model/extractor release and signs the complete input bundle. Candidate records cannot select feature names, reference data, detector parameters or thresholds.
The example below accepts captured JSON tables rather than running a Parquet/media parser in the admission process. The producer exports schema_version aidefend.anomaly-input-bundle.v2, snapshot_id, policy_sha256, dataset_sha256, extractor_release_sha256, groups, reference_members, candidate_members, reference_groups and candidate_groups. Each group has group_id and members with record_id, modality and original-record sha256. Member tables bind record/reference ID, group ID where applicable, modality and policy-named numeric features; group tables bind group/reference-group ID and joint/fusion features. The signed dataset manifest supplies the complete expected record/group population. The trusted capture adapter compares it before signing the bundle; omission cannot shrink the measured denominator.
Calibrate the policy before candidate admission
The signed policy supplies version, paired, modalities, generic_features, joint_features, fusion_features, extractor_release_sha256, seed, estimators, minimum_reference_rows, maximum_reference_rows, maximum_records, maximum_members_per_group, maximum_features, generic_minimum, joint_minimum and fusion_z_maximum. Fit only on a retained clean reference. Choose thresholds using an independent benign/attack suite with an acceptable false-positive budget. paired=false disables only the convergence-specific method; generic record screening still applies. A paired profile requires every declared modality per group.
Score complete groups and publish an allowlist
# File: sanitization/dataset_anomalies.py
from __future__ import annotations
import argparse
import hashlib
import json
import math
import subprocess
import tempfile
from pathlib import Path
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import RobustScaler
class InsufficientData(ValueError):
pass
def canonical(value):
return json.dumps(value, sort_keys=True, separators=(",", ":"),
allow_nan=False).encode()
def sha(value):
return hashlib.sha256(value).hexdigest()
def matrix(rows, names, minimum, maximum):
if (not isinstance(rows, list) or not minimum <= len(rows) <= maximum
or not names or len(set(names)) != len(names)):
raise InsufficientData("feature population is empty, insufficient or unbounded")
values = []
for row in rows:
vector = [row.get(name) for name in names]
if any(type(x) not in (int, float) or not math.isfinite(x) for x in vector):
raise InsufficientData("missing or nonfinite feature")
values.append(vector)
return np.asarray(values, dtype=np.float64)
def isolation_scores(reference, candidate, names, policy):
ref = matrix(reference, names, policy["minimum_reference_rows"], policy["maximum_reference_rows"])
cand = matrix(candidate, names, 1, policy["maximum_records"])
detector = make_pipeline(RobustScaler(), IsolationForest(
n_estimators=policy["estimators"], random_state=policy["seed"],
contamination="auto", n_jobs=1))
detector.fit(ref)
scores = detector.decision_function(cand)
if not np.isfinite(scores).all():
raise InsufficientData("nonfinite anomaly score")
return scores
def exact_ids(rows, field):
ids = [row[field] for row in rows]
if any(not isinstance(x, str) or not x for x in ids) or len(ids) != len(set(ids)):
raise InsufficientData("duplicate or invalid feature identity")
return set(ids)
def score_bundle(bundle, policy):
"""Inputs are already captured, signed and version-bound by the launcher.
The content scorer consumes shared isolation receipts; it does not launch
Docker, change cgroups or attest isolation. Features are produced from the
admitted model/extractor release, not supplied by candidate record authors.
"""
for key in ("minimum_reference_rows", "maximum_reference_rows", "maximum_records",
"maximum_members_per_group", "maximum_features", "estimators"):
if type(policy[key]) is not int or policy[key] < 1:
raise ValueError("invalid positive policy bound")
if type(policy["seed"]) is not int or policy["minimum_reference_rows"] > policy["maximum_reference_rows"]:
raise ValueError("invalid seed/reference policy")
for key in ("generic_minimum", "joint_minimum", "fusion_z_maximum"):
if type(policy[key]) not in (int, float) or not math.isfinite(policy[key]):
raise ValueError("invalid calibrated threshold")
if policy["fusion_z_maximum"] <= 0:
raise ValueError("fusion threshold must be positive")
for key in ("generic_features", "joint_features", "fusion_features"):
names = policy[key]
if (not isinstance(names, list) or not 1 <= len(names) <= policy["maximum_features"]
or any(not isinstance(x, str) or not x for x in names) or len(names) != len(set(names))):
raise ValueError("feature names must come from the approved profile")
if type(policy["paired"]) is not bool or not policy["modalities"]:
raise ValueError("explicit architecture/modality profile required")
if bundle["schema_version"] != "aidefend.anomaly-input-bundle.v2":
raise ValueError("unsupported bundle schema")
groups = bundle["groups"]
if not isinstance(groups, list) or not 1 <= len(groups) <= policy["maximum_records"]:
raise InsufficientData("candidate group population bound")
group_ids = exact_ids(groups, "group_id")
expected, record_hashes = {}, {}
for group in groups:
members = group["members"]
if not 1 <= len(members) <= policy["maximum_members_per_group"]:
raise InsufficientData("invalid group membership")
modalities = set()
for member in members:
rid, modality, content_digest = member["record_id"], member["modality"], member["sha256"]
if (not isinstance(rid, str) or not rid or rid in expected
or modality not in policy["modalities"]
or not isinstance(content_digest, str) or len(content_digest) != 64
or set(content_digest) - set("0123456789abcdef")):
raise InsufficientData("record identity/digest or modality mismatch")
expected[rid] = (group["group_id"], modality)
record_hashes[rid] = content_digest
modalities.add(modality)
if policy["paired"] and modalities != set(policy["modalities"]):
raise InsufficientData("paired/grouped modality population is incomplete")
if len(expected) > policy["maximum_records"]:
raise InsufficientData("record population exceeds policy")
candidates, references = bundle["candidate_members"], bundle["reference_members"]
if exact_ids(candidates, "record_id") != set(expected):
raise InsufficientData("candidate features omit or add dataset records")
if any(expected[row["record_id"]] != (row["group_id"], row["modality"]) for row in candidates):
raise InsufficientData("candidate group/modality binding differs")
exact_ids(references, "reference_id")
if set(row["modality"] for row in references) != set(policy["modalities"]):
raise InsufficientData("clean reference modality population differs")
scores = {}
for modality in sorted(set(policy["modalities"])):
ref = [row for row in references if row["modality"] == modality]
cand = [row for row in candidates if row["modality"] == modality]
if not cand:
continue
measured = isolation_scores(ref, cand, policy["generic_features"], policy)
scores.update(zip([row["record_id"] for row in cand], map(float, measured)))
joint, fusion = {}, {}
if policy["paired"]:
ref, cand = bundle["reference_groups"], bundle["candidate_groups"]
exact_ids(ref, "reference_group_id")
if exact_ids(cand, "group_id") != group_ids:
raise InsufficientData("joint/fusion feature group population differs")
joint = dict(zip([row["group_id"] for row in cand], map(float,
isolation_scores(ref, cand, policy["joint_features"], policy))))
rf = matrix(ref, policy["fusion_features"], policy["minimum_reference_rows"], policy["maximum_reference_rows"])
cf = matrix(cand, policy["fusion_features"], 1, policy["maximum_records"])
median = np.median(rf, axis=0)
mad = np.median(np.abs(rf - median), axis=0)
if np.any(mad <= 0):
raise InsufficientData("zero-variance fusion feature requires recalibration")
z = np.max(np.abs(cf - median)/(1.4826 * mad), axis=1)
fusion = dict(zip([row["group_id"] for row in cand], map(float, z)))
evidence, admitted, quarantine = [], {}, {}
for group in groups:
gid = group["group_id"]
ids = sorted(member["record_id"] for member in group["members"])
minimum = min(scores[rid] for rid in ids)
reasons = []
if minimum < policy["generic_minimum"]:
reasons.append("generic_record_anomaly")
if policy["paired"]:
if joint[gid] < policy["joint_minimum"]:
reasons.append("joint_embedding_anomaly")
if fusion[gid] > policy["fusion_z_maximum"]:
reasons.append("fusion_layer_anomaly")
evidence.append({"group_id": gid, "record_ids": ids,
"member_scores": {rid: scores[rid] for rid in ids},
"joint_score": joint.get(gid), "fusion_z": fusion.get(gid),
"reasons": reasons, "decision": "QUARANTINE" if reasons else "ADMIT"})
target = quarantine if reasons else admitted
target.update({rid: record_hashes[rid] for rid in ids})
return {"schema_version": "aidefend.dataset-admission.v2", "snapshot_id": bundle["snapshot_id"],
"policy_version": policy["version"], "policy_sha256": bundle["policy_sha256"],
"dataset_sha256": bundle["dataset_sha256"], "input_bundle_sha256": sha(canonical(bundle)),
"outcome": "PASS", "record_count": len(expected), "group_count": len(groups),
"convergence_method": "MEASURED" if policy["paired"] else "NOT_APPLICABLE",
"admitted": admitted, "quarantined": quarantine, "evidence": evidence}
def verify_captured(path, bundle_path, key, maximum, timeout):
# The shared broker has already exported these bounded regular files.
raw, proof = path.read_bytes(), bundle_path.read_bytes()
if not raw or not proof or max(len(raw), len(proof)) > maximum:
raise ValueError("broker export exceeds the admitted byte bound")
with tempfile.TemporaryDirectory() as directory:
payload, signature = Path(directory)/"payload", Path(directory)/"proof"
payload.write_bytes(raw)
signature.write_bytes(proof)
subprocess.run(["cosign", "verify-blob", "--key", key, "--bundle", str(signature),
str(payload)], check=True, capture_output=True, timeout=timeout)
def unique(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate signed JSON key")
value[key] = item
return value
return raw, json.loads(raw, object_pairs_hook=unique,
parse_constant=lambda x: (_ for _ in ()).throw(ValueError("nonfinite JSON")))
def main():
parser = argparse.ArgumentParser()
for name in ("policy", "input", "receipt", "output"):
parser.add_argument("--"+name, type=Path, required=True)
for name in ("policy-sha256", "dataset-sha256", "run-id", "policy-key", "extractor-key", "isolation-key"):
parser.add_argument("--"+name, required=True)
parser.add_argument("--maximum-bytes", type=int, required=True)
parser.add_argument("--timeout-seconds", type=float, required=True)
args = parser.parse_args()
if args.maximum_bytes < 1 or not math.isfinite(args.timeout_seconds) or args.timeout_seconds <= 0:
raise ValueError("verified bootstrap bounds required")
try:
loaded = []
for path, key in [(args.policy, args.policy_key), (args.input, args.extractor_key), (args.receipt, args.isolation_key)]:
loaded.append(verify_captured(path, Path(str(path)+".sigstore.json"), key,
args.maximum_bytes, args.timeout_seconds))
(policy_raw, policy), (input_raw, bundle), (_, receipt) = loaded
if (sha(policy_raw) != args.policy_sha256 or bundle["policy_sha256"] != args.policy_sha256
or bundle["dataset_sha256"] != args.dataset_sha256
or bundle["extractor_release_sha256"] != policy["extractor_release_sha256"]
or receipt["schema_version"] != "aidefend.sandbox-consumption.v1"
or receipt["owner_control"] != "AID-I-001"
or receipt["outcome"] != "PASS" or receipt["run_id"] != args.run_id
or receipt["input_bundle_sha256"] != sha(input_raw)
or receipt["policy_sha256"] != args.policy_sha256
or receipt["dataset_sha256"] != args.dataset_sha256
or receipt["extractor_release_sha256"] != policy["extractor_release_sha256"]):
raise ValueError("content, extractor or shared isolation receipt binding differs")
result = score_bundle(bundle, policy)
result["captured_input_sha256"] = sha(input_raw)
result["isolation_run_id"] = args.run_id
except Exception as error:
result = {"schema_version": "aidefend.dataset-admission.v2",
"outcome": "INSUFFICIENT_DATA" if isinstance(error, InsufficientData) else "ERROR",
"reason_code": type(error).__name__, "admitted": {}, "quarantined": {}}
args.output.write_bytes(canonical(result)+b"\n")
if result["outcome"] != "PASS":
raise SystemExit(1)
if __name__ == "__main__":
main()Consume shared isolation evidence and enforce admission
The site-owned broker adapter exports aidefend.sandbox-consumption.v1 with owner_control AID-I-001, outcome, run_id, policy_sha256, dataset_sha256, extractor_release_sha256 and input_bundle_sha256 after verifying its native isolation evidence. This is an adapter contract, not a new vendor API or a second sandbox implementation. Its signer is independent of the dataset author. The trusted launcher supplies --policy-sha256, --dataset-sha256 and --run-id from the admitted snapshot plus the three pinned authority keys and policy-derived size/deadline bounds. It cannot choose those values from the input bundle. Execute the scorer under the same broker resource/deadline profile; a killed or incomplete run cannot publish admission.
Write the result to a new immutable staging object, sign it with the dataset-admission authority and independently verify its signature/readback before publishing. The training/evaluation loader must fetch only record IDs and exact digests in the verified admitted map, never a raw candidate-directory glob. Quarantine removes the entire affected group from that allowlist and retains original immutable record references for investigation; it does not delete source evidence. Compare the signed allowlist to the exact loader population before the first training step. Source pointers or mutable tags cannot substitute for record digests.
Replay the core defense independently
Replay a benign group, a complete joint trigger, and a partial-payload group whose individual generic scores remain benign but whose combined joint/fusion evidence is anomalous. Both trigger fixtures must quarantine the entire group for a convergence reason. Check missing/extra/duplicate records and groups, wrong modality, feature/extractor/profile drift, changed input bytes, forged or wrong-run receipts, zero-MAD fusion features, nonfinite values and broker timeout. Missing features/reference coverage produces INSUFFICIENT_DATA; signature, binding or execution errors produce ERROR and no loader publication. PASS means the complete candidate population received an admission/quarantine decision, not that all data is benign or poisoning is impossible. Preserve calibrated thresholds, model/extractor versions, dataset digests, raw features, scores and loader readback for independent replay.
AID-H-002.001-G004
Concept:
Reduce privacy leakage by detecting and replacing the exact PII classes selected by a signed, versioned policy before a dataset snapshot becomes readable by training or evaluation jobs. Bind the language, NLP model, recognizer threshold, columns, entity population, and reviewed canary set to that policy; otherwise a weakened runtime setting can incorrectly certify its own output. Presidio combines recognizers with an NLP engine, so pin the configured model and add organization-specific recognizers for identifiers the default catalog does not cover.
Verify policy and run the bounded anonymization job
Before this job reads the PII policy, a bootstrap identity must signature-verify a separate versioned runtime profile and export AIDEFEND_H002_PII_POLICY_VERIFY_TIMEOUT_SECONDS. Keep the bootstrap profile version and digest with the job evidence. Never derive this deadline from the PII policy whose signature the subprocess is verifying, because that would make timeout enforcement circular.
# File: data_pipeline/scripts/05_anonymize_pii.py
from __future__ import annotations
from collections import Counter
import hashlib
from importlib.metadata import version
import io
import json
import math
import os
from pathlib import Path
import subprocess
import tempfile
import pandas as pd
from presidio_analyzer import AnalyzerEngine
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine
SOURCE = Path("data/raw/customer_support_chats.csv")
OUTPUT = Path("data/processed/chats_anonymized.csv")
POLICY_PATH = Path(os.environ["PII_POLICY_PATH"])
POLICY_BUNDLE_PATH = Path(os.environ["PII_POLICY_BUNDLE_PATH"])
CANARIES = Path("data/policy/pii_canaries.json")
TRUST_IDENTITY = Path("/opt/aidefend/trust/pii-policy.identity")
TRUST_ISSUER = Path("/opt/aidefend/trust/pii-policy.issuer")
def required_positive_seconds(name: str) -> float:
raw = os.environ.get(name)
if raw is None or not raw.strip():
raise RuntimeError(f"{name} is required from the verified bootstrap profile")
try:
value = float(raw)
except ValueError as exc:
raise RuntimeError(f"{name} must be numeric") from exc
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"{name} must be a finite positive number")
return value
VERIFY_TIMEOUT_SECONDS = required_positive_seconds(
"AIDEFEND_H002_PII_POLICY_VERIFY_TIMEOUT_SECONDS"
)
POLICY_FIELDS = {
"schema_version", "policy_version", "language", "entity_types",
"minimum_recognizer_score", "canaries_sha256", "nlp_engine_name",
"nlp_model_name", "row_id_column", "text_columns",
}
def sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def nonempty_trust_value(path: Path, name: str) -> str:
value = path.read_text(encoding="utf-8").strip()
if not value:
raise RuntimeError(f"trusted {name} is empty")
return value
def verified_policy_bytes() -> tuple[bytes, str]:
raw = POLICY_PATH.read_bytes()
bundle = POLICY_BUNDLE_PATH.read_bytes()
if not raw or not bundle:
raise RuntimeError("PII policy or signature bundle is empty")
expected_digest = sha256(raw)
identity = nonempty_trust_value(TRUST_IDENTITY, "policy identity")
issuer = nonempty_trust_value(TRUST_ISSUER, "policy issuer")
with tempfile.TemporaryDirectory(prefix="aidefend-pii-policy-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
policy_snapshot = root / "policy.json"
bundle_snapshot = root / "policy.bundle"
policy_snapshot.write_bytes(raw)
bundle_snapshot.write_bytes(bundle)
os.chmod(policy_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run(
[
"cosign", "verify-blob",
"--bundle", str(bundle_snapshot),
"--certificate-identity", identity,
"--certificate-oidc-issuer", issuer,
str(policy_snapshot),
],
check=True,
capture_output=True,
text=True,
timeout=VERIFY_TIMEOUT_SECONDS,
)
verified = policy_snapshot.read_bytes()
if sha256(verified) != expected_digest:
raise RuntimeError("verified PII policy snapshot changed")
return raw, expected_digest
def load_policy() -> tuple[dict, str]:
raw, digest = verified_policy_bytes()
policy = json.loads(raw)
if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
raise ValueError("PII policy schema differs")
if policy["schema_version"] != "aidefend.pii-policy.v1":
raise ValueError("PII policy schema version is unsupported")
for field in ("policy_version", "language", "nlp_engine_name", "nlp_model_name", "row_id_column"):
if not isinstance(policy[field], str) or not policy[field]:
raise ValueError(f"{field} must be a non-empty string")
for field in ("entity_types", "text_columns"):
values = policy[field]
if (
not isinstance(values, list) or not values
or any(not isinstance(item, str) or not item for item in values)
or len(values) != len(set(values))
):
raise ValueError(f"{field} must be a unique non-empty string list")
if policy["row_id_column"] in policy["text_columns"]:
raise ValueError("row_id_column must not be a PII text column")
threshold = policy["minimum_recognizer_score"]
if (
isinstance(threshold, bool) or not isinstance(threshold, (int, float))
or not math.isfinite(float(threshold))
or not 0.0 <= float(threshold) <= 1.0
):
raise ValueError("minimum_recognizer_score must be in [0, 1]")
digest_value = policy["canaries_sha256"]
if not isinstance(digest_value, str) or len(digest_value) != 64:
raise ValueError("canaries_sha256 must be a SHA-256 value")
int(digest_value, 16)
return policy, digest
policy, policy_sha256 = load_policy()
canary_bytes = CANARIES.read_bytes()
if sha256(canary_bytes) != policy["canaries_sha256"]:
raise PermissionError("canary population differs from signed policy")
provider = NlpEngineProvider(nlp_configuration={
"nlp_engine_name": policy["nlp_engine_name"],
"models": [{
"lang_code": policy["language"],
"model_name": policy["nlp_model_name"],
}],
})
analyzer = AnalyzerEngine(
nlp_engine=provider.create_engine(),
supported_languages=[policy["language"]],
)
anonymizer = AnonymizerEngine()
supported = set(analyzer.get_supported_entities(language=policy["language"]))
missing_recognizers = sorted(set(policy["entity_types"]) - supported)
if missing_recognizers:
raise RuntimeError(f"required recognizers are unavailable: {missing_recognizers}")
def analyze(text: str):
return analyzer.analyze(
text=text,
language=policy["language"],
entities=policy["entity_types"],
score_threshold=float(policy["minimum_recognizer_score"]),
)
def anonymize_text(text: str) -> tuple[str, list]:
findings = analyze(text)
sanitized = anonymizer.anonymize(text=text, analyzer_results=findings).text
if analyze(sanitized):
raise PermissionError("recognized PII remained after anonymization")
return sanitized, findings
def verify_canaries() -> None:
fixtures = json.loads(canary_bytes)
if not isinstance(fixtures, list) or not fixtures:
raise RuntimeError("PII canary population is missing")
covered = set()
for fixture in fixtures:
if not isinstance(fixture, dict) or set(fixture) != {"entity_type", "text", "sensitive_literal"}:
raise RuntimeError("PII canary schema differs")
entity_type = fixture["entity_type"]
text = fixture["text"]
literal = fixture["sensitive_literal"]
if not all(isinstance(value, str) and value for value in (entity_type, text, literal)):
raise RuntimeError("PII canary values must be non-empty strings")
findings = analyze(text)
if entity_type not in {item.entity_type for item in findings}:
raise PermissionError(f"canary was not detected: {entity_type}")
sanitized = anonymizer.anonymize(text=text, analyzer_results=findings).text
if literal in sanitized or analyze(sanitized):
raise PermissionError(f"canary was not removed: {entity_type}")
covered.add(entity_type)
if covered != set(policy["entity_types"]):
raise RuntimeError("canaries do not cover every required entity type")
verify_canaries()
source_bytes = SOURCE.read_bytes()
df = pd.read_csv(io.BytesIO(source_bytes), dtype="string", keep_default_na=False)
required_columns = {policy["row_id_column"], *policy["text_columns"]}
if df.empty or not required_columns.issubset(df.columns):
raise RuntimeError("source population or policy-selected columns are missing")
row_ids = df[policy["row_id_column"]]
if row_ids.eq("").any() or row_ids.duplicated().any():
raise RuntimeError("row identifier population is empty or duplicated")
result = pd.DataFrame({policy["row_id_column"]: row_ids.copy()})
entity_counts: Counter[str] = Counter()
for column in policy["text_columns"]:
sanitized_values = []
for text in df[column].astype(str):
sanitized, findings = anonymize_text(text)
sanitized_values.append(sanitized)
entity_counts.update(item.entity_type for item in findings)
result[column] = sanitized_values
if len(result) != len(df) or result[policy["row_id_column"]].tolist() != row_ids.tolist():
raise RuntimeError("anonymized population differs from source population")
output_bytes = result.to_csv(index=False).encode("utf-8")
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
temporary_output = OUTPUT.with_name(f".{OUTPUT.name}.{os.getpid()}.tmp")
with temporary_output.open("xb") as handle:
handle.write(output_bytes)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary_output, OUTPUT)
report = {
"schema_version": "aidefend.pii-anonymization.v2",
"policy_version": policy["policy_version"],
"policy_sha256": policy_sha256,
"language": policy["language"],
"nlp_engine_name": policy["nlp_engine_name"],
"nlp_model_name": policy["nlp_model_name"],
"presidio_analyzer_version": version("presidio-analyzer"),
"presidio_anonymizer_version": version("presidio-anonymizer"),
"entity_types": sorted(policy["entity_types"]),
"text_columns": sorted(policy["text_columns"]),
"source_rows": len(df),
"output_rows": len(result),
"detected_entities": dict(sorted(entity_counts.items())),
"source_sha256": sha256(source_bytes),
"output_sha256": sha256(output_bytes),
"canaries_sha256": sha256(canary_bytes),
}
print(json.dumps(report, sort_keys=True))
Verify the deployed detector path
Keep one reviewed canary for every signed policy entity type and supported language, including Unicode and formatting variants expected in the real corpus. Run the canaries and full row reconciliation with every recognizer, NLP-model, threshold, column, or language-policy change. Re-scan produced text and block snapshot admission on an invalid policy signature, policy/canary digest mismatch, missing recognizer, missed canary, residual recognized entity, unsupported language or parser, dependency failure, duplicate or missing row, or output mismatch. A clean re-scan is not proof that unknown PII is absent: sample the output under the organization's privacy review process and add signed recognizer configuration and canaries for domain-specific identifiers without retaining raw sensitive text in logs.
Action: publish only the policy-selected anonymized columns, non-personal row key, and population/digest report to the admitted training or evaluation namespace; keep the raw source and canary literals outside the identities running those jobs.
Before you begin
Apply AID-H-002.001-G004 when preparing a training, fine-tuning, alignment, or evaluation snapshot that may contain PII or other policy-prohibited sensitive data, including internally collected records. The responsible data-pipeline operator sanitizes the snapshot before a training or evaluation job can read it; using an existing model does not remove this requirement.
Open-source tools
- GX Core (Great Expectations)
- TensorFlow Data Validation (TFDV)
- ydata-profiling
- Microsoft Presidio
- cleanlab
- NVIDIA NeMo Safe Synthesizer (Apache-2.0)
- NVIDIA NeMo Curator
Source-available / open-weight tools
- Alibi Detect for outlier detection (Business Source License 1.1; source-available)
Commercial / hosted tools
- Databricks
- Alation
- Collibra
- Tonic Structural
Threat mappings
MITRE ATLAS
- AML.T0115 Publish Poisoned AI Artifacts
- AML.T0020 Training Data Poisoning
- AML.T0059 Erode Dataset Integrity
- AML.T0115.000 Publish Poisoned AI Artifacts: Datasets
- AML.T0057 LLM Data Leakage (removing PII from training data prevents leakage)
MAESTRO
- Data Poisoning (L2)
- Data Poisoning (Training Phase) (L1)
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure
- LLM05:2026 Data and Model Poisoning
OWASP ML Top 10 2023
- ML02:2023 Data Poisoning Attack
- ML04:2023 Membership Inference Attack (removing sensitive records reduces inference risk)
OWASP Top 10 for Agentic Applications 2026
- ASI04:2026 Agentic Supply Chain Vulnerabilities (sanitizing training data from third-party sources)
NIST Adversarial Machine Learning 2025
- NISTAML.013 Data Poisoning
- NISTAML.024 Targeted Poisoning
- NISTAML.023 Backdoor Poisoning
- NISTAML.012 Clean-label Poisoning
- NISTAML.037 Training Data Attacks
- NISTAML.033 Membership Inference (removing sensitive data reduces inference risk)
Cisco Integrated AI Security and Safety Framework
- AISubtech-10.1.3 Sensitive Data Reconstruction (Removes recognized sensitive training literals before memorization can expose them)
- AISubtech-19.2.1 Convergence Payload Injection (Training-side fusion anomaly admission with model-specific partial-payload tests)
- AISubtech-7.3.1 Corrupted Third-Party Data
- AISubtech-8.2.1 Training Data Exposure (Removes recognized sensitive records before training; unknown PII remains outside this guarantee)
- AISubtech-8.2.2 LLM Data Leakage (Removes recognized sensitive records before training; unknown PII remains outside this guarantee)
- AISubtech-8.3.3 Personally Identifiable Information Exposure (Removes recognized sensitive records before training; unknown PII remains outside this guarantee)
- AITech-19.2 Fusion Payload Split (Training-side fusion anomaly admission with model-specific partial-payload tests)
- AITech-6.1 Training Data Poisoning
- AITech-7.3 Data Source Abuse and Manipulation
- AITech-8.2 Data Exfiltration / Exposure (Removes recognized sensitive records before training; unknown PII remains outside this guarantee)
- AITech-8.3 Information Disclosure (Removes recognized sensitive records before training; unknown PII remains outside this guarantee)
Google Secure AI Framework 2.0 - Risks
- DP: Data Poisoning
- SDD: Sensitive Data Disclosure (PII removal from training data prevents disclosure)
- EDH: Excessive Data Handling (sanitization enforces data handling policies)
Databricks AI Security Framework 3.0
- Raw Data 1.3: Poor data quality
- Raw Data 1.7: Lack of data trustworthiness
- Raw Data 1.8: Legality of data (sanitization filters legally restricted data)
- Raw Data 1.11: Compromised 3rd-party datasets
- Datasets 3.1: Data poisoning