AID-H-002.006: Dataset Split Integrity, Partition Governance & Leakage Prevention
Ensure the integrity and security of dataset partitioning into training, validation, and test splits. Adversarial partitioning, post-hoc split manipulation, and train-test leakage can make a compromised model look better than it is and can invalidate downstream evaluation. This sub-technique enforces deterministic and auditable splitting, split immutability, exact and near-duplicate leakage detection across partitions, representative stratification checks, and access controls so training and evaluation code cannot casually read each other's splits.
Pillars: data
Lifecycle: building
Scope boundary
Owns deterministic and auditable train, validation, and test partitioning, split immutability, duplicate and near-duplicate leakage detection, representative stratification, temporal integrity, and split-level access controls.
AID-H-007.004: AID-H-002.006 creates and protects dataset partitions against leakage and post-hoc manipulation; AID-H-007.004 governs the integrity, sufficiency, independence, and promotion use of evaluation evidence. A clean split is necessary evaluation input but does not establish evaluator independence, case sufficiency, or a valid promotion decision.
Implementation guidance
- AID-H-002.006-G001: Use deterministic split generation with recorded seed, split metadata, and split artifact hashing so every split can be reproduced and verified later.
- AID-H-002.006-G002: Check for exact duplicates and near-duplicates across train, validation, and test before training begins, and block if leakage exceeds defined thresholds.
- AID-H-002.006-G003: Apply split-level access control so training jobs can only read the training partition and evaluation jobs can only read the validation or test partitions they need.
- AID-H-002.006-G004: Verify that the split remains representative by checking class balance, temporal integrity, and important feature distributions across partitions.
AID-H-002.006-G001
Use deterministic split generation with recorded seed, split metadata, and split artifact hashing so every split can be reproduced and verified later.
Concept:
A split should be a reproducible artifact, not an informal notebook operation. Record the seed, algorithm, and resulting sample IDs.
Example deterministic split
from __future__ import annotations
import hashlib
import json
from sklearn.model_selection import train_test_split
def create_stratified_split(*, rows, labels, test_fraction: float, split_seed: int) -> dict:
if not 0.0 < test_fraction < 1.0:
raise ValueError("signed test fraction must be inside (0, 1)")
X_train, X_test, y_train, y_test = train_test_split(
rows,
labels,
test_size=test_fraction,
random_state=split_seed,
stratify=labels,
)
split_record = {
"algorithm": "train_test_split",
"random_state": split_seed,
"train_ids": [row["id"] for row in X_train],
"test_ids": [row["id"] for row in X_test],
}
payload = json.dumps(split_record, sort_keys=True, separators=(",", ":")).encode()
split_record["sha256"] = hashlib.sha256(payload).hexdigest()
return {
"X_train": X_train,
"X_test": X_test,
"y_train": y_train,
"y_test": y_test,
"split_record": split_record,
}
Operational notes: Store the split record in version control or artifact storage. If the split changes, require a reviewed regeneration step rather than silently replacing it.
Keep the action safely bounded
block training or evaluation on overlap, missing members, unauthorized partition access, unreproducible assignment, or policy-exceeding representativeness error.
AID-H-002.006-G002
Check for exact duplicates and near-duplicates across train, validation, and test before training begins, and block if leakage exceeds defined thresholds.
Concept
Make split integrity a signed, blocking pre-training gate over the complete train, validation, and test manifests. Exact normalized duplicates are checked within every split and across all three split pairs. Near-duplicate candidates are obtained with a deterministic shingle inverted index and then scored with exact set Jaccard arithmetic; because the signed threshold is positive, every pair at or above the threshold shares a shingle, so this candidate step is lossless for the declared detector. This is evidence about the signed normalization, shingle algorithm, threshold, and population—not proof that no semantic duplicate exists under every possible detector.
Implement the deterministic checker
# File: dataset_checks/split_integrity.py
from __future__ import annotations
import argparse
import csv
import hashlib
import io
import json
import math
import os
import re
import stat
import subprocess
import sys
import tempfile
import unicodedata
from collections import defaultdict
from decimal import Decimal, InvalidOperation
from pathlib import Path
SPLITS = ("train", "validation", "test")
PAIRS = (("train", "validation"), ("train", "test"), ("validation", "test"))
POLICY_SCHEMA = "aidefend.h002006.split-policy.v2"
RESULT_SCHEMA = "aidefend.h002006.split-result.v2"
NORMALIZATION = "unicode_nfc_casefold_whitespace_v1"
SHINGLE_ALGORITHM = "word_5gram_set_jaccard_v1"
POLICY_KEY = Path("/opt/aidefend/trust/h002006-split-policy.pub")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
HARD_LIMITS = {
"max_file_bytes": 10_000_000_000,
"max_rows_per_split": 10_000_000,
"max_candidate_pairs_total": 100_000_000,
"max_findings": 1_000_000,
"max_result_bytes": 64_000_000,
}
class GateError(RuntimeError):
pass
def strict_json(raw: bytes, label: str) -> object:
def pairs(items):
value = {}
for key, item in items:
if key in value:
raise GateError(f"{label}: duplicate key {key!r}")
value[key] = item
return value
def reject(value):
raise GateError(f"{label}: non-finite number {value}")
try:
return json.loads(
raw.decode("utf-8", "strict"),
object_pairs_hook=pairs,
parse_constant=reject,
)
except (UnicodeError, json.JSONDecodeError) as exc:
raise GateError(f"{label}: invalid UTF-8 JSON") from exc
def canonical(value: object) -> bytes:
try:
return json.dumps(
value, ensure_ascii=False, allow_nan=False,
sort_keys=True, separators=(",", ":"),
).encode("utf-8")
except (TypeError, ValueError) as exc:
raise GateError("non-canonical JSON value") from exc
def exact_object(value: object, fields: set[str], label: str) -> dict:
if not isinstance(value, dict) or set(value) != fields:
raise GateError(f"{label}: schema mismatch")
return value
def bounded_int(value: object, label: str, maximum: int) -> int:
if type(value) is not int or not 0 < value <= maximum:
raise GateError(f"{label}: expected integer in [1,{maximum}]")
return value
def stable_read(path: Path, maximum: int, protected_key: bool = False) -> bytes:
nofollow = getattr(os, "O_NOFOLLOW", None)
if nofollow is None:
raise GateError("verification runner must support O_NOFOLLOW")
fd = os.open(path, os.O_RDONLY | nofollow | getattr(os, "O_CLOEXEC", 0))
try:
before = os.fstat(fd)
if not stat.S_ISREG(before.st_mode) or before.st_size > maximum:
raise GateError(f"{path}: invalid file type or size")
if protected_key and (
before.st_uid != 0 or stat.S_IMODE(before.st_mode) & 0o022
):
raise GateError(f"{path}: trust key is not root-owned/read-only")
parts, total = [], 0
while True:
part = os.read(fd, min(1_048_576, maximum + 1 - total))
if not part:
break
parts.append(part)
total += len(part)
if total > maximum:
raise GateError(f"{path}: byte limit exceeded")
after = os.fstat(fd)
identity = lambda item: (item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns)
if identity(before) != identity(after) or total != after.st_size:
raise GateError(f"{path}: changed during read")
return b"".join(parts)
finally:
os.close(fd)
def write_exclusive(path: Path, payload: bytes, maximum: int) -> str:
if len(payload) > maximum:
raise GateError(f"{path}: output exceeds signed byte limit")
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0)
flags |= getattr(os, "O_NOFOLLOW", 0)
fd = os.open(path, flags, 0o600)
try:
view = memoryview(payload)
while view:
count = os.write(fd, view)
if count <= 0:
raise GateError(f"{path}: short write")
view = view[count:]
os.fsync(fd)
finally:
os.close(fd)
if stable_read(path, maximum) != payload:
raise GateError(f"{path}: readback mismatch")
return hashlib.sha256(payload).hexdigest()
def validate_policy(value: object) -> tuple[dict, int, int]:
policy = exact_object(value, {
"schema_version", "policy_version", "normalization", "shingle_algorithm",
"near_duplicate_threshold", "limits", "splits",
}, "policy")
if policy["schema_version"] != POLICY_SCHEMA:
raise GateError("policy: unsupported schema")
if not isinstance(policy["policy_version"], str) or not policy["policy_version"].strip():
raise GateError("policy: missing version")
if policy["normalization"] != NORMALIZATION:
raise GateError("policy: unsupported normalization")
if policy["shingle_algorithm"] != SHINGLE_ALGORITHM:
raise GateError("policy: unsupported shingle algorithm")
if not isinstance(policy["near_duplicate_threshold"], str):
raise GateError("policy: threshold must be a decimal string")
try:
threshold = Decimal(policy["near_duplicate_threshold"])
except InvalidOperation as exc:
raise GateError("policy: invalid threshold") from exc
if not threshold.is_finite() or not Decimal("0") < threshold <= Decimal("1"):
raise GateError("policy: threshold must be in (0,1]")
numerator, denominator = threshold.as_integer_ratio()
limits = exact_object(policy["limits"], set(HARD_LIMITS), "policy.limits")
for name, maximum in HARD_LIMITS.items():
limits[name] = bounded_int(limits[name], f"policy.limits.{name}", maximum)
specs = exact_object(policy["splits"], set(SPLITS), "policy.splits")
for split in SPLITS:
spec = exact_object(specs[split], {"path", "sha256", "row_count"}, split)
if spec["path"] != f"artifacts/{split}.csv":
raise GateError(f"{split}: non-canonical path")
if not isinstance(spec["sha256"], str) or not SHA256.fullmatch(spec["sha256"]):
raise GateError(f"{split}: invalid sha256")
bounded_int(spec["row_count"], f"{split}.row_count", limits["max_rows_per_split"])
return policy, numerator, denominator
def verify_policy(policy_path: Path, bundle_path: Path, timeout: float):
if not math.isfinite(timeout) or timeout <= 0:
raise GateError("cosign timeout must be finite and positive")
policy_raw = stable_read(policy_path, 1_048_576)
bundle_raw = stable_read(bundle_path, 8_000_000)
key_raw = stable_read(POLICY_KEY, 65_536, protected_key=True)
with tempfile.TemporaryDirectory(prefix="h002006-") as directory:
copies = []
for name, raw in (
("policy.json", policy_raw),
("bundle.json", bundle_raw),
("policy.pub", key_raw),
):
path = Path(directory) / name
path.write_bytes(raw)
path.chmod(0o400)
copies.append(path)
try:
completed = subprocess.run(
["cosign", "verify-blob", "--key", str(copies[2]),
"--bundle", str(copies[1]), str(copies[0])],
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, check=False, timeout=timeout,
)
except (OSError, subprocess.SubprocessError) as exc:
raise GateError("policy signature verification could not run") from exc
if completed.returncode != 0:
raise GateError("policy signature verification failed")
policy, numerator, denominator = validate_policy(strict_json(policy_raw, "policy"))
receipt = {
"policy_sha256": hashlib.sha256(policy_raw).hexdigest(),
"policy_bundle_sha256": hashlib.sha256(bundle_raw).hexdigest(),
"policy_key_sha256": hashlib.sha256(key_raw).hexdigest(),
}
return policy, numerator, denominator, receipt
def normalize_text(text: str) -> str:
return " ".join(unicodedata.normalize("NFC", text).casefold().split())
def make_shingles(normalized: str) -> frozenset[str]:
words = normalized.split(" ")
if len(words) < 5:
return frozenset((normalized,))
return frozenset(
" ".join(words[index:index + 5]) for index in range(len(words) - 4)
)
def parse_split(raw: bytes, split: str, maximum_rows: int) -> list[dict]:
try:
reader = csv.reader(
io.StringIO(raw.decode("utf-8", "strict"), newline=""), strict=True
)
header = next(reader)
except (UnicodeError, csv.Error, StopIteration) as exc:
raise GateError(f"{split}: invalid UTF-8 CSV") from exc
if header != ["id", "text"]:
raise GateError(f"{split}: header must be exactly id,text")
result, identifiers = [], set()
try:
for line, fields in enumerate(reader, start=2):
if len(fields) != 2:
raise GateError(f"{split}:{line}: expected two columns")
identifier, source_text = fields
if (
not identifier
or identifier != identifier.strip()
or "\x00" in identifier
or identifier in identifiers
):
raise GateError(f"{split}:{line}: invalid or duplicate id")
normalized = normalize_text(source_text)
if not normalized or "\x00" in source_text:
raise GateError(f"{split}:{line}: empty or invalid text")
identifiers.add(identifier)
result.append({
"id": identifier,
"text_sha256": hashlib.sha256(normalized.encode("utf-8")).hexdigest(),
"shingles": make_shingles(normalized),
})
if len(result) > maximum_rows:
raise GateError(f"{split}: row limit exceeded")
except csv.Error as exc:
raise GateError(f"{split}: malformed CSV") from exc
if not result:
raise GateError(f"{split}: empty population")
result.sort(key=lambda row: row["id"].encode("utf-8"))
return result
def load_splits(dataset_root: Path, policy: dict):
if not dataset_root.is_dir():
raise GateError("dataset root is not a directory")
populations, evidence = {}, {}
for split in SPLITS:
spec = policy["splits"][split]
raw = stable_read(
dataset_root / spec["path"], policy["limits"]["max_file_bytes"]
)
raw_sha256 = hashlib.sha256(raw).hexdigest()
if raw_sha256 != spec["sha256"]:
raise GateError(f"{split}: sha256 differs from signed manifest")
rows = parse_split(raw, split, policy["limits"]["max_rows_per_split"])
if len(rows) != spec["row_count"]:
raise GateError(f"{split}: row count differs from signed manifest")
population = [
{"id": row["id"], "text_sha256": row["text_sha256"]} for row in rows
]
populations[split] = rows
evidence[split] = {
"path": spec["path"],
"raw_sha256": raw_sha256,
"row_count": len(rows),
"population_sha256": hashlib.sha256(canonical(population)).hexdigest(),
}
return populations, evidence
def add_finding(findings: list[dict], value: dict, maximum: int) -> None:
findings.append(value)
if len(findings) > maximum:
raise GateError("finding capacity exceeded; no verdict produced")
def detect_findings(
populations: dict[str, list[dict]],
threshold_numerator: int,
threshold_denominator: int,
limits: dict,
):
if set(populations) != set(SPLITS):
raise GateError("all three split populations are required")
findings, coverage = [], []
candidate_count_total = 0
for split in SPLITS:
by_digest = defaultdict(list)
for row in populations[split]:
by_digest[row["text_sha256"]].append(row["id"])
for digest in sorted(by_digest):
identifiers = sorted(by_digest[digest], key=lambda value: value.encode())
if len(identifiers) > 1:
add_finding(findings, {
"kind": "within_split_exact",
"split": split,
"text_sha256": digest,
"ids": identifiers,
}, limits["max_findings"])
for left_name, right_name in PAIRS:
left_rows, right_rows = populations[left_name], populations[right_name]
left_ids = {row["id"] for row in left_rows}
right_ids = {row["id"] for row in right_rows}
for identifier in sorted(left_ids & right_ids, key=lambda value: value.encode()):
add_finding(findings, {
"kind": "cross_split_id",
"left_split": left_name,
"right_split": right_name,
"id": identifier,
}, limits["max_findings"])
left_digests, right_digests = defaultdict(list), defaultdict(list)
for row in left_rows:
left_digests[row["text_sha256"]].append(row["id"])
for row in right_rows:
right_digests[row["text_sha256"]].append(row["id"])
for digest in sorted(set(left_digests) & set(right_digests)):
add_finding(findings, {
"kind": "cross_split_exact",
"left_split": left_name,
"right_split": right_name,
"text_sha256": digest,
"left_ids": sorted(left_digests[digest]),
"right_ids": sorted(right_digests[digest]),
}, limits["max_findings"])
# Any pair with positive Jaccard similarity shares a shingle. Therefore
# this inverted index is a lossless candidate generator for the signed,
# strictly positive threshold; every candidate is then scored exactly.
index = defaultdict(list)
for position, row in enumerate(right_rows):
for shingle in row["shingles"]:
index[shingle].append(position)
candidate_count_pair = near_count = 0
for left in left_rows:
candidate_positions = set()
for shingle in left["shingles"]:
candidate_positions.update(index.get(shingle, ()))
for position in sorted(candidate_positions):
candidate_count_pair += 1
candidate_count_total += 1
if candidate_count_total > limits["max_candidate_pairs_total"]:
raise GateError(
"aggregate candidate capacity exceeded; no verdict produced"
)
right = right_rows[position]
if left["text_sha256"] == right["text_sha256"]:
continue
intersection = len(left["shingles"] & right["shingles"])
union = len(left["shingles"] | right["shingles"])
if intersection * threshold_denominator >= threshold_numerator * union:
near_count += 1
add_finding(findings, {
"kind": "cross_split_near",
"left_split": left_name,
"right_split": right_name,
"left_id": left["id"],
"right_id": right["id"],
"jaccard": f"{intersection}/{union}",
}, limits["max_findings"])
coverage.append({
"left_split": left_name,
"right_split": right_name,
"left_rows": len(left_rows),
"right_rows": len(right_rows),
"candidate_pairs_exactly_scored": candidate_count_pair,
"near_findings": near_count,
})
return findings, coverage, candidate_count_total
def inspect_split(path: Path, split: str, maximum_bytes: int, maximum_rows: int):
raw = stable_read(path, maximum_bytes)
rows = parse_split(raw, split, maximum_rows)
return hashlib.sha256(raw).hexdigest(), len(rows)
def build_policy(args: argparse.Namespace) -> int:
try:
threshold = Decimal(args.threshold)
except InvalidOperation as exc:
raise GateError("threshold must be a decimal string") from exc
if not threshold.is_finite() or not Decimal("0") < threshold <= Decimal("1"):
raise GateError("threshold must be in (0,1]")
if not args.policy_version.strip():
raise GateError("policy version is required")
limits = {
name: bounded_int(getattr(args, name), name, maximum)
for name, maximum in HARD_LIMITS.items()
}
root = args.dataset_root.resolve(strict=True)
specs = {}
for split in SPLITS:
relative = f"artifacts/{split}.csv"
digest, count = inspect_split(
root / relative, split,
limits["max_file_bytes"], limits["max_rows_per_split"],
)
specs[split] = {"path": relative, "sha256": digest, "row_count": count}
policy = {
"schema_version": POLICY_SCHEMA,
"policy_version": args.policy_version,
"normalization": NORMALIZATION,
"shingle_algorithm": SHINGLE_ALGORITHM,
"near_duplicate_threshold": format(threshold, "f"),
"limits": limits,
"splits": specs,
}
validate_policy(policy)
payload = canonical(policy) + b"\n"
print(write_exclusive(args.output, payload, 1_048_576))
return 0
def check(args: argparse.Namespace) -> int:
policy, numerator, denominator, receipt = verify_policy(
args.policy, args.bundle, args.cosign_timeout_seconds
)
populations, population_evidence = load_splits(
args.dataset_root.resolve(strict=True), policy
)
findings, pair_coverage, candidate_count_total = detect_findings(
populations, numerator, denominator, policy["limits"]
)
result = {
"schema_version": RESULT_SCHEMA,
"outcome": "FAIL" if findings else "PASS",
"policy_version": policy["policy_version"],
**receipt,
"normalization": NORMALIZATION,
"shingle_algorithm": SHINGLE_ALGORITHM,
"near_duplicate_threshold": policy["near_duplicate_threshold"],
"population": population_evidence,
"coverage": {
"within_split_exact": list(SPLITS),
"cross_split_exact_and_near": pair_coverage,
"candidate_pairs_exactly_scored_total": candidate_count_total,
},
"finding_count": len(findings),
"findings": findings,
}
payload = canonical(result) + b"\n"
result_sha256 = write_exclusive(
args.result, payload, policy["limits"]["max_result_bytes"]
)
print(json.dumps({"outcome": result["outcome"], "result_sha256": result_sha256}))
return 0 if result["outcome"] == "PASS" else 1
def validate_result_readback(
result_path: Path,
policy_path: Path,
bundle_path: Path,
expected_outcome: str,
cosign_timeout_seconds: float,
) -> str:
policy, threshold_numerator, threshold_denominator, receipt = verify_policy(
policy_path, bundle_path, cosign_timeout_seconds
)
result_raw = stable_read(result_path, policy["limits"]["max_result_bytes"])
result = exact_object(strict_json(result_raw, "result"), {
"schema_version", "outcome", "policy_version", "policy_sha256",
"policy_bundle_sha256", "policy_key_sha256", "normalization",
"shingle_algorithm", "near_duplicate_threshold", "population",
"coverage", "finding_count", "findings",
}, "result")
if result_raw != canonical(result) + b"\n":
raise GateError("result: bytes are not canonical JSON plus one newline")
expected = {
"schema_version": RESULT_SCHEMA,
"outcome": expected_outcome,
"policy_version": policy["policy_version"],
**receipt,
"normalization": NORMALIZATION,
"shingle_algorithm": SHINGLE_ALGORITHM,
"near_duplicate_threshold": policy["near_duplicate_threshold"],
}
for field, expected_value in expected.items():
if result[field] != expected_value:
raise GateError(f"result: {field} differs from the verified run")
findings = result["findings"]
finding_count = result["finding_count"]
if (
not isinstance(findings, list)
or type(finding_count) is not int
or not 0 <= finding_count <= policy["limits"]["max_findings"]
or finding_count != len(findings)
):
raise GateError("result: finding population is invalid")
if (expected_outcome == "PASS") != (finding_count == 0):
raise GateError("result: outcome and finding population disagree")
near_findings = defaultdict(int)
finding_fields = {
"within_split_exact": {"kind", "split", "text_sha256", "ids"},
"cross_split_id": {"kind", "left_split", "right_split", "id"},
"cross_split_exact": {
"kind", "left_split", "right_split", "text_sha256",
"left_ids", "right_ids",
},
"cross_split_near": {
"kind", "left_split", "right_split", "left_id", "right_id",
"jaccard",
},
}
for index, finding_value in enumerate(findings):
if not isinstance(finding_value, dict):
raise GateError(f"result.findings[{index}]: expected object")
kind = finding_value.get("kind")
if kind not in finding_fields:
raise GateError(f"result.findings[{index}]: unknown finding kind")
finding = exact_object(
finding_value, finding_fields[kind], f"result.findings[{index}]"
)
if kind == "within_split_exact":
if finding["split"] not in SPLITS:
raise GateError("result: invalid within-split coverage")
identifiers = finding["ids"]
if (
not isinstance(identifiers, list) or len(identifiers) < 2
or any(not isinstance(item, str) or not item for item in identifiers)
or len(set(identifiers)) != len(identifiers)
or not isinstance(finding["text_sha256"], str)
or SHA256.fullmatch(finding["text_sha256"]) is None
):
raise GateError("result: invalid within-split finding")
continue
pair = (finding["left_split"], finding["right_split"])
if pair not in PAIRS:
raise GateError("result: invalid cross-split pair")
if kind == "cross_split_id":
if not isinstance(finding["id"], str) or not finding["id"]:
raise GateError("result: invalid cross-split identifier")
elif kind == "cross_split_exact":
if (
not isinstance(finding["text_sha256"], str)
or SHA256.fullmatch(finding["text_sha256"]) is None
):
raise GateError("result: invalid exact-duplicate digest")
for field in ("left_ids", "right_ids"):
identifiers = finding[field]
if (
not isinstance(identifiers, list) or not identifiers
or any(not isinstance(item, str) or not item for item in identifiers)
or len(set(identifiers)) != len(identifiers)
):
raise GateError("result: invalid exact-duplicate identifiers")
else:
if any(
not isinstance(finding[field], str) or not finding[field]
for field in ("left_id", "right_id")
):
raise GateError("result: invalid near-duplicate identifiers")
match = re.fullmatch(r"([1-9][0-9]*)/([1-9][0-9]*)", finding["jaccard"])
if match is None:
raise GateError("result: invalid Jaccard evidence")
numerator, denominator = map(int, match.groups())
if (
numerator > denominator
or numerator * threshold_denominator
< threshold_numerator * denominator
):
raise GateError("result: near-duplicate evidence is below policy")
near_findings[pair] += 1
population = exact_object(result["population"], set(SPLITS), "result.population")
for split in SPLITS:
evidence = exact_object(population[split], {
"path", "raw_sha256", "row_count", "population_sha256",
}, f"result.population.{split}")
spec = policy["splits"][split]
if (
evidence["path"] != spec["path"]
or evidence["raw_sha256"] != spec["sha256"]
or evidence["row_count"] != spec["row_count"]
or not isinstance(evidence["population_sha256"], str)
or SHA256.fullmatch(evidence["population_sha256"]) is None
):
raise GateError(f"result.population.{split}: binding differs")
coverage = exact_object(result["coverage"], {
"within_split_exact", "cross_split_exact_and_near",
"candidate_pairs_exactly_scored_total",
}, "result.coverage")
if coverage["within_split_exact"] != list(SPLITS):
raise GateError("result.coverage: within-split population is incomplete")
pair_coverage = coverage["cross_split_exact_and_near"]
if not isinstance(pair_coverage, list) or len(pair_coverage) != len(PAIRS):
raise GateError("result.coverage: split-pair population is incomplete")
candidate_total = 0
for index, expected_pair in enumerate(PAIRS):
item = exact_object(pair_coverage[index], {
"left_split", "right_split", "left_rows", "right_rows",
"candidate_pairs_exactly_scored", "near_findings",
}, f"result.coverage.pair[{index}]")
pair = (item["left_split"], item["right_split"])
candidate_count = item["candidate_pairs_exactly_scored"]
near_count = item["near_findings"]
if pair != expected_pair:
raise GateError("result.coverage: split-pair order differs")
if (
item["left_rows"] != policy["splits"][pair[0]]["row_count"]
or item["right_rows"] != policy["splits"][pair[1]]["row_count"]
or type(candidate_count) is not int or candidate_count < 0
or candidate_count > item["left_rows"] * item["right_rows"]
or type(near_count) is not int or not 0 <= near_count <= candidate_count
or near_count != near_findings[pair]
):
raise GateError("result.coverage: split-pair evidence differs")
candidate_total += candidate_count
reported_total = coverage["candidate_pairs_exactly_scored_total"]
if (
type(reported_total) is not int
or reported_total != candidate_total
or reported_total > policy["limits"]["max_candidate_pairs_total"]
):
raise GateError("result.coverage: aggregate candidate count differs")
return hashlib.sha256(result_raw).hexdigest()
def validate_result_command(args: argparse.Namespace) -> int:
print(validate_result_readback(
args.result, args.policy, args.bundle, args.expected_outcome,
args.cosign_timeout_seconds,
))
return 0
def make_parser() -> argparse.ArgumentParser:
root = argparse.ArgumentParser()
commands = root.add_subparsers(dest="command", required=True)
build = commands.add_parser("build-policy")
build.add_argument("--dataset-root", type=Path, required=True)
build.add_argument("--policy-version", required=True)
build.add_argument("--threshold", required=True)
for name in HARD_LIMITS:
build.add_argument("--" + name.replace("_", "-"), dest=name, type=int, required=True)
build.add_argument("--output", type=Path, required=True)
build.set_defaults(handler=build_policy)
verify = commands.add_parser("check")
verify.add_argument("--dataset-root", type=Path, required=True)
verify.add_argument("--policy", type=Path, required=True)
verify.add_argument("--bundle", type=Path, required=True)
verify.add_argument("--result", type=Path, required=True)
verify.add_argument("--cosign-timeout-seconds", type=float, required=True)
verify.set_defaults(handler=check)
readback = commands.add_parser("validate-result")
readback.add_argument("--result", type=Path, required=True)
readback.add_argument("--policy", type=Path, required=True)
readback.add_argument("--bundle", type=Path, required=True)
readback.add_argument("--expected-outcome", choices=("PASS", "FAIL"), required=True)
readback.add_argument("--cosign-timeout-seconds", type=float, required=True)
readback.set_defaults(handler=validate_result_command)
return root
def main() -> int:
try:
args = make_parser().parse_args()
return args.handler(args)
except Exception as exc:
print(f"split_integrity_error:{type(exc).__name__}:{exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
Build, sign, and enforce the versioned manifest
#!/usr/bin/env bash
# Linux CI gate; use a toolchain image pinned to Cosign v3.1.2.
set -euo pipefail
umask 077
: "${H002006_POLICY_VERSION:?reviewed policy version required}"
: "${H002006_NEAR_DUPLICATE_THRESHOLD:?reviewed threshold required}"
: "${H002006_MAX_FILE_BYTES:?signed byte cap required}"
: "${H002006_MAX_ROWS_PER_SPLIT:?signed row cap required}"
: "${H002006_MAX_CANDIDATE_PAIRS_TOTAL:?signed aggregate candidate cap required}"
: "${H002006_MAX_FINDINGS:?signed finding cap required}"
: "${H002006_MAX_RESULT_BYTES:?signed result byte cap required}"
: "${H002006_COSIGN_TIMEOUT_SECONDS:?verification timeout required}"
: "${H002006_POLICY_SIGNING_KEY:?policy signing key required}"
: "${H002006_RESULT_SIGNING_KEY:?result signing key required}"
install -d -m 0700 evidence/runs
run_dir="$(mktemp -d evidence/runs/h002006.XXXXXXXX)"
readonly run_dir
readonly policy_path="$run_dir/split-policy.json"
readonly policy_bundle_path="$run_dir/split-policy.sigstore.json"
readonly result_path="$run_dir/split-result.json"
readonly result_bundle_path="$run_dir/split-result.sigstore.json"
receipt_ready=0
cleanup() {
status=$?
if [ "$receipt_ready" -ne 1 ]; then
rm -f -- "$result_bundle_path" "$result_path" || true
fi
trap - EXIT HUP INT TERM
exit "$status"
}
on_signal() {
trap - HUP INT TERM
exit 2
}
trap cleanup EXIT
trap on_signal HUP INT TERM
python dataset_checks/split_integrity.py build-policy \
--dataset-root . \
--policy-version "$H002006_POLICY_VERSION" \
--threshold "$H002006_NEAR_DUPLICATE_THRESHOLD" \
--max-file-bytes "$H002006_MAX_FILE_BYTES" \
--max-rows-per-split "$H002006_MAX_ROWS_PER_SPLIT" \
--max-candidate-pairs-total "$H002006_MAX_CANDIDATE_PAIRS_TOTAL" \
--max-findings "$H002006_MAX_FINDINGS" \
--max-result-bytes "$H002006_MAX_RESULT_BYTES" \
--output "$policy_path"
# Sign only the just-created v2 policy from this run's private directory.
cosign sign-blob --yes \
--key env://H002006_POLICY_SIGNING_KEY \
--bundle "$policy_bundle_path" \
"$policy_path"
chmod 0400 "$policy_path" "$policy_bundle_path"
set +e
python dataset_checks/split_integrity.py check \
--dataset-root . \
--policy "$policy_path" \
--bundle "$policy_bundle_path" \
--result "$result_path" \
--cosign-timeout-seconds "$H002006_COSIGN_TIMEOUT_SECONDS"
check_rc=$?
set -e
case "$check_rc" in
0) expected_outcome=PASS ;;
1) expected_outcome=FAIL ;;
*) echo "split-integrity execution ERROR (exit $check_rc); no receipt signed" >&2
exit "$check_rc" ;;
esac
readonly expected_outcome
# Reject a stale, partial, malformed, or policy-mismatched result before signing.
result_sha256="$(python dataset_checks/split_integrity.py validate-result \
--result "$result_path" \
--policy "$policy_path" \
--bundle "$policy_bundle_path" \
--expected-outcome "$expected_outcome" \
--cosign-timeout-seconds "$H002006_COSIGN_TIMEOUT_SECONDS")"
readonly result_sha256
chmod 0400 "$result_path"
cosign sign-blob --yes \
--key env://H002006_RESULT_SIGNING_KEY \
--bundle "$result_bundle_path" \
"$result_path"
chmod 0400 "$result_bundle_path"
cosign verify-blob \
--key /opt/aidefend/trust/h002006-result.pub \
--bundle "$result_bundle_path" \
"$result_path"
verified_result_sha256="$(python dataset_checks/split_integrity.py validate-result \
--result "$result_path" \
--policy "$policy_path" \
--bundle "$policy_bundle_path" \
--expected-outcome "$expected_outcome" \
--cosign-timeout-seconds "$H002006_COSIGN_TIMEOUT_SECONDS")"
readonly verified_result_sha256
test "$verified_result_sha256" = "$result_sha256"
# Only validated PASS/FAIL receipts survive cleanup; CI publishes these exact paths.
receipt_ready=1
printf 'result=%s\nbundle=%s\n' "$result_path" "$result_bundle_path"
exit "$check_rc"
Action: Run the checker before training and evaluation. A signed v2 manifest binds the exact bytes and row counts for all three split files, the versioned normalization and shingle algorithms, the near-duplicate threshold, and one aggregate execution cap across all three split pairs. PASS requires complete processing of all three populations; any finding is FAIL. Each run writes to a new private directory, accepts only checker exit 0 or 1, validates canonical result bytes and policy/population/coverage bindings before and after signing, and preserves only the resulting signed PASS/FAIL receipt. Signature, input, aggregate-capacity, execution, or readback problems are ERROR; cleanup removes the result and bundle so they cannot be published as usable evidence. Publish the printed paths through the CI system's immutable artifact store under its retention policy.
AID-H-002.006-G003
Apply split-level access control so training jobs can only read the training partition and evaluation jobs can only read the validation or test partitions they need.
Concept:
Prevent accidental or intentional cross-partition access through storage-level controls, not only developer discipline.
Example storage layout
s3://ml-datasets/project-x/train/
s3://ml-datasets/project-x/validation/
s3://ml-datasets/project-x/test/
Operational notes: Give the training role read access to train/ only. Give the evaluation role read access to validation/ and test/ only. Turn on storage access logging so you can investigate cross-partition reads.
Before you begin
Apply AID-H-002.006-G003 only when one source population is partitioned into training, validation, test, calibration, holdout, or temporal slices.
Keep the action safely bounded
block training or evaluation on overlap, missing members, unauthorized partition access, unreproducible assignment, or policy-exceeding representativeness error.
AID-H-002.006-G004
Verify that the split remains representative by checking class balance, temporal integrity, and important feature distributions across partitions.
Concept:
A secure split is not only isolated; it also needs to be representative enough for valid evaluation.
Example checks
1. Compare class proportions across train/validation/test.
2. For time-series data, ensure test data does not leak earlier-than-training time windows.
3. Compare key feature distributions with KS test or similar methods.
4. Alert if important features drift too far across partitions.
Operational notes: Keep these reports as part of the evaluation evidence. They are useful both for science quality and for security review because non-representative splits can hide model weakness.
Before you begin
Apply AID-H-002.006-G004 only when one source population is partitioned into training, validation, test, calibration, holdout, or temporal slices.
Keep the action safely bounded
block training or evaluation on overlap, missing members, unauthorized partition access, unreproducible assignment, or policy-exceeding representativeness error.
Open-source tools
- scikit-learn
- DVC
- datasketch
- simhash
- GX Core (Great Expectations)
- Pandera
- Python hashlib (content-addressable split hashing)
- NVIDIA NeMo Curator (duplicate and near-duplicate analysis for cross-partition leakage checks only; split creation, immutability, stratification, and access controls remain external)
Commercial / hosted tools
- Databricks Unity Catalog
- Amazon SageMaker Processing (managed split jobs with auditability)
- Weights & Biases Artifacts (versioned split artifacts and lineage)
Threat mappings
MITRE ATLAS
- AML.T0059 Erode Dataset Integrity (split manipulation erodes partition integrity)
MAESTRO
- Manipulation of Evaluation Metrics (L5) (train-test leakage inflates evaluation)
OWASP LLM Top 10 2026
- N/A
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- AISubtech-2.3.2 Verifier Manipulation
- AISubtech-2.3.3 Shortcut Acquisition
- AITech-2.3 Reward Hacking
Google Secure AI Framework 2.0 - Risks
- N/A
Databricks AI Security Framework 3.0
- Data Prep 2.4: Adversarial partitions
- Evaluation 6.1: Evaluation data poisoning (train-test leakage corrupts evaluation)
- Evaluation 6.2: Insufficient evaluation data