AID-M-002.003: External Dataset Content-Risk Characterization

Version 1.20261015 · Actionable sub-technique

Characterize the security- and privacy-relevant contents of each externally obtained dataset before it is used for training, tuning, evaluation, retrieval, or grounding, including measured sensitive-data presence, secrets, label and feature distributions, malformed records, and policy-defined prohibited content.

Pillars: data, model

Lifecycle: scoping, building

Scope boundary

Owns content-risk characterization of an external dataset's actual records, distributions, labels, anomalies, sensitive content, and poisoning indicators before downstream admission decisions.

AID-H-003.003: AID-M-002.003 characterizes what risks are present in the dataset contents; AID-H-003.003 decides whether the external dataset and its supplier provenance satisfy the admission policy. A low-risk content scan does not approve an untrusted supplier, and trusted provenance does not prove that dataset contents are safe.

Implementation guidance

AID-M-002.003-G001

Define a versioned content-risk policy for the dataset class and bind it to the complete immutable input manifest.
1. Freeze the dataset-class content policy

Consume the complete manifest already authenticated by AID-H-003.003, then version the supported formats, required fields and types, content fields, duplicate keys, distribution baselines, PII entities, detector versions, and finding limits used for this dataset class. Manifest signatures, publisher identity, rights, and lineage stay with AID-H-003.003; a trusted source is not evidence that its content is safe.

{"schema_version":"aidefend.content-risk-policy.v1","policy_id":"support-corpus-v3","presidio":{"package_version":"2.2.355","language":"en","entities":["EMAIL_ADDRESS","PHONE_NUMBER","US_SSN","CREDIT_CARD"],"min_score":0.70},"trufflehog":{"version_output":"trufflehog 3.90.8"},"formats":{".csv":{"kind":"csv","required_fields":["text","label"],"field_types":{"text":"string","label":"string"},"content_fields":["text"],"duplicate_key_fields":["text"],"distributions":{"label":{"expected":{"safe":0.8,"unsafe":0.2},"max_total_variation":0.10}}},".jsonl":{"kind":"jsonl","required_fields":["text","label"],"field_types":{"text":"string","label":"string"},"content_fields":["text"],"duplicate_key_fields":["text"],"distributions":{"label":{"expected":{"safe":0.8,"unsafe":0.2},"max_total_variation":0.10}}}},"limits":{"pii_findings":0,"verified_secrets":0,"schema_violations":0,"malformed_records":0,"duplicate_records":0,"distribution_breaches":0}}

The values and tool versions above are a reviewed deployment example, not universal thresholds. Generate and pin them through the version-locked build and organizational policy for the actual dataset class.

AID-M-002.003-G002

Stream every CSV and JSONL manifest member through the content-risk profiler and block promotion on findings or incomplete coverage.
2. Profile every manifest member

Save the following as tools/profile_dataset.py in the locked scanner image. It normalizes the manifest, inventories every file, rejects missing, extra, changed, linked, escaped, unreadable, or unsupported inputs, streams CSV and JSONL records, checks schema and malformed records, detects cross-file duplicates, measures policy-defined distributions, runs Presidio and TruffleHog --only-verified, and reports tool/version failures as incomplete coverage.

from __future__ import annotations
import argparse,csv,hashlib,json,math,os,sqlite3,subprocess,tempfile
from collections import Counter,defaultdict
from importlib.metadata import PackageNotFoundError,version
from pathlib import Path
try:
 from presidio_analyzer import AnalyzerEngine
except ImportError:
 AnalyzerEngine=None
try:
 COMMAND_TIMEOUT_SECONDS=float(os.environ["M002003_SCANNER_COMMAND_TIMEOUT_SECONDS"])
except (KeyError,ValueError) as error:
 raise RuntimeError("signed runtime-profile scanner timeout is unavailable") from error
if not math.isfinite(COMMAND_TIMEOUT_SECONDS) or COMMAND_TIMEOUT_SECONDS<=0:
 raise RuntimeError("signed runtime-profile scanner timeout is invalid")
T={"string":str,"integer":int,"number":(int,float),"boolean":bool}
def canon(v): return json.dumps(v,sort_keys=True,separators=(",",":"),ensure_ascii=False).encode("utf-8")
def sha(p):
 h=hashlib.sha256()
 with p.open("rb") as f:
  for b in iter(lambda:f.read(1048576),b""): h.update(b)
 return h.hexdigest()
