AID-D-014.003: Retrieval Source Provenance, Diversity & Coverage Monitoring
Compare the ordered top-k retrieval population with an authenticated gateway completeness record and versioned source registry. Detect missing or unregistered provenance, minimum-k shortfall, collection-coverage gaps, duplicate or reordered chunks, and calibrated source-diversity or concentration violations. Calibrate thresholds per query class, route, and index on a versioned benign and adversarial corpus; the framework publishes no universal concentration percentage. Missing provenance, unsupported source identity, incomplete population, and unreadable or unverifiable policy must remain explicit and must not be reported as clean retrieval. This detector emits findings; it does not suppress or re-rank sources.
Pillars: data
Lifecycle: operation
Scope boundary
Owns ordered top-k retrieval provenance, registry membership, minimum-count coverage, collection coverage, duplicate and reorder detection, and calibrated source-diversity or concentration findings. It does not suppress, re-rank, sign, or admit sources.
AID-D-014.002: AID-D-014.003 verifies source and population properties of the retrieval set; AID-D-014.002 measures semantic relevance of query-to-chunk pairs. Provenance and coverage do not establish relevance, and relevance does not establish authorized source identity or complete retrieval.
AID-H-020: AID-D-014.003 emits provenance, diversity, and coverage findings; AID-H-020 enforces signed chunk integrity, source authorization, and retrieval-time hygiene. The detector observes and reports the retrieval population, while Harden decides whether it may enter context.
Implementation guidance
- AID-D-014.003-G001: Verify ordered top-k source provenance and minimum-k coverage, then evaluate calibrated source-diversity and concentration policy without treating incomplete evidence as clean.
AID-D-014.003-G001
Verify ordered top-k source provenance and minimum-k coverage, then evaluate calibrated source-diversity and concentration policy without treating incomplete evidence as clean.
Pin the runtime inputs
Bind every retrieved chunk to an independently signed source registry containing exact allowed source versions. Install Cosign at /usr/local/bin/cosign and keep the bootstrap public key at /etc/aidefend/d014/trust/runtime-profile-root.pub. The key, profile, bundle, and every parent directory must be root-owned and non-writable by group or other users. The deployment-only provisioner below computes lowercase SHA-256 values from the exact deployed role-key bytes and refuses to overwrite an activated profile. Supply its version and validity interval through controlled release automation, then sign the exact output with the offline bootstrap role. An evaluator caller cannot select keys, paths, evaluation time, limits, or the result directory.
from __future__ import annotations
# File: detection/provision_d014003_g001_provenance_profile.py
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import re
import stat
import sys
from detection.d014_verified_artifacts import (
BOOTSTRAP_DIRECTORY, HARD_KEY_CAPTURE_CEILING, read_stable, write_exclusive,
)
CONTROL = "AID-D-014.003-G001"
PROFILE_NAME = "d014003-g001-provenance"
ARTIFACTS = {
"snapshot": {
"payload": "/var/lib/aidefend/d014/input/d014003-g001-provenance-snapshot.jsonl",
"bundle": "/var/lib/aidefend/d014/input/d014003-g001-provenance-snapshot.jsonl.sigstore.json"
},
"receipt": {
"payload": "/var/lib/aidefend/d014/input/d014003-g001-provenance-receipt.json",
"bundle": "/var/lib/aidefend/d014/input/d014003-g001-provenance-receipt.json.sigstore.json"
},
"policy": {
"payload": "/etc/aidefend/d014/input/d014003-g001-provenance-policy.json",
"bundle": "/etc/aidefend/d014/input/d014003-g001-provenance-policy.json.sigstore.json"
},
"registry": {
"payload": "/etc/aidefend/d014/input/d014003-g001-provenance-registry.json",
"bundle": "/etc/aidefend/d014/input/d014003-g001-provenance-registry.json.sigstore.json"
}
}
KEY_PATHS = {
"snapshot": "/etc/aidefend/d014/trust/d014003-g001-provenance-snapshot.pub",
"receipt": "/etc/aidefend/d014/trust/d014003-g001-provenance-receipt.pub",
"policy": "/etc/aidefend/d014/trust/d014003-g001-provenance-policy.pub",
"registry": "/etc/aidefend/d014/trust/d014003-g001-provenance-registry.pub",
"result": "/etc/aidefend/d014/trust/d014003-g001-provenance-result.pub"
}
RESULT_BASE = "/var/lib/aidefend/d014/results/d014003-g001-provenance"
PROFILE_VERSION = re.compile(r"^[A-Za-z0-9._-]{1,128}$")
def parse_time(value: str) -> datetime:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("profile validity timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
def utc_text(value: datetime) -> str:
return value.isoformat(timespec="seconds").replace("+00:00", "Z")
def main() -> int:
if len(sys.argv) != 4 or not PROFILE_VERSION.fullmatch(sys.argv[1]):
raise SystemExit(
"usage: provision_profile PROFILE_VERSION VALID_FROM VALID_UNTIL"
)
valid_from = parse_time(sys.argv[2])
valid_until = parse_time(sys.argv[3])
if valid_from >= valid_until:
raise ValueError("profile validity interval is empty or reversed")
directory = BOOTSTRAP_DIRECTORY.lstat()
if (
not stat.S_ISDIR(directory.st_mode)
or BOOTSTRAP_DIRECTORY.is_symlink()
or directory.st_uid != 0
or directory.st_mode & 0o022
):
raise PermissionError("profile directory is not root-owned and non-writable")
keys = {}
for role, value in KEY_PATHS.items():
raw = read_stable(
Path(value), maximum_bytes=HARD_KEY_CAPTURE_CEILING, trust_root=True,
)
keys[role] = {"path": value, "sha256": hashlib.sha256(raw).hexdigest()}
if len({spec["sha256"] for spec in keys.values()}) != len(keys):
raise ValueError("verification roles do not use distinct public keys")
profile = {
"schema_version": "aidefend.d014-runtime-profile.v1",
"profile_version": sys.argv[1],
"control": CONTROL,
"valid_from": utc_text(valid_from),
"valid_until": utc_text(valid_until),
"artifacts": ARTIFACTS,
"verification_keys": keys,
"result_base": RESULT_BASE,
}
output = BOOTSTRAP_DIRECTORY / (PROFILE_NAME + ".json")
raw = json.dumps(
profile, sort_keys=True, indent=2, allow_nan=False,
).encode("utf-8") + b"\n"
write_exclusive(output, raw, mode=0o640)
print(output)
return 0
if __name__ == "__main__":
raise SystemExit(main())Each provenance line contains query ID/class, rank, chunk ID, source ID, and source version. The receipt binds the exact top-k snapshot, counts, asset, registry digest/version, completion time, and completeness. The registry contains sorted unique source IDs and sorted allowed versions. The policy binds that registry plus expected top-k, maximum concentration, minimum distinct sources, population limits, and freshness by query class.
Capture and verify exact bytes
The shared module below opens every input with no-follow and stable-descriptor checks, copies payload, bundle, and pinned key bytes into a private directory, and invokes cosign verify-blob --bundle only on those copies. Its fixed hard ceilings protect the evaluator process; method thresholds, freshness windows, and population limits remain signed policy values. The signed model path uses a deterministic uncompressed tar, two streaming inspection passes, regular files only, and safetensors; it never calls extractall or enables remote model code.
from __future__ import annotations
# File: detection/d014_verified_artifacts.py
from contextlib import contextmanager
from dataclasses import dataclass
import ctypes
from datetime import datetime, timezone
import errno
import hashlib
import hmac
from io import BytesIO
import json
import os
from pathlib import Path, PurePosixPath
import re
import shutil
import stat
import subprocess
import tarfile
import tempfile
COSIGN = "/usr/local/bin/cosign"
BOOTSTRAP_DIRECTORY = Path("/etc/aidefend/d014/profiles")
BOOTSTRAP_KEY = Path("/etc/aidefend/d014/trust/runtime-profile-root.pub")
# These are non-policy process-safety ceilings. Signed policies and callers
# may choose lower limits but cannot raise these values.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
HARD_ARTIFACT_CAPTURE_CEILING = 256 * 1024 * 1024
HARD_METADATA_CAPTURE_CEILING = 8 * 1024 * 1024
HARD_KEY_CAPTURE_CEILING = 1024 * 1024
HARD_MODEL_MEMBER_COUNT = 4096
HARD_MODEL_MEMBER_BYTES = HARD_ARTIFACT_CAPTURE_CEILING
HARD_MODEL_TOTAL_BYTES = HARD_ARTIFACT_CAPTURE_CEILING
HARD_MODEL_NAME_CAPTURE_CEILING = 512
SHA256 = re.compile(r"^[0-9a-f]{64}$")
PROFILE_FIELDS = {
"schema_version", "profile_version", "control", "artifacts",
"verification_keys", "result_base", "valid_from", "valid_until",
}
ARTIFACT_FIELDS = {"payload", "bundle"}
KEY_FIELDS = {"path", "sha256"}
AT_FDCWD = -100
RENAME_NOREPLACE = 1
@dataclass(frozen=True)
class Captured:
raw: bytes
bundle: bytes
key: bytes
def evidence(self) -> dict:
return {
"payload_sha256": hashlib.sha256(self.raw).hexdigest(),
"bundle_sha256": hashlib.sha256(self.bundle).hexdigest(),
"verification_key_sha256": hashlib.sha256(self.key).hexdigest(),
}
def reject_constant(value: str):
raise ValueError("non-finite JSON value is forbidden: " + value)
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError("duplicate JSON key: " + key)
result[key] = value
return result
def strict_json_object(raw: bytes, label: str) -> dict:
value = json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=reject_constant,
)
if not isinstance(value, dict):
raise ValueError(label + " must be a JSON object")
return value
def canonical(value: object) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
def read_stable(
path: Path, *, maximum_bytes: int, allow_empty: bool = False,
trust_root: bool = False,
) -> bytes:
if trust_root:
current = Path(path.anchor)
for part in path.parts[1:-1]:
current /= part
metadata = current.lstat()
if (
not stat.S_ISDIR(metadata.st_mode)
or current.is_symlink()
or metadata.st_uid != 0
or metadata.st_mode & 0o022
):
raise PermissionError(
"trusted directory is not root-owned and non-writable: "
+ str(current)
)
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or before.st_size < 0
or before.st_size > maximum_bytes
or (before.st_size == 0 and not allow_empty)
):
raise ValueError("input is not a bounded regular file: " + str(path))
if trust_root and (before.st_uid != 0 or before.st_mode & 0o022):
raise PermissionError(
"trusted file must be root-owned and non-writable: " + str(path)
)
chunks = []
total = 0
while True:
block = os.read(descriptor, min(1024 * 1024, maximum_bytes + 1 - total))
if not block:
break
chunks.append(block)
total += len(block)
if total > maximum_bytes:
raise ValueError("input exceeds its hard byte ceiling: " + str(path))
after = os.fstat(descriptor)
finally:
os.close(descriptor)
identity = lambda item: (
item.st_dev, item.st_ino, item.st_size,
item.st_mtime_ns, item.st_ctime_ns,
)
raw = b"".join(chunks)
if identity(before) != identity(after) or len(raw) != before.st_size:
raise RuntimeError("input changed during capture: " + str(path))
return raw
def write_exclusive(path: Path, raw: bytes, mode: int = 0o600) -> None:
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0),
mode,
)
with os.fdopen(descriptor, "wb") as stream:
stream.write(raw)
stream.flush()
os.fsync(stream.fileno())
def verify_bytes(label: str, raw: bytes, bundle: bytes, key: bytes) -> None:
with tempfile.TemporaryDirectory(prefix="aidefend-d014-verify-") as name:
root = Path(name)
os.chmod(root, 0o700)
payload = root / "payload"
signature = root / "payload.sigstore.json"
verify_key = root / "verify.pub"
for path, value in ((payload, raw), (signature, bundle), (verify_key, key)):
write_exclusive(path, value)
subprocess.run(
[
COSIGN, "verify-blob", "--key", str(verify_key),
"--bundle", str(signature), str(payload),
],
check=True, capture_output=True, text=True,
timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
for path, expected in ((payload, raw), (signature, bundle), (verify_key, key)):
actual = read_stable(
path, maximum_bytes=max(len(expected), 1),
allow_empty=(path == payload and not expected),
)
if not hmac.compare_digest(actual, expected):
raise RuntimeError(label + " private verification snapshot changed")
def capture_explicit(
label: str, payload: Path, bundle: Path, key_path: Path,
*, maximum_bytes: int, allow_empty: bool = False,
trust_payload_paths: bool = False,
) -> Captured:
raw = read_stable(
payload, maximum_bytes=maximum_bytes, allow_empty=allow_empty,
trust_root=trust_payload_paths,
)
bundle_raw = read_stable(
bundle, maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
trust_root=trust_payload_paths,
)
key = read_stable(
key_path, maximum_bytes=HARD_KEY_CAPTURE_CEILING, trust_root=True
)
verify_bytes(label, raw, bundle_raw, key)
return Captured(raw, bundle_raw, key)
def canonical_absolute(value: object, label: str, roots: tuple[str, ...]) -> Path:
if not isinstance(value, str) or not value or "\\" in value:
raise ValueError(label + " is not a canonical absolute POSIX path")
pure = PurePosixPath(value)
if not pure.is_absolute() or ".." in pure.parts or str(pure) != value:
raise ValueError(label + " is not a canonical absolute POSIX path")
if not any(value.startswith(root + "/") for root in roots):
raise ValueError(label + " is outside its fixed deployment roots")
return Path(value)
def load_profile(
expected_control: str, artifact_roles: tuple[str, ...], *,
profile_name: str, evaluated_at: datetime,
) -> tuple[dict, Captured]:
if not re.fullmatch(r"[a-z0-9-]+", profile_name):
raise ValueError("fixed runtime-profile name is invalid")
profile_path = BOOTSTRAP_DIRECTORY / (profile_name + ".json")
profile_bundle = BOOTSTRAP_DIRECTORY / (profile_name + ".sigstore.json")
bootstrap = capture_explicit(
"D014 runtime profile", profile_path, profile_bundle,
BOOTSTRAP_KEY, maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
trust_payload_paths=True,
)
profile = strict_json_object(bootstrap.raw, "D014 runtime profile")
roles = set(artifact_roles)
if (
set(profile) != PROFILE_FIELDS
or profile.get("schema_version") != "aidefend.d014-runtime-profile.v1"
or profile.get("control") != expected_control
or not isinstance(profile.get("profile_version"), str)
or not profile["profile_version"]
or not isinstance(profile.get("artifacts"), dict)
or set(profile["artifacts"]) != roles
or not isinstance(profile.get("verification_keys"), dict)
or set(profile["verification_keys"]) != roles | {"result"}
):
raise ValueError("signed D014 runtime profile schema or scope differs")
if evaluated_at.utcoffset() is None:
raise ValueError("runtime-profile evaluation time lacks timezone")
valid_from = datetime.fromisoformat(
str(profile["valid_from"]).replace("Z", "+00:00")
)
valid_until = datetime.fromisoformat(
str(profile["valid_until"]).replace("Z", "+00:00")
)
if (
valid_from.utcoffset() is None or valid_until.utcoffset() is None
or valid_from.astimezone(timezone.utc) >= valid_until.astimezone(timezone.utc)
or not (
valid_from.astimezone(timezone.utc)
<= evaluated_at.astimezone(timezone.utc)
< valid_until.astimezone(timezone.utc)
)
):
raise ValueError("signed D014 runtime profile is outside its validity interval")
for role, artifact in profile["artifacts"].items():
if not isinstance(artifact, dict) or set(artifact) != ARTIFACT_FIELDS:
raise ValueError("artifact path contract differs for role " + role)
for field in ARTIFACT_FIELDS:
canonical_absolute(
artifact[field], role + "." + field,
("/etc/aidefend/d014", "/var/lib/aidefend/d014"),
)
key_digests = []
for role, spec in profile["verification_keys"].items():
if (
not isinstance(spec, dict)
or set(spec) != KEY_FIELDS
or not SHA256.fullmatch(str(spec.get("sha256")))
):
raise ValueError("verification-key contract differs for role " + role)
canonical_absolute(
spec["path"], role + ".key",
("/etc/aidefend/d014/trust",),
)
key_digests.append(spec["sha256"])
bootstrap_key_sha = bootstrap.evidence()["verification_key_sha256"]
if len(key_digests) != len(set(key_digests)) or bootstrap_key_sha in key_digests:
raise ValueError("D014 trust roles are not key-separated")
canonical_absolute(
profile["result_base"], "result_base",
("/var/lib/aidefend/d014/results",),
)
return profile, bootstrap
def capture_role(
profile: dict, role: str, *, maximum_bytes: int,
allow_empty: bool = False,
) -> Captured:
artifact = profile["artifacts"][role]
key_spec = profile["verification_keys"][role]
captured = capture_explicit(
role,
Path(artifact["payload"]), Path(artifact["bundle"]),
Path(key_spec["path"]), maximum_bytes=maximum_bytes,
allow_empty=allow_empty,
)
if not hmac.compare_digest(
captured.evidence()["verification_key_sha256"], key_spec["sha256"]
):
raise ValueError("captured verification key differs for role " + role)
return captured
def safe_member_name(value: str) -> PurePosixPath:
if (
not value or "\\" in value
or len(value.encode("utf-8")) > HARD_MODEL_NAME_CAPTURE_CEILING
):
raise ValueError("model archive member name is invalid")
path = PurePosixPath(value)
if path.is_absolute() or ".." in path.parts or str(path) != value:
raise ValueError("model archive member escapes its extraction root")
return path
@contextmanager
def safe_model_root(model_tar_raw: bytes):
"""Preinspect and manually copy an uncompressed, signed model tar."""
with tempfile.TemporaryDirectory(prefix="aidefend-d014-model-") as name:
root = Path(name)
os.chmod(root, 0o700)
manifest = []
with tarfile.open(fileobj=BytesIO(model_tar_raw), mode="r|") as archive:
seen = set()
regular = set()
total = 0
for member in archive:
if len(manifest) >= HARD_MODEL_MEMBER_COUNT:
raise ValueError("model archive member count exceeds its hard ceiling")
path = safe_member_name(member.name)
folded = str(path).casefold()
if folded in seen:
raise ValueError("duplicate model archive member")
seen.add(folded)
if not (member.isdir() or member.isreg()) or member.issym() or member.islnk():
raise ValueError("links and special model archive members are forbidden")
if member.isreg() and member.mode & 0o111:
raise ValueError("executable model archive member is forbidden")
if member.isreg():
if member.size < 0 or member.size > HARD_MODEL_MEMBER_BYTES:
raise ValueError("model member exceeds its hard byte ceiling")
total += member.size
if total > HARD_MODEL_TOTAL_BYTES:
raise ValueError("model archive exceeds its hard total-byte ceiling")
suffix = path.suffix.lower()
if suffix not in {".json", ".safetensors", ".txt", ".model", ".vocab", ".merges"}:
raise ValueError("unsafe or unsupported model member type: " + suffix)
regular.add(str(path))
manifest.append((
member.name, member.size, member.mode,
member.isdir(), member.isreg(),
))
if not manifest:
raise ValueError("model archive is empty")
for filename in regular:
parts = PurePosixPath(filename).parts
for index in range(1, len(parts)):
if str(PurePosixPath(*parts[:index])) in regular:
raise ValueError("model archive file shadows a parent directory")
if "config.json" not in regular:
raise ValueError("model archive lacks config.json")
if not any(name.endswith(".safetensors") for name in regular):
raise ValueError("model archive lacks safetensors weights")
tokenizer_names = {
"tokenizer.json", "tokenizer.model", "vocab.json", "vocab.txt",
"sentencepiece.bpe.model", "merges.txt",
}
if not regular.intersection(tokenizer_names):
raise ValueError("model archive lacks a supported tokenizer asset")
with tarfile.open(fileobj=BytesIO(model_tar_raw), mode="r|") as archive:
position = -1
for position, member in enumerate(archive):
observed = (
member.name, member.size, member.mode,
member.isdir(), member.isreg(),
)
if position >= len(manifest) or observed != manifest[position]:
raise RuntimeError("model archive changed between inspection passes")
relative = safe_member_name(member.name)
target = root.joinpath(*relative.parts)
if member.isdir():
target.mkdir(parents=True, exist_ok=True, mode=0o700)
continue
target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
stream = archive.extractfile(member)
if stream is None:
raise ValueError("regular model member cannot be read")
raw = stream.read(member.size + 1)
if len(raw) != member.size:
raise ValueError("model member size differs during extraction")
write_exclusive(target, raw, mode=0o400)
if position + 1 != len(manifest):
raise RuntimeError("model archive member count changed between passes")
yield root
def fsync_directory(path: Path) -> None:
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def fsync_regular_file(path: Path) -> None:
descriptor = os.open(
path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_CLOEXEC", 0),
)
try:
metadata = os.fstat(descriptor)
if not stat.S_ISREG(metadata.st_mode):
raise ValueError("publication member is not a regular file")
os.fsync(descriptor)
finally:
os.close(descriptor)
def assert_private_result_base(path: Path) -> None:
current = Path(path.anchor)
for part in path.parts[1:]:
current /= part
metadata = current.lstat()
is_result_base = current == path
expected_uid = os.geteuid() if is_result_base else 0
if (
not stat.S_ISDIR(metadata.st_mode)
or current.is_symlink()
or metadata.st_uid != expected_uid
or metadata.st_mode & 0o022
):
raise PermissionError(
"result directory chain is not privileged and non-writable: "
+ str(current)
)
def rename_noreplace(source: Path, target: Path) -> None:
libc = ctypes.CDLL(None, use_errno=True)
function = getattr(libc, "renameat2", None)
if function is None:
raise OSError(errno.ENOSYS, "renameat2 is required for no-replace publication")
function.argtypes = [
ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p,
ctypes.c_uint,
]
function.restype = ctypes.c_int
result = function(
AT_FDCWD, os.fsencode(source), AT_FDCWD, os.fsencode(target),
RENAME_NOREPLACE,
)
if result != 0:
number = ctypes.get_errno()
raise OSError(number, os.strerror(number), str(target))
def publish_result(
value: dict, profile: dict, input_key_sha256s: list[str],
*, signing_key_uri: str,
) -> tuple[Path, dict]:
if not signing_key_uri.startswith("env://D014"):
raise ValueError("result signing-key URI is not a fixed D014 secret name")
result_base = Path(profile["result_base"])
assert_private_result_base(result_base)
key_spec = profile["verification_keys"]["result"]
result_key = read_stable(
Path(key_spec["path"]), maximum_bytes=HARD_KEY_CAPTURE_CEILING,
trust_root=True,
)
result_key_sha = hashlib.sha256(result_key).hexdigest()
if (
not hmac.compare_digest(result_key_sha, key_spec["sha256"])
or result_key_sha in set(input_key_sha256s)
):
raise ValueError("result verification key is unpinned or not role-separated")
published = {**value, "result_verification_key_sha256": result_key_sha}
raw = canonical(published) + b"\n"
if len(raw) > HARD_METADATA_CAPTURE_CEILING:
raise ValueError("result exceeds its hard byte ceiling")
result_sha = hashlib.sha256(raw).hexdigest()
target = result_base / result_sha
if target.exists() or target.is_symlink():
raise FileExistsError("content-addressed result already exists: " + result_sha)
staging = Path(tempfile.mkdtemp(prefix="." + result_sha + ".", dir=result_base))
os.chmod(staging, 0o700)
try:
result_path = staging / "result.json"
bundle_path = staging / "result.sigstore.json"
key_path = staging / "result.pub"
write_exclusive(result_path, raw)
write_exclusive(key_path, result_key)
subprocess.run(
[
COSIGN, "sign-blob", "--yes", "--key", signing_key_uri,
"--bundle", str(bundle_path), str(result_path),
],
check=True, timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
fsync_regular_file(bundle_path)
bundle_raw = read_stable(
bundle_path, maximum_bytes=HARD_METADATA_CAPTURE_CEILING
)
if not hmac.compare_digest(
read_stable(result_path, maximum_bytes=HARD_METADATA_CAPTURE_CEILING), raw
):
raise RuntimeError("private result changed before verification")
verify_bytes("D014 result", raw, bundle_raw, result_key)
committed = canonical({
"schema_version": "aidefend.signed-result-publication.v1",
"payload_sha256": result_sha,
"bundle_sha256": hashlib.sha256(bundle_raw).hexdigest(),
"verification_key_sha256": result_key_sha,
}) + b"\n"
write_exclusive(staging / "COMMITTED.json", committed)
expected = {
"result.json": raw,
"result.sigstore.json": bundle_raw,
"result.pub": result_key,
"COMMITTED.json": committed,
}
if {path.name for path in staging.iterdir()} != set(expected):
raise RuntimeError("staged result directory contents differ")
for path in staging.iterdir():
os.chmod(path, 0o440)
fsync_regular_file(path)
os.chmod(staging, 0o550)
fsync_directory(staging)
rename_noreplace(staging, target)
fsync_directory(result_base)
staging = None
if {path.name for path in target.iterdir()} != set(expected):
raise RuntimeError("published result directory contents differ")
readback = {
name: read_stable(path, maximum_bytes=HARD_METADATA_CAPTURE_CEILING)
for name, path in ((name, target / name) for name in expected)
}
for name, wanted in expected.items():
if not hmac.compare_digest(readback[name], wanted):
raise RuntimeError("published result readback differs: " + name)
verify_bytes(
"published D014 result", readback["result.json"],
readback["result.sigstore.json"], readback["result.pub"],
)
return target, published
finally:
if staging is not None and staging.exists():
os.chmod(staging, 0o700)
for path in staging.iterdir():
os.chmod(path, 0o600)
shutil.rmtree(staging)
Evaluate the complete population
The evaluator proves every query contains ranks 1..k, validates that policy concentration/diversity requirements are feasible for the signed registry, and checks each exact (source_id, source_version). Concentration is the largest per-source share of the full top-k; equality with the maximum passes, while diversity must equal or exceed its minimum. Unknown sources and unapproved versions remain findings even when aggregate diversity looks healthy.
from __future__ import annotations
# File: detection/evaluate_retrieval_provenance.py
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
import hashlib
import hmac
from io import BytesIO
import math
from detection.d014_verified_artifacts import (
HARD_ARTIFACT_CAPTURE_CEILING, HARD_METADATA_CAPTURE_CEILING,
SHA256, canonical, capture_role, load_profile, publish_result,
strict_json_object,
)
CONTROL = "AID-D-014.003-G001"
PROFILE_NAME = "d014003-g001-provenance"
ROLES = ("snapshot", "receipt", "policy", "registry")
SIGNING_KEY_URI = "env://D014003_G001_RESULT_SIGNING_KEY"
HARD_QUERY_COUNT = 500_000
HARD_PAIR_COUNT = 5_000_000
HARD_LINE_CAPTURE_CEILING = 2 * 1024 * 1024
HARD_FINDING_SAMPLE_COUNT = 1_000
POLICY_FIELDS = {
"schema_version", "policy_version", "expected_top_k_by_query_class",
"maximum_concentration_by_query_class",
"minimum_distinct_sources_by_query_class", "source_registry_sha256",
"source_registry_version", "maximum_queries", "maximum_pairs",
"maximum_line_bytes", "maximum_snapshot_age_seconds",
"maximum_future_skew_seconds", "retention_policy_sha256",
}
RECEIPT_FIELDS = {
"schema_version", "subject", "snapshot_completed_at", "emitted_at",
"query_count", "pair_count", "collection_complete", "snapshot_sha256",
"source_registry_sha256", "source_registry_version",
}
REGISTRY_FIELDS = {"schema_version", "registry_version", "sources"}
SOURCE_FIELDS = {"source_id", "allowed_source_versions"}
RECORD_FIELDS = {
"query_id", "query_class", "rank", "chunk_id", "source_id", "source_version",
}
EXIT_CODES = {
"PASS": 0, "FAIL": 1, "INSUFFICIENT_DATA": 2,
"NOT_APPLICABLE": 0, "ERROR": 3,
}
def parse_time(value: object) -> datetime:
if not isinstance(value, str):
raise ValueError("timestamp is not text")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
def integer(value: object, label: str, *, positive: bool = False) -> int:
if isinstance(value, bool) or not isinstance(value, int):
raise ValueError(label + " must be an integer")
if value < (1 if positive else 0):
raise ValueError(label + " is outside its allowed range")
return value
def unit_interval(value: object, label: str) -> float:
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or not math.isfinite(float(value))
or not 0 <= float(value) <= 1
):
raise ValueError(label + " must be finite and in [0,1]")
return float(value)
def sorted_strings(value: object, label: str, *, allow_empty: bool = False) -> list[str]:
if (
not isinstance(value, list)
or (not allow_empty and not value)
or any(not isinstance(item, str) or not item for item in value)
or value != sorted(set(value))
):
raise ValueError(label + " must be a sorted unique string list")
return value
def records(raw: bytes, *, maximum_pairs: int, maximum_line_bytes: int) -> list[dict]:
if raw and not raw.endswith(b"\n"):
raise ValueError("provenance JSONL must end with a newline")
stream = BytesIO(raw)
output = []
while True:
line = stream.readline(maximum_line_bytes + 1)
if not line:
break
if len(line) > maximum_line_bytes:
raise ValueError("provenance record exceeds its signed line limit")
if not line.strip():
raise ValueError("provenance JSONL contains a blank record")
record = strict_json_object(line, "retrieval provenance record")
if (
set(record) != RECORD_FIELDS
or any(
not isinstance(record[field], str) or not record[field]
for field in RECORD_FIELDS - {"rank"}
)
):
raise ValueError("provenance record schema differs")
integer(record["rank"], "rank", positive=True)
output.append(record)
if len(output) > maximum_pairs:
raise ValueError("provenance population exceeds policy")
if len({(item["query_id"], item["rank"]) for item in output}) != len(output):
raise ValueError("duplicate query rank in provenance snapshot")
if len({(item["query_id"], item["chunk_id"]) for item in output}) != len(output):
raise ValueError("duplicate query-chunk pair in provenance snapshot")
return output
def load_registry(raw: bytes, policy: dict) -> dict[str, set[str]]:
registry = strict_json_object(raw, "source registry")
if (
set(registry) != REGISTRY_FIELDS
or registry.get("schema_version") != "aidefend.d014003-source-registry.v1"
or registry.get("registry_version") != policy["source_registry_version"]
or not isinstance(registry.get("sources"), list)
):
raise ValueError("source registry schema or version differs")
parsed = {}
observed_order = []
for position, source in enumerate(registry["sources"]):
if not isinstance(source, dict) or set(source) != SOURCE_FIELDS:
raise ValueError("source registry entry differs at " + str(position))
source_id = source.get("source_id")
if not isinstance(source_id, str) or not source_id or source_id in parsed:
raise ValueError("source registry contains an invalid or duplicate ID")
versions = sorted_strings(
source.get("allowed_source_versions"),
"allowed versions for " + source_id,
)
parsed[source_id] = set(versions)
observed_order.append(source_id)
if observed_order != sorted(observed_order):
raise ValueError("source registry must be sorted by source_id")
return parsed
def load_inputs(profile: dict, bootstrap, evaluated_at: datetime) -> dict:
captures = {
"snapshot": capture_role(
profile, "snapshot", maximum_bytes=HARD_ARTIFACT_CAPTURE_CEILING,
allow_empty=True,
),
"receipt": capture_role(
profile, "receipt", maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
),
"policy": capture_role(
profile, "policy", maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
),
"registry": capture_role(
profile, "registry", maximum_bytes=HARD_METADATA_CAPTURE_CEILING,
),
}
policy = strict_json_object(captures["policy"].raw, "provenance policy")
receipt = strict_json_object(captures["receipt"].raw, "provenance receipt")
if (
set(policy) != POLICY_FIELDS
or policy.get("schema_version") != "aidefend.d014003-provenance-policy.v2"
or any(
not isinstance(policy.get(field), str) or not policy[field]
for field in ("policy_version", "source_registry_version")
)
or not SHA256.fullmatch(str(policy.get("source_registry_sha256")))
or not SHA256.fullmatch(str(policy.get("retention_policy_sha256")))
):
raise ValueError("provenance policy schema or identity differs")
mappings = (
policy["expected_top_k_by_query_class"],
policy["maximum_concentration_by_query_class"],
policy["minimum_distinct_sources_by_query_class"],
)
if any(not isinstance(value, dict) or not value for value in mappings):
raise ValueError("query-class provenance mappings are absent")
classes = set(mappings[0])
if any(set(value) != classes for value in mappings[1:]):
raise ValueError("query-class provenance mappings differ")
for query_class in classes:
if not isinstance(query_class, str) or not query_class:
raise ValueError("query class is invalid")
top_k = integer(
policy["expected_top_k_by_query_class"][query_class],
"top-k for " + query_class, positive=True,
)
minimum_sources = integer(
policy["minimum_distinct_sources_by_query_class"][query_class],
"minimum sources for " + query_class, positive=True,
)
maximum_concentration = unit_interval(
policy["maximum_concentration_by_query_class"][query_class],
"concentration for " + query_class,
)
if minimum_sources > top_k or maximum_concentration < 1 / top_k:
raise ValueError("provenance policy is mathematically infeasible")
for field, ceiling in (
("maximum_queries", HARD_QUERY_COUNT),
("maximum_pairs", HARD_PAIR_COUNT),
("maximum_line_bytes", HARD_LINE_CAPTURE_CEILING),
):
if integer(policy[field], field, positive=True) > ceiling:
raise ValueError(field + " exceeds its process-safety ceiling")
integer(policy["maximum_snapshot_age_seconds"], "maximum_snapshot_age_seconds", positive=True)
integer(policy["maximum_future_skew_seconds"], "maximum_future_skew_seconds")
if not hmac.compare_digest(
captures["registry"].evidence()["payload_sha256"],
policy["source_registry_sha256"],
):
raise ValueError("policy does not bind the exact source registry")
registry = load_registry(captures["registry"].raw, policy)
for query_class in classes:
top_k = policy["expected_top_k_by_query_class"][query_class]
minimum_sources = policy[
"minimum_distinct_sources_by_query_class"
][query_class]
maximum_concentration = float(
policy["maximum_concentration_by_query_class"][query_class]
)
if (
minimum_sources > len(registry)
or not registry
or maximum_concentration
< math.ceil(top_k / len(registry)) / top_k
):
raise ValueError(
"provenance policy is infeasible for the signed source registry"
)
if (
set(receipt) != RECEIPT_FIELDS
or receipt.get("schema_version") != "aidefend.d014003-provenance-receipt.v2"
or set(receipt.get("subject", {})) != {"asset_id", "asset_version"}
or any(
not isinstance(receipt["subject"].get(field), str)
or not receipt["subject"][field]
for field in ("asset_id", "asset_version")
)
or receipt.get("source_registry_version") != policy["source_registry_version"]
or not hmac.compare_digest(
str(receipt.get("source_registry_sha256")), policy["source_registry_sha256"]
)
or not isinstance(receipt.get("collection_complete"), bool)
or not SHA256.fullmatch(str(receipt.get("snapshot_sha256")))
):
raise ValueError("provenance receipt schema, subject, or registry differs")
query_count = integer(receipt["query_count"], "query_count")
pair_count = integer(receipt["pair_count"], "pair_count")
if query_count > policy["maximum_queries"] or pair_count > policy["maximum_pairs"]:
raise ValueError("signed provenance population exceeds policy")
if not hmac.compare_digest(
captures["snapshot"].evidence()["payload_sha256"], receipt["snapshot_sha256"]
):
raise ValueError("provenance receipt does not bind the exact snapshot")
population = records(
captures["snapshot"].raw, maximum_pairs=policy["maximum_pairs"],
maximum_line_bytes=policy["maximum_line_bytes"],
)
if len(population) != pair_count:
raise ValueError("provenance pair count differs from receipt")
grouped = defaultdict(list)
for record in population:
if record["query_class"] not in classes:
raise ValueError("provenance query class is outside signed policy")
grouped[record["query_id"]].append(record)
if len(grouped) != query_count:
raise ValueError("provenance query count differs from receipt")
for query_id, items in grouped.items():
query_classes = {item["query_class"] for item in items}
if len(query_classes) != 1:
raise ValueError("query identity has inconsistent classes")
query_class = items[0]["query_class"]
expected = policy["expected_top_k_by_query_class"][query_class]
if sorted(item["rank"] for item in items) != list(range(1, expected + 1)):
raise ValueError("query does not contain the complete signed top-k")
completed = parse_time(receipt["snapshot_completed_at"])
emitted = parse_time(receipt["emitted_at"])
if emitted < completed:
raise ValueError("receipt predates provenance snapshot completion")
if emitted > evaluated_at + timedelta(seconds=policy["maximum_future_skew_seconds"]):
raise ValueError("receipt is future-dated beyond policy skew")
age = max(0.0, (evaluated_at - completed).total_seconds())
if age > policy["maximum_snapshot_age_seconds"]:
readiness, reason = "INSUFFICIENT_DATA", "signed provenance snapshot is stale"
elif receipt["collection_complete"] is not True:
readiness, reason = "INSUFFICIENT_DATA", "authenticated top-k collection is incomplete"
else:
readiness, reason = "READY", None
return {
"status": readiness, "reason": reason, "profile": profile,
"bootstrap": bootstrap, "captures": captures, "policy": policy,
"receipt": receipt, "population": population, "registry": registry,
"snapshot_age_seconds": age,
}
def assess(population: list[dict], registry: dict[str, set[str]], policy: dict) -> dict:
grouped = defaultdict(list)
for record in population:
grouped[record["query_id"]].append(record)
findings = []
finding_count = 0
digest = hashlib.sha256()
for query_id in sorted(grouped):
items = sorted(grouped[query_id], key=lambda item: item["rank"])
query_class = items[0]["query_class"]
source_counts = Counter(item["source_id"] for item in items)
concentration = max(source_counts.values()) / len(items)
distinct_sources = len(source_counts)
query_findings = []
for item in items:
allowed = registry.get(item["source_id"])
if allowed is None:
query_findings.append({
"type": "unknown_source", "chunk_id": item["chunk_id"],
"source_id": item["source_id"], "source_version": item["source_version"],
})
elif item["source_version"] not in allowed:
query_findings.append({
"type": "unapproved_source_version", "chunk_id": item["chunk_id"],
"source_id": item["source_id"], "source_version": item["source_version"],
})
maximum = float(policy["maximum_concentration_by_query_class"][query_class])
minimum = policy["minimum_distinct_sources_by_query_class"][query_class]
if concentration > maximum:
query_findings.append({
"type": "source_concentration", "observed": concentration,
"maximum": maximum,
})
if distinct_sources < minimum:
query_findings.append({
"type": "source_diversity", "observed": distinct_sources,
"minimum": minimum,
})
record = {
"query_id": query_id, "query_class": query_class,
"source_counts": dict(sorted(source_counts.items())),
"concentration": concentration, "distinct_sources": distinct_sources,
"finding_count": len(query_findings),
}
digest.update(canonical(record) + b"\n")
finding_count += len(query_findings)
for finding in query_findings:
if len(findings) < HARD_FINDING_SAMPLE_COUNT:
findings.append({"query_id": query_id, **finding})
return {
"evaluated_queries": len(grouped), "evaluated_pairs": len(population),
"finding_count": finding_count, "finding_sample": findings,
"query_records_sha256": digest.hexdigest(),
}
def evaluate(profile: dict, bootstrap, evaluated_at: datetime) -> tuple[dict, list[str]]:
inputs = load_inputs(profile, bootstrap, evaluated_at)
receipt, policy = inputs["receipt"], inputs["policy"]
evidence = {name: item.evidence() for name, item in inputs["captures"].items()}
base = {
"schema_version": "aidefend.d014003-provenance-result.v3",
"control": CONTROL, "subject": receipt["subject"],
"policy_version": policy["policy_version"],
"source_registry_version": policy["source_registry_version"],
"profile_version": profile["profile_version"],
"profile_sha256": bootstrap.evidence()["payload_sha256"],
"snapshot_age_seconds": inputs["snapshot_age_seconds"],
"retention_policy_sha256": policy["retention_policy_sha256"],
"input_evidence": evidence,
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
}
keys = [bootstrap.evidence()["verification_key_sha256"]] + [
item.evidence()["verification_key_sha256"] for item in inputs["captures"].values()
]
if inputs["status"] != "READY":
return {**base, "status": inputs["status"], "reason": inputs["reason"], "metrics": None}, keys
if not inputs["population"]:
return {**base, "status": "INSUFFICIENT_DATA", "reason": "no complete top-k query occurred", "metrics": None}, keys
metrics = assess(inputs["population"], inputs["registry"], policy)
if metrics["evaluated_queries"] != receipt["query_count"] or metrics["evaluated_pairs"] != receipt["pair_count"]:
raise ValueError("provenance population does not reconcile")
if metrics["finding_count"]:
status, reason = "FAIL", "source identity, version, concentration, or diversity finding exists"
else:
status, reason = "PASS", "every complete top-k met signed provenance constraints"
return {**base, "status": status, "reason": reason, "metrics": metrics}, keys
def main() -> int:
evaluated_at = datetime.now(timezone.utc)
profile, bootstrap = load_profile(
CONTROL, ROLES, profile_name=PROFILE_NAME, evaluated_at=evaluated_at,
)
reserved = [bootstrap.evidence()["verification_key_sha256"]] + [
spec["sha256"] for role, spec in profile["verification_keys"].items()
if role != "result"
]
try:
result, observed = evaluate(profile, bootstrap, evaluated_at)
reserved = list(set(reserved + observed))
except Exception as error:
result = {
"schema_version": "aidefend.d014003-provenance-result.v3",
"control": CONTROL, "profile_version": profile["profile_version"],
"profile_sha256": bootstrap.evidence()["payload_sha256"],
"status": "ERROR",
"reason": "input verification, provenance evaluation, or result preparation failed",
"error_class": type(error).__name__,
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
}
target, published = publish_result(
result, profile, reserved, signing_key_uri=SIGNING_KEY_URI,
)
print(str(target))
return EXIT_CODES[published["status"]]
if __name__ == "__main__":
raise SystemExit(main())
Run and retain the result
Use Python 3.11 or newer with the Python standard library and Cosign. Provision the result directory for the dedicated evaluator identity, sign each exact input with its separately authorized role key, then sign the profile with the offline bootstrap role:
install -d -m 0755 -o root -g root /var/lib/aidefend/d014 /var/lib/aidefend/d014/results
install -d -m 0750 -o aidefend-d014 -g aidefend-d014 /var/lib/aidefend/d014/results/d014003-g001-provenance
cosign sign-blob --yes --key env://D014_SNAPSHOT_SIGNING_KEY --bundle /var/lib/aidefend/d014/input/d014003-g001-provenance-snapshot.jsonl.sigstore.json /var/lib/aidefend/d014/input/d014003-g001-provenance-snapshot.jsonl
cosign sign-blob --yes --key env://D014_RECEIPT_SIGNING_KEY --bundle /var/lib/aidefend/d014/input/d014003-g001-provenance-receipt.json.sigstore.json /var/lib/aidefend/d014/input/d014003-g001-provenance-receipt.json
cosign sign-blob --yes --key env://D014_POLICY_SIGNING_KEY --bundle /etc/aidefend/d014/input/d014003-g001-provenance-policy.json.sigstore.json /etc/aidefend/d014/input/d014003-g001-provenance-policy.json
cosign sign-blob --yes --key env://D014_REGISTRY_SIGNING_KEY --bundle /etc/aidefend/d014/input/d014003-g001-provenance-registry.json.sigstore.json /etc/aidefend/d014/input/d014003-g001-provenance-registry.json
: "${D014_PROFILE_VERSION:?set a deployment-controlled profile version}"
: "${D014_PROFILE_VALID_FROM:?set an RFC 3339 validity start}"
: "${D014_PROFILE_VALID_UNTIL:?set an RFC 3339 validity end}"
python -m detection.provision_d014003_g001_provenance_profile "${D014_PROFILE_VERSION}" "${D014_PROFILE_VALID_FROM}" "${D014_PROFILE_VALID_UNTIL}"
cosign sign-blob --yes --key env://D014_RUNTIME_PROFILE_SIGNING_KEY --bundle /etc/aidefend/d014/profiles/d014003-g001-provenance.sigstore.json /etc/aidefend/d014/profiles/d014003-g001-provenance.json
python -m detection.evaluate_retrieval_provenanceA complete top-k with approved versions and compliant diversity/concentration is PASS; any source, version, concentration, or diversity finding is FAIL; stale, incomplete, or empty applicable populations are INSUFFICIENT_DATA; infeasible policy, registry, integrity, schema, or publication faults are ERROR. The evaluator obtains UTC time once inside main(). It signs a canonical v3 result, verifies it against the profile-pinned result key, commits it under its content digest with no-replace semantics, and reads every file back before returning. Replicate the committed directory and exact signed inputs to append-only or WORM storage under a separate retention identity; replay from those bytes rather than reopening mutable producer paths.
Open-source tools
- pandas
- Cosign
- GX Core (custom source-coverage checks)
- Evidently (custom retrieval-diversity checks)
Commercial / hosted tools
- Datadog
- New Relic
- Splunk Observability Cloud
Threat mappings
MITRE ATLAS
- AML.T0099 AI Agent Tool Data Poisoning
- AML.T0070 RAG Poisoning
- AML.T0071 False RAG Entry Injection
- AML.T0066 Retrieval Content Crafting (source concentration reveals crafted content campaigns)
MAESTRO
- Compromised RAG Pipelines (L2)
OWASP LLM Top 10 2026
- LLM05:2026 Data and Model Poisoning (unregistered provenance and source concentration expose poisoned RAG content)
- LLM09:2026 Vector and Embedding Weaknesses
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI06:2026 Memory & Context Poisoning
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- AISubtech-6.1.1 Knowledge Base Poisoning (unapproved source versions and excessive source concentration in retrieved top-k)
- AISubtech-7.2.2 Memory Index Manipulation (unapproved source versions and excessive source concentration in retrieved top-k)
- AITech-7.2 Memory System Corruption (unapproved source versions and excessive source concentration in retrieved top-k)
- AITech-7.3 Data Source Abuse and Manipulation (unapproved source versions and excessive source concentration in retrieved top-k)
Google Secure AI Framework 2.0 - Risks
- N/A
Databricks AI Security Framework 3.0
- Raw Data 1.11: Compromised 3rd-party datasets (source concentration reveals compromised data sources)
- Raw Data 1.7: Lack of data trustworthiness