AID-DV-007: Training-Data Provenance & Behavioral Canaries
Insert authorized, synthetic, non-sensitive behavioral canaries into a controlled training or fine-tuning corpus and later run a pre-registered, statistically powered recall audit against the exact model version. The control produces evidence that a measured model exhibits behavior associated with the registered canary set; it does not by itself prove who copied data, when training occurred, or that the surrounding corpus was stolen.
Pillars: data, model
Lifecycle: building, validation, improvement
Scope boundary
Owns authorized behavioral-canary design, insertion provenance, blinded measurement against the exact model and query profile, statistical evidence, and handoff. Its result associates measured behavior with the registered canary set; it does not prove who copied data, when training occurred, that the surrounding corpus was stolen, or that a training anomaly is poisoning.
AID-DV-002: AID-DV-007 inserts and measures training-specific behavioral canaries; AID-DV-002 owns operational honey records, decoy artifacts, fake credentials, and access-triggered tokens. AID-DV-002 alerts on interaction with planted material, while this technique requires a registered training insertion and a statistically powered model-behavior audit.
AID-DV-004: AID-DV-007 produces training-bound behavioral-canary evidence; AID-DV-004 owns external exact-marker observation and leak investigation. An externally observed marker is handled by AID-DV-004 only when the behavioral canary was deliberately registered for that investigation path.
AID-H-007.002: AID-DV-007 measures registered canary behavior after authorized insertion; AID-H-007.002 monitors the integrity and behavior of the running training job. Canary loss or gradient signals may support investigation, but training telemetry and poisoning detection do not become a provenance verdict under this technique.
Warning
Experimental Behavioral-Canary Limits
Behavioral canaries can be filtered, forgotten, paraphrased, suppressed by alignment, or reproduced coincidentally. Insertion also changes the training corpus and may create privacy, licensing, quality, or contamination concerns. Require data-owner, legal, privacy, and model-owner approval; use only synthetic non-sensitive facts; pre-register the canary count, repetition budget, prompts, negative controls, model/version, decoding settings, statistical test, power target, and escalation threshold. A positive result is an investigation signal that requires separate lineage, access, contractual, or leak corroboration. A negative result applies only to the exact measured protocol and does not establish the absence of unauthorized training. Poisoning claims likewise require separate corroboration.
Implementation guidance
- AID-DV-007-G001: Create a versioned registry of synthetic behavioral canaries, negative controls, authorized insertion records, repetitions, prompts, and expected responses before the training run begins.
- AID-DV-007-G002: Run a blinded canary-versus-negative-control audit with a pre-registered statistical test and emit conservative provenance findings bound to the exact model and query profile.
AID-DV-007-G001
Create a versioned registry of synthetic behavioral canaries, negative controls, authorized insertion records, repetitions, prompts, and expected responses before the training run begins.
Method boundary and prerequisites
This step handles authorized synthetic-canary design, blinded negative controls, the exact insertion plan, and insertion provenance. The next step handles behavioral querying and statistical interpretation. Reuse AID-H-021.002 to verify the exact signed design before use and AID-M-002.002 to sign the generated registry, insertion plan, and insertion receipt. A hash supplied in an environment variable is not authorization or signature evidence.
Verify the signed design
The security/data/model owners approve a strict design containing the exact dataset digest, counts, repetition budget, authorization receipt, and expiry. Numeric values are policy inputs, not AIDEFEND defaults.
{
"schema_version": "aidefend.behavioral_canary_design.v2",
"policy_version": "canary-policy-2026-07",
"build_id": "support-finetune-2026q3-build-001",
"authorization_id": "approval://ml-governance/CANARY-1842",
"authorization_receipt_sha256": "cfb79f4ed61e88f248d69a56b7b22b0149f3d6544413eef235659018b4d558cb",
"authorization_expires_at": "2026-12-31T23:59:59Z",
"dataset_id": "support-finetune-2026q3",
"source_dataset_sha256": "91df40bf4d880c9f5f69c5d7e9bd2b67d1bf1e61c342e30256f3a7c641b1d724",
"canary_count": 40,
"control_count": 40,
"repetitions": 3,
"maximum_registry_records": 100,
"maximum_insertion_rows": 200,
"maximum_registry_bytes": 1048576,
"maximum_insertion_plan_bytes": 1048576,
"maximum_build_receipt_bytes": 131072
}The example build ID, counts, digests, byte ceilings, and population ceilings are illustrative complete values. Resolve them, power, expiry, and repetition from the signed experiment policy. Preprovision the fixed /var/lib/aidefend/canary-builds/<build_id> directory under root-owned, non-writable ancestors, chown only that empty leaf to the builder identity, and set it to mode 0700; the builder never creates or selects an output path. Use only synthetic non-sensitive text and keep expected phrases inaccessible to model developers, trainers, and the later blinded query operator.
Build the registry and exact insertion plan
# File: provenance/build_behavioral_canaries.py
from __future__ import annotations
import hashlib
import json
import math
import os
import re
import stat
import subprocess
import tempfile
import secrets
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
CANARY_DESIGN_PATH = Path("/run/policy/behavioral-canary-design.json")
CANARY_DESIGN_BUNDLE_PATH = Path("/run/policy/behavioral-canary-design.sigstore.json")
CANARY_DESIGN_VERIFY_KEY_PATH = Path("/run/trust/behavioral-canary-design.pub")
ARTIFACT_ROOT_COMPONENTS = ("var", "lib", "aidefend", "canary-builds")
# Compiled ceilings are used only before the signed design can supply output bounds.
DESIGN_BOOTSTRAP_ARTIFACT_CEILING = 1 * 1024 * 1024 # bytes
DESIGN_BUNDLE_BOOTSTRAP_ARTIFACT_CEILING = 1 * 1024 * 1024 # bytes
DESIGN_KEY_BOOTSTRAP_ARTIFACT_CEILING = 64 * 1024 # bytes
COSIGN_VERIFICATION_DEADLINE_CEILING_SECONDS = 10.0
BUILD_ID_PATTERN = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}")
DESIGN_FIELDS = {
"schema_version", "policy_version", "build_id", "authorization_id",
"authorization_receipt_sha256", "authorization_expires_at",
"dataset_id", "source_dataset_sha256", "canary_count",
"control_count", "repetitions", "maximum_registry_records",
"maximum_insertion_rows", "maximum_registry_bytes",
"maximum_insertion_plan_bytes", "maximum_build_receipt_bytes",
}
HEX = set("0123456789abcdef")
@dataclass(frozen=True)
class VerifiedJson:
payload: dict
payload_sha256: str
signature_sha256: str
signer_identity: str
verification_receipt_sha256: str
def reject_duplicate_keys(pairs: list[tuple[str, object]]) -> dict:
result: dict = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON key: {key}")
result[key] = value
return result
class SignedJsonVerifier:
def __init__(
self,
*,
verify_key_path: Path,
maximum_payload_bytes: int,
maximum_signature_bytes: int,
maximum_key_bytes: int,
command_timeout_seconds: float,
require_privileged_inputs: bool = False,
) -> None:
for name, value in (
("maximum_payload_bytes", maximum_payload_bytes),
("maximum_signature_bytes", maximum_signature_bytes),
("maximum_key_bytes", maximum_key_bytes),
):
if type(value) is not int or value < 1:
raise ValueError(f"{name} must be a positive policy integer")
if not math.isfinite(command_timeout_seconds) or command_timeout_seconds <= 0:
raise ValueError("command timeout must be a positive policy value")
self.maximum_payload_bytes = maximum_payload_bytes
self.maximum_signature_bytes = maximum_signature_bytes
self.command_timeout_seconds = command_timeout_seconds
self.require_privileged_inputs = require_privileged_inputs
self.key_bytes = self._read_exact(
verify_key_path, maximum_key_bytes, privileged_trust=True,
)
self.key_sha256 = hashlib.sha256(self.key_bytes).hexdigest()
self.signer_identity = "key-sha256:" + self.key_sha256
@staticmethod
def _read_exact(
path: Path, maximum_bytes: int, *, privileged_trust: bool = False,
) -> bytes:
if not hasattr(os, "O_NOFOLLOW"):
raise OSError("secure no-follow reads are unavailable")
descriptor = os.open(
path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW
)
try:
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode):
raise ValueError("signed input must be a regular file")
if privileged_trust and (before.st_uid != 0 or before.st_mode & 0o022):
raise PermissionError(
"verification key must be root-owned and not group/world writable"
)
if before.st_size < 1 or before.st_size > maximum_bytes:
raise ValueError("signed input is empty or exceeds policy bound")
chunks: list[bytes] = []
remaining = before.st_size
while remaining:
chunk = os.read(descriptor, min(65536, remaining))
if not chunk:
raise OSError("short signed-input read")
chunks.append(chunk)
remaining -= len(chunk)
if os.read(descriptor, 1):
raise OSError("signed input grew during read")
after = os.fstat(descriptor)
fingerprint = lambda item: (
item.st_dev, item.st_ino, item.st_size,
item.st_mtime_ns, item.st_ctime_ns,
)
if fingerprint(before) != fingerprint(after):
raise OSError("signed input changed during read")
return b"".join(chunks)
finally:
os.close(descriptor)
def load_exact(
self, *, payload_path: Path, signature_path: Path
) -> VerifiedJson:
payload = self._read_exact(
payload_path, self.maximum_payload_bytes,
privileged_trust=self.require_privileged_inputs,
)
signature = self._read_exact(
signature_path, self.maximum_signature_bytes,
privileged_trust=self.require_privileged_inputs,
)
with tempfile.TemporaryDirectory(prefix="aidefend-signed-json-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload.json"
signature_snapshot = root / "payload.sigstore.json"
key_snapshot = root / "verify.pub"
for path, value in (
(payload_snapshot, payload),
(signature_snapshot, signature),
(key_snapshot, self.key_bytes),
):
path.write_bytes(value)
os.chmod(path, 0o400)
subprocess.run(
[
"cosign", "verify-blob", "--key", str(key_snapshot),
"--bundle", str(signature_snapshot),
str(payload_snapshot),
],
check=True,
capture_output=True,
text=True,
timeout=self.command_timeout_seconds,
)
verified = payload_snapshot.read_bytes()
verified_bundle = signature_snapshot.read_bytes()
verified_key = key_snapshot.read_bytes()
if (
verified != payload
or verified_bundle != signature
or verified_key != self.key_bytes
):
raise OSError("verified payload, bundle, or key snapshot changed")
parsed = json.loads(
verified.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError("non-finite JSON number: " + value)
),
)
if not isinstance(parsed, dict):
raise ValueError("signed JSON root must be an object")
payload_sha256 = hashlib.sha256(verified).hexdigest()
signature_sha256 = hashlib.sha256(signature).hexdigest()
receipt = canonical({
"schema_version": "aidefend.signed-json-verification.v1",
"payload_sha256": payload_sha256,
"signature_sha256": signature_sha256,
"verify_key_sha256": self.key_sha256,
"signer_identity": self.signer_identity,
})
return VerifiedJson(
payload=parsed,
payload_sha256=payload_sha256,
signature_sha256=signature_sha256,
signer_identity=self.signer_identity,
verification_receipt_sha256=hashlib.sha256(receipt).hexdigest(),
)
@dataclass(frozen=True)
class Canary:
canary_id: str
cohort: str
prompt: str
expected_phrase: str
repetitions: int
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def digest_ok(value: object) -> bool:
return isinstance(value, str) and len(value) == 64 and not (set(value) - HEX)
def parse_utc(value: object) -> datetime:
if not isinstance(value, str) or not value.endswith("Z"):
raise ValueError("authorization expiry must be UTC Z time")
parsed = datetime.fromisoformat(value[:-1] + "+00:00")
if parsed.tzinfo is None:
raise ValueError("authorization expiry must be timezone aware")
return parsed
def validate_verified_design(verified: VerifiedJson, now: datetime) -> dict:
if not isinstance(verified, VerifiedJson):
raise TypeError("verified signed design is required")
for digest in (
verified.payload_sha256,
verified.signature_sha256,
verified.verification_receipt_sha256,
):
if not digest_ok(digest):
raise ValueError("signature verification receipt has an invalid digest")
if not verified.signer_identity:
raise ValueError("verified signer identity is required")
design = verified.payload
if not isinstance(design, dict) or set(design) != DESIGN_FIELDS:
raise ValueError("canary design schema mismatch")
if design["schema_version"] != "aidefend.behavioral_canary_design.v2":
raise ValueError("unsupported canary design schema")
for field in ("policy_version", "build_id", "authorization_id", "dataset_id"):
if not isinstance(design[field], str) or not design[field].strip():
raise ValueError(f"invalid {field}")
if (
BUILD_ID_PATTERN.fullmatch(design["build_id"]) is None
or design["build_id"] in {".", ".."}
):
raise ValueError("build_id must be a canonical safe directory component")
if not digest_ok(design["authorization_receipt_sha256"]) or not digest_ok(
design["source_dataset_sha256"]
):
raise ValueError("design artifact digest is invalid")
for field in (
"canary_count", "control_count", "repetitions",
"maximum_registry_records", "maximum_insertion_rows",
"maximum_registry_bytes", "maximum_insertion_plan_bytes",
"maximum_build_receipt_bytes",
):
if type(design[field]) is not int or design[field] < 1:
raise ValueError(f"{field} must be a positive policy integer")
if (
design["canary_count"] + design["control_count"]
> design["maximum_registry_records"]
or design["canary_count"] * design["repetitions"]
> design["maximum_insertion_rows"]
):
raise ValueError("requested canary population exceeds the signed design bounds")
if parse_utc(design["authorization_expires_at"]) <= now:
raise PermissionError("canary authorization expired")
return design
def build_registry(verified: VerifiedJson, now: datetime) -> tuple[dict, list[dict], dict]:
design = validate_verified_design(verified, now)
records: list[Canary] = []
for cohort, count in (
("inserted", design["canary_count"]),
("negative_control", design["control_count"]),
):
for _ in range(count):
identifier = "cny_" + secrets.token_hex(12)
phrase = "SYNTH-" + secrets.token_hex(16)
records.append(
Canary(
canary_id=identifier,
cohort=cohort,
prompt=f"What synthetic catalog phrase is registered for record {identifier}?",
expected_phrase=phrase,
repetitions=design["repetitions"] if cohort == "inserted" else 0,
)
)
ids = [record.canary_id for record in records]
phrases = [record.expected_phrase for record in records]
if len(ids) != len(set(ids)) or len(phrases) != len(set(phrases)):
raise RuntimeError("random canary identifier collision")
registry_payload = {
"schema_version": "aidefend.behavioral_canary_registry.v1",
"design_sha256": verified.payload_sha256,
"policy_version": design["policy_version"],
"authorization_id": design["authorization_id"],
"authorization_receipt_sha256": design["authorization_receipt_sha256"],
"dataset_id": design["dataset_id"],
"source_dataset_sha256": design["source_dataset_sha256"],
"records": [asdict(record) for record in records],
}
rows = [
{
"canary_id": record.canary_id,
"repetition_id": repetition,
"text": record.prompt + " " + record.expected_phrase,
}
for record in records
if record.cohort == "inserted"
for repetition in range(record.repetitions)
]
population = sorted(
(row["canary_id"], row["repetition_id"]) for row in rows
)
registry_bytes = canonical(registry_payload)
insertion_plan_bytes = canonical(rows)
if (
len(registry_bytes) > design["maximum_registry_bytes"]
or len(insertion_plan_bytes) > design["maximum_insertion_plan_bytes"]
):
raise RuntimeError("generated artifact exceeds its signed byte bound")
build_receipt = {
"schema_version": "aidefend.behavioral_canary_build.v2",
"commit_state": "COMMITTED",
"build_id": design["build_id"],
"design_sha256": verified.payload_sha256,
"registry_sha256": hashlib.sha256(registry_bytes).hexdigest(),
"registry_bytes": len(registry_bytes),
"insertion_plan_sha256": hashlib.sha256(insertion_plan_bytes).hexdigest(),
"insertion_plan_bytes": len(insertion_plan_bytes),
"expected_population_sha256": hashlib.sha256(canonical(population)).hexdigest(),
"inserted_canary_count": design["canary_count"],
"negative_control_count": design["control_count"],
"registry_record_count": len(records),
"insertion_row_count": len(rows),
"maximum_registry_records": design["maximum_registry_records"],
"maximum_insertion_rows": design["maximum_insertion_rows"],
"maximum_registry_bytes": design["maximum_registry_bytes"],
"maximum_insertion_plan_bytes": design["maximum_insertion_plan_bytes"],
"built_at": now.isoformat(),
"signer_identity": verified.signer_identity,
"design_verification_receipt_sha256": verified.verification_receipt_sha256,
}
if len(canonical(build_receipt)) > design["maximum_build_receipt_bytes"]:
raise RuntimeError("build receipt exceeds its signed byte bound")
return registry_payload, rows, build_receipt
def directory_open_flags() -> int:
if not all(hasattr(os, name) for name in ("O_DIRECTORY", "O_NOFOLLOW")):
raise OSError("secure directory-descriptor operations are unavailable")
return os.O_RDONLY | os.O_CLOEXEC | os.O_DIRECTORY | os.O_NOFOLLOW
def validate_directory(descriptor: int, *, owner_uid: int, private: bool) -> None:
metadata = os.fstat(descriptor)
forbidden = 0o077 if private else 0o022
if (
not stat.S_ISDIR(metadata.st_mode)
or metadata.st_uid != owner_uid
or metadata.st_mode & forbidden
):
raise PermissionError("artifact directory ownership or mode is unsafe")
def open_trusted_artifact_root() -> int:
flags = directory_open_flags()
descriptor = os.open("/", flags)
try:
validate_directory(descriptor, owner_uid=0, private=False)
for component in ARTIFACT_ROOT_COMPONENTS:
candidate = os.open(component, flags, dir_fd=descriptor)
try:
validate_directory(candidate, owner_uid=0, private=False)
except Exception:
os.close(candidate)
raise
os.close(descriptor)
descriptor = candidate
return descriptor
except Exception:
os.close(descriptor)
raise
def open_preprovisioned_build_directory(root_descriptor: int, build_id: str) -> int:
descriptor = os.open(
build_id, directory_open_flags(), dir_fd=root_descriptor,
)
try:
validate_directory(descriptor, owner_uid=os.geteuid(), private=True)
if os.listdir(descriptor):
raise FileExistsError(
"preprovisioned build directory must be empty before publication"
)
return descriptor
except Exception:
os.close(descriptor)
raise
def read_descriptor_exact(descriptor: int, expected_bytes: int) -> bytes:
os.lseek(descriptor, 0, os.SEEK_SET)
chunks: list[bytes] = []
remaining = expected_bytes
while remaining:
chunk = os.read(descriptor, min(65536, remaining))
if not chunk:
raise OSError("short artifact readback")
chunks.append(chunk)
remaining -= len(chunk)
if os.read(descriptor, 1):
raise OSError("artifact grew during readback")
return b"".join(chunks)
def publish_new(
directory_descriptor: int, name: str, payload: bytes, maximum_bytes: int,
) -> None:
allowed_names = {
"canary-registry.json", "canary-insertions.json",
"canary-build-receipt.json",
}
if name not in allowed_names or not 1 <= len(payload) <= maximum_bytes:
raise ValueError("artifact name or signed byte bound is invalid")
temporary_name = f".{name}.{secrets.token_hex(12)}.tmp"
flags = os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | os.O_NOFOLLOW
descriptor = os.open(
temporary_name, flags, 0o600, dir_fd=directory_descriptor,
)
temporary_exists = True
try:
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode) or before.st_size != 0:
raise OSError("new artifact is not an empty regular file")
remaining = memoryview(payload)
while remaining:
written = os.write(descriptor, remaining)
if written <= 0:
raise OSError("short artifact write")
remaining = remaining[written:]
os.fsync(descriptor)
written_metadata = os.fstat(descriptor)
if written_metadata.st_size != len(payload):
raise OSError("artifact size differs after write")
if read_descriptor_exact(descriptor, len(payload)) != payload:
raise OSError("artifact readback differs")
read_metadata = os.fstat(descriptor)
if (
written_metadata.st_dev, written_metadata.st_ino, written_metadata.st_size,
written_metadata.st_mtime_ns, written_metadata.st_ctime_ns,
) != (
read_metadata.st_dev, read_metadata.st_ino, read_metadata.st_size,
read_metadata.st_mtime_ns, read_metadata.st_ctime_ns,
):
raise OSError("artifact changed during readback")
os.link(
temporary_name, name, src_dir_fd=directory_descriptor,
dst_dir_fd=directory_descriptor, follow_symlinks=False,
)
final_descriptor = os.open(
name, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW,
dir_fd=directory_descriptor,
)
try:
final_metadata = os.fstat(final_descriptor)
if (final_metadata.st_dev, final_metadata.st_ino) != (
written_metadata.st_dev, written_metadata.st_ino
) or read_descriptor_exact(final_descriptor, len(payload)) != payload:
raise OSError("atomically published artifact differs")
finally:
os.close(final_descriptor)
os.unlink(temporary_name, dir_fd=directory_descriptor)
temporary_exists = False
os.fsync(directory_descriptor)
finally:
os.close(descriptor)
if temporary_exists:
try:
os.unlink(temporary_name, dir_fd=directory_descriptor)
except FileNotFoundError:
pass
def run(verifier: SignedJsonVerifier) -> None:
verified = verifier.load_exact(
payload_path=CANARY_DESIGN_PATH,
signature_path=CANARY_DESIGN_BUNDLE_PATH,
)
registry, insertions, receipt = build_registry(
verified, datetime.now(timezone.utc)
)
design = verified.payload
registry_bytes = canonical(registry)
insertion_bytes = canonical(insertions)
receipt_bytes = canonical(receipt)
root_descriptor = open_trusted_artifact_root()
try:
build_descriptor = open_preprovisioned_build_directory(
root_descriptor, design["build_id"],
)
try:
publish_new(
build_descriptor, "canary-registry.json", registry_bytes,
design["maximum_registry_bytes"],
)
publish_new(
build_descriptor, "canary-insertions.json", insertion_bytes,
design["maximum_insertion_plan_bytes"],
)
# Publish the COMMITTED receipt last; its absence means the build is incomplete.
publish_new(
build_descriptor, "canary-build-receipt.json", receipt_bytes,
design["maximum_build_receipt_bytes"],
)
expected = {
"canary-registry.json", "canary-insertions.json",
"canary-build-receipt.json",
}
if set(os.listdir(build_descriptor)) != expected:
raise OSError("published build population differs")
os.fsync(build_descriptor)
finally:
os.close(build_descriptor)
finally:
os.close(root_descriptor)
def build_verifier() -> SignedJsonVerifier:
return SignedJsonVerifier(
verify_key_path=CANARY_DESIGN_VERIFY_KEY_PATH,
maximum_payload_bytes=DESIGN_BOOTSTRAP_ARTIFACT_CEILING,
maximum_signature_bytes=DESIGN_BUNDLE_BOOTSTRAP_ARTIFACT_CEILING,
maximum_key_bytes=DESIGN_KEY_BOOTSTRAP_ARTIFACT_CEILING,
command_timeout_seconds=COSIGN_VERIFICATION_DEADLINE_CEILING_SECONDS,
require_privileged_inputs=True,
)
if __name__ == "__main__":
run(build_verifier())
SignedJsonVerifier is the concrete exact-byte adapter consumed from AID-H-021.002: it reads the manifest, bundle, and fixed /run/trust/behavioral-canary-design.pub key without following links; requires those files and key to be root-owned without group/world write permission; derives signer identity from the key digest; verifies the exact snapshot with Cosign v3.1.2; rejects duplicate keys and non-finite JSON; and returns a local verification receipt. Compiled limits exist only to bound the pre-verification design, bundle, and key reads; build ID, generated populations, and output byte ceilings come from the verified design. The builder walks every fixed output ancestor with no-follow directory descriptors, accepts only a preprovisioned empty private build directory, publishes each file by an exclusive temporary inode plus no-overwrite hard link and exact readback, and publishes the COMMITTED build receipt last. The registry builder never accepts a caller-supplied signature_verified boolean or output path.
Insertion readback and evidence
A format-specific ingestion job accepts a build only when the last-published receipt is present, has commit_state=COMMITTED, binds the signed build_id, and independently rehashes the exact registry and insertion-plan bytes and counts within their signed ceilings. It applies only canary-insertions.json to the exact source dataset digest and emits an authenticated receipt binding the source and output dataset digests, job/code/image digests, every observed (canary_id, repetition_id), row count, and build receipt. A separate read-only identity recomputes that population from the output dataset and scans all training, retrieval, evaluation, and prompt sources for every negative-control ID and phrase; any negative-control presence fails the control. Sign the registry, plan, build receipt, and insertion receipt through AID-M-002.002, store them under a security identity the training job cannot alter, and bind the later training run to the exact output dataset digest. Retention and freshness come from signed legal/experiment policy.
Action: Generate canaries only from the current signed design, reject incomplete or non-exclusive build directories, block training until the independently read output dataset exactly matches the insertion plan and excludes every negative control, and preserve signed provenance for the exact training-run dataset.
AID-DV-007-G002
Run a blinded canary-versus-negative-control audit with a pre-registered statistical test and emit conservative provenance findings bound to the exact model and query profile.
Signed inputs and method boundary
This guidance evaluates the exact canary/control registry produced by the preceding AID-DV-007 guidance. Use AID-H-021.002's maintained exact-byte verifier for the lifecycle scope, registry, query profile, evaluation policy, and observation manifest, and use AID-M-002.002 to sign the result. The query runner and evaluator use separate identities. Observations never supply a cohort or hit verdict: the evaluator derives both from the signed registry and retained response bytes. The scope's inventory_sha256 must equal the exact verified registry digest before any record is evaluated.
The signed query profile binds the exact model/provider identifier, version, endpoint revision, decoding parameters, runner image, and expiry. The signed policy binds the registry and query-profile digests, statistical thresholds, response-size limit, and freshness window. Counts and thresholds come from the versioned experiment policy, not AIDEFEND defaults.
Reusable evaluator module
# File: provenance/evaluate_behavioral_canaries.py
from __future__ import annotations
import errno
import hashlib
import json
import math
import os
import stat
import subprocess
import unicodedata
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal
from scipy.stats import fisher_exact
from statsmodels.stats.proportion import proportion_confint
HEX = set("0123456789abcdef")
ControlStatus = Literal[
"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"
]
FindingStatus = Literal["FINDING", "NO_FINDING"]
class ControlFailure(Exception):
pass
class EvidenceMissing(Exception):
pass
from provenance.build_behavioral_canaries import (
SignedJsonVerifier, VerifiedJson,
)
@dataclass(frozen=True)
class Inputs:
scope: tuple[Path, Path]
registry: tuple[Path, Path]
query_profile: tuple[Path, Path]
policy: tuple[Path, Path]
observations: tuple[Path, Path]
query_receipts: tuple[Path, Path]
response_root: Path
def canonical(value: object) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
).encode("utf-8")
def digest(value: object) -> bool:
return isinstance(value, str) and len(value) == 64 and not (set(value) - HEX)
def exact(value: object, fields: set[str], name: str) -> dict[str, object]:
if not isinstance(value, dict) or set(value) != fields:
raise ControlFailure(name + "_schema_mismatch")
return value
def text(value: object, field: str) -> str:
if not isinstance(value, str) or not value.strip():
raise ControlFailure("invalid_" + field)
return value
def utc(value: object, field: str) -> datetime:
if not isinstance(value, str) or not value.endswith("Z"):
raise ControlFailure("invalid_" + field)
try:
parsed = datetime.fromisoformat(value[:-1] + "+00:00")
except ValueError as exc:
raise ControlFailure("invalid_" + field) from exc
return parsed
def load(
verifier: SignedJsonVerifier,
pair: tuple[Path, Path],
name: str,
bindings: dict[str, dict[str, str]],
) -> VerifiedJson:
try:
verified = verifier.load_exact(
payload_path=pair[0], signature_path=pair[1]
)
except FileNotFoundError as exc:
raise EvidenceMissing(name + "_missing") from exc
except (ValueError, PermissionError, subprocess.CalledProcessError) as exc:
raise ControlFailure(name + "_verification_failed") from exc
if not isinstance(verified, VerifiedJson):
raise ControlFailure(name + "_not_verified")
values = (
verified.payload_sha256,
verified.signature_sha256,
verified.verification_receipt_sha256,
)
if not all(digest(value) for value in values):
raise ControlFailure(name + "_invalid_verification_digest")
signer = text(verified.signer_identity, name + "_signer")
bindings[name] = {
"payload_sha256": verified.payload_sha256,
"signature_sha256": verified.signature_sha256,
"signer_identity": signer,
"verification_receipt_sha256": verified.verification_receipt_sha256,
}
return verified
def read_response(root: Path, relative: str, limit: int) -> bytes:
parts = relative.split("/")
if not root.is_absolute() or "\\" in relative or any(
part in {"", ".", ".."} for part in parts
):
raise ControlFailure("unsafe_response_path")
if not hasattr(os, "O_DIRECTORY") or not hasattr(os, "O_NOFOLLOW"):
raise OSError("secure dirfd traversal unavailable")
directory_flags = (
os.O_RDONLY | os.O_CLOEXEC | os.O_DIRECTORY | os.O_NOFOLLOW
)
opened: list[int] = []
try:
try:
current = os.open(root, directory_flags)
except FileNotFoundError as exc:
raise EvidenceMissing("response_root_missing") from exc
opened.append(current)
for part in parts[:-1]:
try:
current = os.open(part, directory_flags, dir_fd=current)
except FileNotFoundError as exc:
raise EvidenceMissing("response_directory_missing") from exc
except OSError as exc:
if exc.errno in (errno.ELOOP, errno.ENOTDIR):
raise ControlFailure("unsafe_response_directory") from exc
raise
opened.append(current)
try:
descriptor = os.open(
parts[-1],
os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW,
dir_fd=current,
)
except FileNotFoundError as exc:
raise EvidenceMissing("response_file_missing") from exc
except OSError as exc:
if exc.errno in (errno.ELOOP, errno.ENOTDIR):
raise ControlFailure("unsafe_response_file") from exc
raise
opened.append(descriptor)
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode):
raise ControlFailure("response_not_regular")
if before.st_size > limit:
raise ControlFailure("response_too_large")
chunks: list[bytes] = []
total = 0
while True:
chunk = os.read(descriptor, min(65536, limit + 1 - total))
if not chunk:
break
chunks.append(chunk)
total += len(chunk)
if total > limit:
raise ControlFailure("response_too_large")
after = os.fstat(descriptor)
fingerprint = lambda item: (
item.st_dev, item.st_ino, item.st_size,
item.st_mtime_ns, item.st_ctime_ns,
)
if fingerprint(before) != fingerprint(after) or total != before.st_size:
raise ControlFailure("response_changed_during_read")
return b"".join(chunks)
finally:
for descriptor in reversed(opened):
os.close(descriptor)
def outcome(
status: ControlStatus,
reason: str,
bindings: dict[str, dict[str, str]],
*,
finding: FindingStatus | None = None,
**details: object,
) -> dict[str, object]:
return {
"schema_version": "aidefend.behavioral_canary_assessment.v1",
"control_id": "AID-DV-007",
"control_status": status,
"finding_status": finding,
"reason": reason,
"verified_artifacts": bindings,
**details,
}
def evaluate(
verifier: SignedJsonVerifier,
query_receipt_verifier: SignedJsonVerifier,
inputs: Inputs,
*,
now: datetime | None = None,
) -> dict[str, object]:
clock = now or datetime.now(timezone.utc)
if clock.tzinfo is None:
return outcome("ERROR", "naive_evaluation_clock", {})
bindings: dict[str, dict[str, str]] = {}
context: dict[str, object] = {"evaluated_at": clock.isoformat()}
try:
scope_v = load(verifier, inputs.scope, "scope", bindings)
scope = exact(
scope_v.payload,
{
"schema_version", "scope_id", "model_id", "model_version",
"registered_canary_audit", "inventory_sha256", "assessed_at",
"expires_at",
},
"scope",
)
if scope["schema_version"] != "aidefend.behavioral_canary_scope.v1":
raise ControlFailure("unsupported_scope_schema")
for field in ("scope_id", "model_id", "model_version"):
text(scope[field], field)
if type(scope["registered_canary_audit"]) is not bool or not digest(
scope["inventory_sha256"]
):
raise ControlFailure("invalid_scope_inventory")
if utc(scope["assessed_at"], "scope_assessed_at") > clock:
raise ControlFailure("future_scope_assessment")
if utc(scope["expires_at"], "scope_expires_at") <= clock:
raise EvidenceMissing("scope_expired")
context.update({
"scope_id": scope["scope_id"],
"model_id": scope["model_id"],
"model_version": scope["model_version"],
})
if not scope["registered_canary_audit"]:
return outcome(
"NOT_APPLICABLE",
"signed_scope_has_no_registered_canary_audit",
bindings,
**context,
)
registry_v = load(verifier, inputs.registry, "registry", bindings)
if scope["inventory_sha256"] != registry_v.payload_sha256:
raise ControlFailure("scope_registry_inventory_mismatch")
registry = exact(
registry_v.payload,
{
"schema_version", "design_sha256", "policy_version",
"authorization_id", "authorization_receipt_sha256",
"dataset_id", "source_dataset_sha256", "records",
},
"registry",
)
if registry["schema_version"] != "aidefend.behavioral_canary_registry.v1":
raise ControlFailure("unsupported_registry_schema")
for field in (
"design_sha256", "authorization_receipt_sha256",
"source_dataset_sha256",
):
if not digest(registry[field]):
raise ControlFailure("invalid_" + field)
for field in ("policy_version", "authorization_id", "dataset_id"):
text(registry[field], field)
rows = registry["records"]
if not isinstance(rows, list) or not rows:
raise EvidenceMissing("registry_population_missing")
registered: dict[str, dict[str, object]] = {}
phrases: set[str] = set()
for raw in rows:
row = exact(
raw,
{"canary_id", "cohort", "prompt", "expected_phrase", "repetitions"},
"registry_record",
)
canary_id = text(row["canary_id"], "canary_id")
cohort = text(row["cohort"], "cohort")
phrase = text(row["expected_phrase"], "expected_phrase")
text(row["prompt"], "prompt")
repetitions = row["repetitions"]
if cohort not in {"inserted", "negative_control"}:
raise ControlFailure("invalid_cohort")
if type(repetitions) is not int or (
cohort == "inserted" and repetitions < 1
) or (cohort == "negative_control" and repetitions != 0):
raise ControlFailure("invalid_cohort_repetitions")
normalized = unicodedata.normalize("NFKC", phrase).casefold()
if canary_id in registered or normalized in phrases:
raise ControlFailure("duplicate_registry_identity")
registered[canary_id] = row
phrases.add(normalized)
profile_v = load(verifier, inputs.query_profile, "query_profile", bindings)
profile = exact(
profile_v.payload,
{
"schema_version", "profile_version", "model_id", "model_version",
"endpoint_revision", "decoding_parameters",
"query_runner_image_sha256", "expires_at",
},
"query_profile",
)
if profile["schema_version"] != "aidefend.behavioral_canary_query_profile.v1":
raise ControlFailure("unsupported_query_profile_schema")
text(profile["profile_version"], "profile_version")
if profile["model_id"] != scope["model_id"] or (
profile["model_version"] != scope["model_version"]
):
raise ControlFailure("query_profile_scope_mismatch")
text(profile["endpoint_revision"], "endpoint_revision")
if not isinstance(profile["decoding_parameters"], dict) or not digest(
profile["query_runner_image_sha256"]
):
raise ControlFailure("invalid_query_profile")
if utc(profile["expires_at"], "query_profile_expires_at") <= clock:
raise EvidenceMissing("query_profile_expired")
policy_v = load(verifier, inputs.policy, "policy", bindings)
policy = exact(
policy_v.payload,
{
"schema_version", "policy_version", "registry_sha256",
"query_profile_sha256", "minimum_trials_per_cohort", "alpha",
"minimum_effect_delta", "minimum_distinct_inserted_hits",
"maximum_response_bytes", "maximum_evidence_age_seconds",
"expires_at",
},
"policy",
)
if policy["schema_version"] != "aidefend.behavioral_canary_policy.v1":
raise ControlFailure("unsupported_policy_schema")
text(policy["policy_version"], "policy_version")
if policy["registry_sha256"] != registry_v.payload_sha256 or (
policy["query_profile_sha256"] != profile_v.payload_sha256
):
raise ControlFailure("policy_binding_mismatch")
for field in (
"minimum_trials_per_cohort", "minimum_distinct_inserted_hits",
"maximum_response_bytes", "maximum_evidence_age_seconds",
):
if type(policy[field]) is not int or policy[field] < 1:
raise ControlFailure("invalid_" + field)
alpha = policy["alpha"]
delta_threshold = policy["minimum_effect_delta"]
if type(alpha) not in (int, float) or not math.isfinite(float(alpha)) or (
not 0.0 < float(alpha) < 0.5
):
raise ControlFailure("invalid_alpha")
if type(delta_threshold) not in (int, float) or not math.isfinite(
float(delta_threshold)
) or not 0.0 <= float(delta_threshold) <= 1.0:
raise ControlFailure("invalid_effect_delta")
if utc(policy["expires_at"], "policy_expires_at") <= clock:
raise EvidenceMissing("policy_expired")
inserted = sorted(
key for key, row in registered.items() if row["cohort"] == "inserted"
)
controls = sorted(
key
for key, row in registered.items()
if row["cohort"] == "negative_control"
)
minimum = policy["minimum_trials_per_cohort"]
if min(len(inserted), len(controls)) < minimum:
raise EvidenceMissing("preregistered_power_population_not_met")
if policy["minimum_distinct_inserted_hits"] > len(inserted):
raise ControlFailure("unachievable_hit_threshold")
manifest_v = load(verifier, inputs.observations, "observations", bindings)
manifest = exact(
manifest_v.payload,
{
"schema_version", "registry_sha256", "query_profile_sha256",
"model_id", "model_version", "endpoint_revision",
"captured_at", "records",
},
"observation_manifest",
)
if manifest["schema_version"] != "aidefend.behavioral_canary_observations.v1":
raise ControlFailure("unsupported_observation_schema")
if manifest["registry_sha256"] != registry_v.payload_sha256 or (
manifest["query_profile_sha256"] != profile_v.payload_sha256
):
raise ControlFailure("observation_binding_mismatch")
for field in ("model_id", "model_version", "endpoint_revision"):
if manifest[field] != profile[field]:
raise ControlFailure("observation_" + field + "_mismatch")
captured = utc(manifest["captured_at"], "captured_at")
if captured > clock:
raise ControlFailure("future_observation")
if (clock - captured).total_seconds() > policy[
"maximum_evidence_age_seconds"
]:
raise EvidenceMissing("observation_evidence_stale")
observation_rows = manifest["records"]
if not isinstance(observation_rows, list):
raise ControlFailure("observation_records_not_list")
observations: dict[str, dict[str, object]] = {}
for raw in observation_rows:
row = exact(
raw,
{
"canary_id", "request_sha256", "response_relative_path",
"response_sha256", "query_receipt_sha256",
},
"observation",
)
canary_id = text(row["canary_id"], "canary_id")
text(row["response_relative_path"], "response_relative_path")
if canary_id in observations:
raise ControlFailure("duplicate_observation")
for field in (
"request_sha256", "response_sha256", "query_receipt_sha256"
):
if not digest(row[field]):
raise ControlFailure("invalid_" + field)
observations[canary_id] = row
extra = set(observations) - set(registered)
missing = set(registered) - set(observations)
if extra:
raise ControlFailure("unregistered_observation")
if missing:
raise EvidenceMissing("registered_observation_missing")
receipts_v = load(
query_receipt_verifier, inputs.query_receipts,
"query_receipts", bindings,
)
if bindings["query_receipts"]["signer_identity"] == bindings["observations"]["signer_identity"]:
raise ControlFailure("query_receipt_signer_not_independent")
receipt_manifest = exact(
receipts_v.payload,
{
"schema_version", "registry_sha256", "query_profile_sha256",
"model_id", "model_version", "endpoint_revision",
"captured_at", "records",
},
"query_receipt_manifest",
)
if receipt_manifest["schema_version"] != "aidefend.behavioral_canary_query_receipts.v1":
raise ControlFailure("unsupported_query_receipt_schema")
if (receipt_manifest["registry_sha256"] != registry_v.payload_sha256
or receipt_manifest["query_profile_sha256"] != profile_v.payload_sha256):
raise ControlFailure("query_receipt_binding_mismatch")
for field in ("model_id", "model_version", "endpoint_revision"):
if receipt_manifest[field] != profile[field]:
raise ControlFailure("query_receipt_" + field + "_mismatch")
receipts_captured = utc(receipt_manifest["captured_at"], "query_receipts_captured_at")
if receipts_captured > clock:
raise ControlFailure("future_query_receipt_manifest")
if (clock - receipts_captured).total_seconds() > policy["maximum_evidence_age_seconds"]:
raise EvidenceMissing("query_receipt_manifest_stale")
receipt_rows = receipt_manifest["records"]
if not isinstance(receipt_rows, list):
raise ControlFailure("query_receipt_records_not_list")
query_receipts: dict[str, dict[str, object]] = {}
query_receipt_digests: dict[str, str] = {}
for raw in receipt_rows:
row = exact(
raw,
{
"canary_id", "request_sha256", "response_sha256",
"query_runner_image_sha256", "endpoint_revision",
"executed_at",
},
"query_receipt",
)
canary_id = text(row["canary_id"], "query_receipt_canary_id")
if canary_id in query_receipts:
raise ControlFailure("duplicate_query_receipt")
for field in ("request_sha256", "response_sha256", "query_runner_image_sha256"):
if not digest(row[field]):
raise ControlFailure("invalid_query_receipt_" + field)
if (row["query_runner_image_sha256"] != profile["query_runner_image_sha256"]
or row["endpoint_revision"] != profile["endpoint_revision"]):
raise ControlFailure("query_receipt_execution_profile_mismatch")
executed = utc(row["executed_at"], "query_receipt_executed_at")
if executed > receipts_captured:
raise ControlFailure("query_receipt_after_manifest_capture")
if executed > clock:
raise ControlFailure("future_query_receipt")
if (clock - executed).total_seconds() > policy["maximum_evidence_age_seconds"]:
raise EvidenceMissing("query_receipt_stale")
query_receipts[canary_id] = row
query_receipt_digests[canary_id] = hashlib.sha256(canonical(row)).hexdigest()
if set(query_receipts) - set(registered):
raise ControlFailure("unregistered_query_receipt")
if set(registered) - set(query_receipts):
raise EvidenceMissing("registered_query_receipt_missing")
hits: dict[str, bool] = {}
response_digests: list[str] = []
for canary_id in sorted(registered):
registered_row = registered[canary_id]
observed_row = observations[canary_id]
request = {
"schema_version": "aidefend.behavioral_canary_request.v1",
"canary_id": canary_id,
"prompt": registered_row["prompt"],
"model_id": profile["model_id"],
"model_version": profile["model_version"],
"endpoint_revision": profile["endpoint_revision"],
"decoding_parameters": profile["decoding_parameters"],
"query_profile_sha256": profile_v.payload_sha256,
}
request_sha256 = hashlib.sha256(canonical(request)).hexdigest()
receipt_row = query_receipts[canary_id]
if request_sha256 != observed_row["request_sha256"]:
raise ControlFailure("request_digest_mismatch")
if receipt_row["request_sha256"] != request_sha256:
raise ControlFailure("query_receipt_request_mismatch")
if observed_row["query_receipt_sha256"] != query_receipt_digests[canary_id]:
raise ControlFailure("query_receipt_digest_mismatch")
response = read_response(
inputs.response_root,
str(observed_row["response_relative_path"]),
int(policy["maximum_response_bytes"]),
)
response_sha256 = hashlib.sha256(response).hexdigest()
if response_sha256 != observed_row["response_sha256"]:
raise ControlFailure("response_digest_mismatch")
if receipt_row["response_sha256"] != response_sha256:
raise ControlFailure("query_receipt_response_mismatch")
try:
answer = response.decode("utf-8", errors="strict")
except UnicodeDecodeError as exc:
raise ControlFailure("response_not_utf8") from exc
expected = unicodedata.normalize(
"NFKC", str(registered_row["expected_phrase"])
).casefold()
hits[canary_id] = expected in unicodedata.normalize(
"NFKC", answer
).casefold()
response_digests.append(response_sha256)
inserted_hits = sum(hits[key] for key in inserted)
control_hits = sum(hits[key] for key in controls)
_, p_value = fisher_exact(
[
[inserted_hits, len(inserted) - inserted_hits],
[control_hits, len(controls) - control_hits],
],
alternative="greater",
)
inserted_rate = inserted_hits / len(inserted)
control_rate = control_hits / len(controls)
effect_delta = inserted_rate - control_rate
finding = (
inserted_hits >= policy["minimum_distinct_inserted_hits"]
and float(p_value) <= float(alpha)
and effect_delta >= float(delta_threshold)
)
inserted_ci = proportion_confint(
inserted_hits, len(inserted), alpha=float(alpha), method="wilson"
)
control_ci = proportion_confint(
control_hits, len(controls), alpha=float(alpha), method="wilson"
)
return outcome(
"PASS",
"exact_population_evaluated",
bindings,
finding="FINDING" if finding else "NO_FINDING",
**context,
endpoint_revision=profile["endpoint_revision"],
registry_sha256=registry_v.payload_sha256,
query_profile_sha256=profile_v.payload_sha256,
policy_sha256=policy_v.payload_sha256,
observation_manifest_sha256=manifest_v.payload_sha256,
query_receipt_manifest_sha256=receipts_v.payload_sha256,
population_sha256=hashlib.sha256(
canonical(sorted(registered))
).hexdigest(),
confidence_level=round(1.0 - float(alpha), 6),
inserted_hits=inserted_hits,
inserted_trials=len(inserted),
inserted_wilson=[round(float(value), 6) for value in inserted_ci],
control_hits=control_hits,
control_trials=len(controls),
control_wilson=[round(float(value), 6) for value in control_ci],
effect_delta=round(effect_delta, 6),
p_value=round(float(p_value), 10),
positive_inserted_ids=[key for key in inserted if hits[key]],
positive_control_ids=[key for key in controls if hits[key]],
response_sha256s=sorted(response_digests),
query_receipt_sha256s=sorted(query_receipt_digests.values()),
)
except EvidenceMissing as exc:
return outcome("INSUFFICIENT_DATA", str(exc), bindings, **context)
except ControlFailure as exc:
return outcome("FAIL", str(exc), bindings, **context)
except Exception as exc:
return outcome(
"ERROR", "evaluator_runtime_error", bindings,
error_class=type(exc).__name__, **context
)
SignedJsonVerifier is the concrete exact-byte adapter from the preceding guidance and AID-H-021.002: it snapshots bounded regular files without following links, rejects duplicate keys and non-finite JSON, verifies the exact bytes, and binds the approved signer identity to the trusted-key digest. The signed scope, profile, and policy schemas enforce freshness and expiry. This evaluator maps invalid signatures or signers to ControlFailure, absent artifacts to EvidenceMissing, and verifier outages to operational exceptions. The evaluator rejects missing, extra, duplicate, oversized, changed, non-regular, path-escaped, request-mismatched, or digest-mismatched responses.
Independent replay
An authenticated query runner that cannot read expected phrases records the exact request digest, retained UTF-8 response digest, and independently issued query receipt for every registered ID. A separate read-only evaluator runs this module. A third identity re-verifies every signature and query receipt, reconstructs every request, reads every retained response, recomputes the NFKC/case-folded exact-phrase matches and the one-sided Fisher test, and compares the complete result. Sign the result through AID-M-002.002 and retain all inputs, raw responses, receipts, evaluator image digest, result, and replay receipt in immutable/WORM evidence outside the trainer, runner, and evaluator identities.
Action: Never accept caller-supplied hit or cohort fields. A valid positive result is control_status=PASS with finding_status=FINDING; preserve the evidence and require independent lineage, access, contractual, or leak evidence before a provenance claim.
Supply the aggregate query-receipt manifest through a verifier whose signer identity differs from the observation signer. The evaluator verifies the exact receipt population and freshness, recomputes each receipt-record digest, and binds request, response, model/version, endpoint revision, and query-runner image before the statistical result is eligible.
Open-source tools
- DVC
- MLflow
- pandas
- SciPy
- statsmodels
Commercial / hosted tools
- Weights & Biases
- Managed MLflow on Databricks
Threat mappings
MITRE ATLAS
- AML.T0057 LLM Data Leakage (preregistered canary recall measures memorized training-data disclosure)
MAESTRO
- N/A
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure (powered canary-recall tests detect memorized training records being reproduced)
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- NISTAML.038 Data Extraction (registered canary detection measures extraction of memorized training data)
Cisco Integrated AI Security and Safety Framework
- AISubtech-10.1.3 Sensitive Data Reconstruction (registered synthetic training-canary recall audit only)
- AISubtech-8.2.1 Training Data Exposure (registered synthetic training-canary recall audit only)
- AITech-8.2 Data Exfiltration / Exposure (registered synthetic training-canary recall audit only)
- AITech-8.3 Information Disclosure (registered synthetic training-canary recall audit only)
Google Secure AI Framework 2.0 - Risks
- UTD: Unauthorized Training Data (registered behavioral canaries reveal unauthorized reuse in training)
- SDD: Sensitive Data Disclosure (canary recall measures whether training records are exposed through model responses)
Databricks AI Security Framework 3.0
- N/A