def inventory(root):
 files=[]; unsafe=[]
 for p in sorted(root.rglob("*")):
  rel=p.relative_to(root).as_posix()
  if p.is_symlink(): unsafe.append(f"symlink:{rel}")
  elif p.is_file(): files.append({"path":rel,"size_bytes":p.stat().st_size,"sha256":sha(p)})
 return files,unsafe
def rows(path,kind):
 if kind=="csv":
  try:
   with path.open("r",encoding="utf-8",newline="") as f:
    r=csv.reader(f,strict=True)
    try: head=next(r)
    except StopIteration: yield 1,None,"empty_file",False; return
    if not head or len(head)!=len(set(head)) or any(not x for x in head): yield 1,None,"invalid_header",True; return
    try:
     for n,values in enumerate(r,2):
      yield (n,None,"column_count",False) if len(values)!=len(head) else (n,dict(zip(head,values)),None,False)
    except csv.Error as e: yield r.line_num,None,f"csv_error:{e}",True
  except (OSError,UnicodeError) as e: yield 0,None,f"read_error:{e}",True
 elif kind=="jsonl":
  try:
   with path.open("r",encoding="utf-8") as f:
    for n,raw in enumerate(f,1):
     try: value=json.loads(raw)
     except json.JSONDecodeError as e: yield n,None,f"json_error:{e.msg}",False; continue
     yield (n,value,None,False) if isinstance(value,dict) else (n,None,"record_not_object",False)
  except (OSError,UnicodeError) as e: yield 0,None,f"read_error:{e}",True
def type_ok(value,name):
 typ=T.get(name)
 if typ is None: raise ValueError(f"unsupported policy type:{name}")
 return not(name in {"integer","number"} and isinstance(value,bool)) and isinstance(value,typ)
def th_version():
 p=subprocess.run(["trufflehog","--version"],capture_output=True,text=True,check=False,timeout=COMMAND_TIMEOUT_SECONDS)
 if p.returncode: raise RuntimeError("TruffleHog version query failed")
 return (p.stdout or p.stderr).strip()
def secrets(path):
 p=subprocess.run(["trufflehog","filesystem","--json","--no-update","--only-verified",str(path)],capture_output=True,text=True,check=False,timeout=COMMAND_TIMEOUT_SECONDS)
 if p.returncode: raise RuntimeError((p.stderr or p.stdout).strip() or "TruffleHog failed")
 out=Counter()
 for raw in p.stdout.splitlines():
  if raw.strip():
   item=json.loads(raw)
   if item.get("Verified") is True: out[str(item.get("DetectorName","unknown"))]+=1
 return out
def normalized_manifest(manifest,errors):
 raw=manifest.get("files")
 if not isinstance(raw,list) or not raw: errors.append("manifest_has_no_files"); return []
 out=[]; paths=set()
 for i,item in enumerate(raw):
  try: row={"path":str(item["path"]),"size_bytes":int(item["size_bytes"]),"sha256":str(item["sha256"]).lower()}
  except (KeyError,TypeError,ValueError): errors.append(f"invalid_manifest_entry:{i}"); continue
  if row["path"] in paths: errors.append(f"duplicate_manifest_path:{row['path']}")
  paths.add(row["path"]); out.append(row)
 return sorted(out,key=lambda x:x["path"])
def build(root,manifest_path,policy_path):
 root=root.resolve(); manifest=json.loads(manifest_path.read_text(encoding="utf-8")); policy=json.loads(policy_path.read_text(encoding="utf-8"))
 errors=[]; coverage=[]; expected=normalized_manifest(manifest,errors); actual,unsafe=inventory(root); errors+=unsafe
 if expected!=actual: errors.append("manifest_inventory_or_digest_mismatch")
 for item in expected:
  rel=item["path"]
  try:
   p=(root/rel).resolve(strict=True)
   if not rel or not p.is_relative_to(root): errors.append(f"unsafe_manifest_path:{rel}")
  except OSError: errors.append(f"unreadable_manifest_path:{rel}")
 try: pv=version("presidio-analyzer")
 except PackageNotFoundError: pv="NOT_INSTALLED"
 analyzer=None
 if AnalyzerEngine is None or pv!=policy["presidio"]["package_version"]: errors.append("presidio_version_mismatch")
 else:
  try: analyzer=AnalyzerEngine()
  except Exception as e: errors.append(f"presidio_initialization_error:{type(e).__name__}")
 th_ok=False
 try: th_ok=th_version()==policy["trufflehog"]["version_output"]
 except (OSError,RuntimeError,subprocess.SubprocessError): pass
 if not th_ok: errors.append("trufflehog_version_mismatch")
 totals=Counter(); pii=Counter(); secret_types=Counter(); dists=defaultdict(Counter); dist_rules={}
 with tempfile.TemporaryDirectory(prefix="aidefend-content-risk-") as td:
  db=sqlite3.connect(str(Path(td)/"duplicates.sqlite3")); db.execute("CREATE TABLE seen(digest TEXT PRIMARY KEY)")
  manifest_bad=any(x.startswith(("manifest_","invalid_manifest","duplicate_manifest","symlink","unsafe_manifest","unreadable_manifest")) for x in errors)
  if not manifest_bad:
   for item in expected:
    rel=item["path"]; path=root/rel; spec=policy.get("formats",{}).get(path.suffix.lower())
    if spec is None: coverage.append({"path":rel,"state":"UNSUPPORTED","reason":path.suffix.lower() or "no_suffix"}); errors.append(f"unsupported_media:{rel}"); continue
    if spec.get("kind") not in {"csv","jsonl"}: coverage.append({"path":rel,"state":"UNSUPPORTED","reason":"invalid_format_adapter"}); errors.append(f"unsupported_adapter:{rel}"); continue
    counts=Counter(); malformed=Counter(); fatal=[]
    for field,rule in spec.get("distributions",{}).items():
     old=dist_rules.setdefault(field,rule)
     if old!=rule: fatal.append(f"conflicting_distribution_policy:{field}")
    for _n,record,issue,is_fatal in rows(path,spec["kind"]):
     if issue:
      counts["malformed_records"]+=1; malformed[issue]+=1
      if is_fatal: fatal.append(issue)
      continue
     counts["records"]+=1
     bad=any(record.get(f) in (None,"") for f in spec.get("required_fields",[]))
     for f,want in spec.get("field_types",{}).items():
      if f in record and not type_ok(record[f],want): bad=True
     counts["schema_violations"]+=int(bad)
     key=[record.get(f) for f in spec.get("duplicate_key_fields",[])]
     if key:
      d=hashlib.sha256(canon(key)).hexdigest()
      if db.execute("INSERT OR IGNORE INTO seen VALUES(?)",(d,)).rowcount==0: counts["duplicate_records"]+=1
     for f in spec.get("distributions",{}): dists[f][str(record.get(f,"<MISSING>"))]+=1
     if analyzer is not None:
      for f in spec.get("content_fields",[]):
       value=record.get(f)
       if isinstance(value,str):
        for hit in analyzer.analyze(text=value,language=policy["presidio"]["language"],entities=policy["presidio"]["entities"],score_threshold=float(policy["presidio"]["min_score"])):
         pii[hit.entity_type]+=1; counts["pii_findings"]+=1
    if analyzer is None: fatal.append("pii_scan_unavailable")
    if th_ok:
     try: found=secrets(path); secret_types.update(found); counts["verified_secrets"]+=sum(found.values())
     except (OSError,RuntimeError,subprocess.SubprocessError,json.JSONDecodeError) as e: fatal.append(f"secret_scan_error:{type(e).__name__}")
    else: fatal.append("secret_scan_unavailable")
    totals.update(counts); coverage.append({"path":rel,"state":"INCOMPLETE" if fatal else "MEASURED","counts":dict(sorted(counts.items())),"malformed_by_reason":dict(sorted(malformed.items())),"fatal_errors":sorted(set(fatal))})
    errors += [f"scan_incomplete:{rel}:{reason}" for reason in fatal]
  db.close()
 dist_report={}; dist_breaches=0
 for field,counts in sorted(dists.items()):
  rule=dist_rules[field]; total=sum(counts.values()); observed={k:v/total for k,v in sorted(counts.items())} if total else {}; expected_dist={k:float(v) for k,v in rule.get("expected",{}).items()}
  if abs(sum(expected_dist.values())-1)>1e-9: errors.append(f"invalid_expected_distribution:{field}")
  keys=set(observed)|set(expected_dist); tvd=.5*sum(abs(observed.get(k,0)-expected_dist.get(k,0)) for k in keys); breached=total==0 or tvd>float(rule["max_total_variation"]); dist_breaches+=int(breached)
  dist_report[field]={"counts":dict(sorted(counts.items())),"observed":observed,"total_variation":round(tvd,8),"breached":breached}
 totals["pii_findings"]=sum(pii.values()); totals["verified_secrets"]=sum(secret_types.values()); totals["distribution_breaches"]=dist_breaches
 violations=[{"metric":m,"observed":totals.get(m,0),"maximum":int(mx)} for m,mx in sorted(policy["limits"].items()) if totals.get(m,0)>int(mx)]
 complete=bool(expected) and not errors and len(coverage)==len(expected) and all(x["state"]=="MEASURED" for x in coverage)
 return {"schema_version":"aidefend.content-risk-report.v1","dataset_id":manifest.get("dataset_id"),"manifest_sha256":sha(manifest_path),"policy_sha256":sha(policy_path),"complete":complete,"promotion_allowed":complete and not violations,"blocking_reasons":sorted(set(errors)),"violations":violations,"totals":dict(sorted(totals.items())),"pii_by_entity":dict(sorted(pii.items())),"verified_secrets_by_detector":dict(sorted(secret_types.items())),"distributions":dist_report,"coverage":coverage}
def main():
 p=argparse.ArgumentParser(); p.add_argument("--root",type=Path,required=True); p.add_argument("--manifest",type=Path,required=True); p.add_argument("--policy",type=Path,required=True); mode=p.add_mutually_exclusive_group(required=True); mode.add_argument("--out",type=Path); mode.add_argument("--verify",type=Path); a=p.parse_args()
 before=inventory(a.root.resolve()); actual=build(a.root,a.manifest,a.policy); after=inventory(a.root.resolve())
 if before!=after: raise SystemExit("source dataset changed during read-only scan")
 if a.verify:
  expected=json.loads(a.verify.read_text(encoding="utf-8"))
  if canon(actual)!=canon(expected): raise SystemExit("replay mismatch")
  print(hashlib.sha256(canon(actual)).hexdigest())
 else: a.out.write_text(json.dumps(actual,indent=2,sort_keys=True)+"\n",encoding="utf-8")
 raise SystemExit(0 if actual["promotion_allowed"] else 2)
if __name__=="__main__": main()

Pin the actual Presidio package, TruffleHog binary, and language model in the container lock. The report retains only counts, entity types, detector names, coverage states, and policy/digest bindings; never store raw PII, secret values, or matching excerpts.

python tools/profile_dataset.py --root quarantine/dataset --manifest evidence/dataset-manifest.json --policy policies/content-risk.json --out evidence/content-risk-report.json

Promotion is allowed only when the complete manifest population is measured and no policy limit is exceeded. Unsupported media, tool/version mismatch, scanner failure, unsafe path, population drift, or unreadable input remains incomplete and blocks promotion.

AID-M-002.003-G003

Replay the complete content-risk profile from read-only inputs before promotion.
3. Replay the profile at the promotion gate

Mount the dataset root, manifest, policy, and recorded report read-only in an independent promotion runner. First generate the signed candidate report, then rerun the same locked profiler with --verify; the verifier re-inventories the inputs and requires canonical byte-equivalence with the recorded deterministic report.

python tools/profile_dataset.py --root quarantine/dataset --manifest evidence/dataset-manifest.json --policy policies/content-risk.json --out evidence/content-risk-report.json
python tools/profile_dataset.py --root quarantine/dataset --manifest evidence/dataset-manifest.json --policy policies/content-risk.json --verify evidence/content-risk-report.json

Exercise negative fixtures for an extra file, missing member, digest change, symlink, malformed record, unsupported suffix, Presidio or TruffleHog version drift, verified secret, PII finding, duplicate, distribution breach, and replay mismatch. Any nonzero scanner result, incomplete coverage, changed read-only inventory, or replay mismatch prevents PASS.

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-M-002.003: External Dataset Content-Risk Characterization | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service