AID-D-014.001: Post-Retrieval Malicious Content Scanning
Treat every chunk in the exact post-retrieval population as untrusted input; inspect its content or policy-bound embedding representation for prompt-injection patterns, malicious payloads, or known malicious concepts and emit poisoning findings before context assembly. Use the same pinned detector contract as prompt ingress, with modern guardrail options such as Llama Prompt Guard 2, Llama Guard 4, Granite Guardian 4.1, Qwen3Guard, or gpt-oss-safeguard where they fit the language, modality, latency, and policy requirements.
Pillars: data
Lifecycle: operation
Scope boundary
Owns malicious-instruction, prompt-injection, and payload findings for chunks already retrieved from a RAG index or vector store before context assembly. It does not remove, rewrite, quarantine, or admit the chunk.
AID-D-001.001: AID-D-014.001 scans retrieved knowledge chunks at the post-retrieval observation point; AID-D-001.001 scans the inbound user prompt and recovered obfuscation views. They may share detector technology, but the source identity, population, and downstream policy binding are different.
AID-D-014.002: AID-D-014.001 asks whether retrieved content is malicious or policy-violating; AID-D-014.002 asks whether each retrieved chunk is semantically relevant to the query. A relevant chunk can contain an injection, and a benign chunk can be suspiciously irrelevant.
Implementation guidance
- AID-D-014.001-G001: Scan retrieved chunks with prompt-safety detectors and emit poisoning findings for tainted content.
- AID-D-014.001-G002: Compare every chunk in the exact ordered retrieval population with policy-owned malicious-concept embeddings and emit content findings before context assembly.
AID-D-014.001-G001
Scan retrieved chunks with prompt-safety detectors and emit poisoning findings for tainted content.
Pin the runtime inputs
Run this classifier only on the exact retrieved-chunk snapshot produced for one asset version and retrieval route. 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_d014001_g001_post_retrieval_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.001-G001"
PROFILE_NAME = "d014001-g001-post-retrieval"
ARTIFACTS = {
"snapshot": {
"payload": "/var/lib/aidefend/d014/input/d014001-g001-post-retrieval-snapshot.jsonl",
"bundle": "/var/lib/aidefend/d014/input/d014001-g001-post-retrieval-snapshot.jsonl.sigstore.json"
},
"receipt": {
"payload": "/var/lib/aidefend/d014/input/d014001-g001-post-retrieval-receipt.json",
"bundle": "/var/lib/aidefend/d014/input/d014001-g001-post-retrieval-receipt.json.sigstore.json"
},
"policy": {
"payload": "/etc/aidefend/d014/input/d014001-g001-post-retrieval-policy.json",
"bundle": "/etc/aidefend/d014/input/d014001-g001-post-retrieval-policy.json.sigstore.json"
},
"model": {
"payload": "/var/lib/aidefend/d014/input/d014001-g001-post-retrieval-model.tar",
"bundle": "/var/lib/aidefend/d014/input/d014001-g001-post-retrieval-model.tar.sigstore.json"
}
}
KEY_PATHS = {
"snapshot": "/etc/aidefend/d014/trust/d014001-g001-post-retrieval-snapshot.pub",
"receipt": "/etc/aidefend/d014/trust/d014001-g001-post-retrieval-receipt.pub",
"policy": "/etc/aidefend/d014/trust/d014001-g001-post-retrieval-policy.pub",
"model": "/etc/aidefend/d014/trust/d014001-g001-post-retrieval-model.pub",
"result": "/etc/aidefend/d014/trust/d014001-g001-post-retrieval-result.pub"
}
RESULT_BASE = "/var/lib/aidefend/d014/results/d014001-g001-post-retrieval"
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())The collector-signed JSONL contains exactly chunk_id, route, and text. An independently signed receipt binds its digest, count, route, asset, completion time, and completeness flag. The policy separately binds the model-tar digest and revision, exact id2label, score transform, per-label thresholds, sequence size, overlap, and population/freshness limits.
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 tokenizer runs with truncation=False. Every chunk is split into overlapping token windows with no uncovered tail, every window is scored under the explicit sigmoid or softmax contract, and each label uses the maximum score across all windows. A score equal to its category threshold is a finding. The result records complete chunk/token/window counts and a digest of every score record; a resource-bounded unmeasured chunk cannot become a clean result.
from __future__ import annotations
# File: detection/evaluate_post_retrieval_classifier.py
from datetime import datetime, timedelta, timezone
import hashlib
import hmac
from io import BytesIO
import json
import math
import re
from detection.d014_verified_artifacts import (
HARD_ARTIFACT_CAPTURE_CEILING, HARD_METADATA_CAPTURE_CEILING,
SHA256, canonical, capture_role, load_profile, publish_result,
safe_model_root, strict_json_object,
)
CONTROL = "AID-D-014.001-G001"
PROFILE_NAME = "d014001-g001-post-retrieval"
ROLES = ("snapshot", "receipt", "policy", "model")
SIGNING_KEY_URI = "env://D014001_G001_RESULT_SIGNING_KEY"
HARD_CHUNK_COUNT = 50_000
HARD_LINE_CAPTURE_CEILING = 8 * 1024 * 1024
HARD_CHUNK_TOKEN_COUNT = 1_000_000
HARD_TOTAL_WINDOW_COUNT = 200_000
HARD_SEQUENCE_TOKEN_COUNT = 32_768
HARD_FINDING_SAMPLE_COUNT = 1_000
POLICY_FIELDS = {
"schema_version", "policy_version", "applicable_routes",
"model_sha256", "model_revision", "id2label", "score_transform",
"category_thresholds", "max_sequence_tokens", "window_overlap_tokens",
"maximum_chunk_tokens", "maximum_total_windows", "maximum_chunks",
"maximum_snapshot_age_seconds", "maximum_future_skew_seconds",
"retention_policy_sha256",
}
RECEIPT_FIELDS = {
"schema_version", "subject", "route", "snapshot_completed_at",
"emitted_at", "chunk_count", "collection_complete", "snapshot_sha256",
}
RECORD_FIELDS = {"chunk_id", "route", "text"}
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 ratio(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 jsonl_records(raw: bytes, maximum_records: int) -> list[dict]:
if not raw:
return []
if not raw.endswith(b"\n"):
raise ValueError("chunk JSONL must end with a newline")
records = []
stream = BytesIO(raw)
while True:
line = stream.readline(HARD_LINE_CAPTURE_CEILING + 1)
if not line:
break
if len(line) > HARD_LINE_CAPTURE_CEILING:
raise ValueError("chunk record exceeds its hard line ceiling")
if not line.strip():
raise ValueError("chunk JSONL contains a blank record")
record = strict_json_object(line, "retrieved chunk")
if (
set(record) != RECORD_FIELDS
or any(
not isinstance(record[field], str) or not record[field]
for field in RECORD_FIELDS
)
):
raise ValueError("retrieved chunk schema differs")
records.append(record)
if len(records) > maximum_records:
raise ValueError("chunk population exceeds its signed limit")
if len({record["chunk_id"] for record in records}) != len(records):
raise ValueError("duplicate chunk_id in retrieval snapshot")
return records
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,
),
"model": capture_role(
profile, "model", maximum_bytes=HARD_ARTIFACT_CAPTURE_CEILING,
),
}
policy = strict_json_object(captures["policy"].raw, "classifier policy")
receipt = strict_json_object(captures["receipt"].raw, "retrieval receipt")
if (
set(policy) != POLICY_FIELDS
or policy.get("schema_version")
!= "aidefend.d014001-post-retrieval-policy.v2"
or not isinstance(policy.get("policy_version"), str)
or not policy["policy_version"]
or policy.get("score_transform") not in {"sigmoid", "softmax"}
or not SHA256.fullmatch(str(policy.get("model_sha256")))
or not SHA256.fullmatch(str(policy.get("retention_policy_sha256")))
or not isinstance(policy.get("model_revision"), str)
or not policy["model_revision"]
):
raise ValueError("classifier policy schema or identity differs")
routes = sorted_strings(policy["applicable_routes"], "applicable_routes")
id2label = policy["id2label"]
if not isinstance(id2label, dict) or not id2label:
raise ValueError("id2label is absent")
expected_indexes = [str(index) for index in range(len(id2label))]
if (
sorted(id2label, key=lambda value: int(value) if value.isdigit() else -1)
!= expected_indexes
or any(not isinstance(id2label[index], str) or not id2label[index]
for index in expected_indexes)
or len(set(id2label.values())) != len(id2label)
):
raise ValueError("id2label must cover contiguous unique labels")
thresholds = policy["category_thresholds"]
if not isinstance(thresholds, dict) or set(thresholds) != set(id2label.values()):
raise ValueError("category thresholds do not cover the model labels")
for label, value in thresholds.items():
ratio(value, "threshold for " + label)
for field, ceiling in (
("max_sequence_tokens", HARD_SEQUENCE_TOKEN_COUNT),
("maximum_chunk_tokens", HARD_CHUNK_TOKEN_COUNT),
("maximum_total_windows", HARD_TOTAL_WINDOW_COUNT),
("maximum_chunks", HARD_CHUNK_COUNT),
):
if integer(policy[field], field, positive=True) > ceiling:
raise ValueError(field + " exceeds its process-safety ceiling")
overlap = integer(policy["window_overlap_tokens"], "window_overlap_tokens")
if overlap >= policy["max_sequence_tokens"]:
raise ValueError("window overlap leaves no forward progress")
for field in ("maximum_snapshot_age_seconds",):
if integer(policy[field], field, positive=True) <= 0:
raise ValueError(field + " must be positive")
integer(policy["maximum_future_skew_seconds"], "maximum_future_skew_seconds")
if not hmac.compare_digest(
captures["model"].evidence()["payload_sha256"], policy["model_sha256"]
):
raise ValueError("classifier policy does not bind the model tar")
if (
set(receipt) != RECEIPT_FIELDS
or receipt.get("schema_version")
!= "aidefend.d014001-post-retrieval-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 not isinstance(receipt.get("route"), str)
or not receipt["route"]
or not isinstance(receipt.get("collection_complete"), bool)
or not SHA256.fullmatch(str(receipt.get("snapshot_sha256")))
):
raise ValueError("retrieval receipt schema or subject differs")
integer(receipt["chunk_count"], "chunk_count")
if receipt["chunk_count"] > policy["maximum_chunks"]:
raise ValueError("signed chunk count exceeds policy")
if not hmac.compare_digest(
captures["snapshot"].evidence()["payload_sha256"],
receipt["snapshot_sha256"],
):
raise ValueError("retrieval receipt does not bind the exact snapshot")
records = jsonl_records(captures["snapshot"].raw, policy["maximum_chunks"])
if len(records) != receipt["chunk_count"]:
raise ValueError("retrieval snapshot count differs from receipt")
if any(record["route"] != receipt["route"] for record in records):
raise ValueError("chunk route differs from signed receipt")
completed = parse_time(receipt["snapshot_completed_at"])
emitted = parse_time(receipt["emitted_at"])
if emitted < completed:
raise ValueError("receipt predates 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 = "INSUFFICIENT_DATA"
reason = "signed retrieval snapshot is stale"
elif receipt["collection_complete"] is not True:
readiness = "INSUFFICIENT_DATA"
reason = "authenticated retrieval collection is incomplete"
else:
readiness, reason = "READY", None
return {
"status": readiness, "reason": reason, "profile": profile,
"bootstrap": bootstrap, "captures": captures, "policy": policy,
"receipt": receipt, "records": records, "snapshot_age_seconds": age,
"applicable": receipt["route"] in routes,
}
def window_starts(token_count: int, capacity: int, overlap: int) -> list[int]:
if token_count <= capacity:
return [0]
step = capacity - overlap
final_start = token_count - capacity
starts = list(range(0, final_start + 1, step))
if starts[-1] != final_start:
starts.append(final_start)
return starts
def scan(records: list[dict], model_raw: bytes, policy: dict) -> dict:
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
labels = [policy["id2label"][str(index)] for index in range(len(policy["id2label"]))]
digest = hashlib.sha256()
findings = []
finding_count = 0
scanned_chunks = 0
scanned_tokens = 0
scanned_windows = 0
unmeasured_chunks = 0
with safe_model_root(model_raw) as model_root:
tokenizer = AutoTokenizer.from_pretrained(
str(model_root), local_files_only=True, trust_remote_code=False,
)
model = AutoModelForSequenceClassification.from_pretrained(
str(model_root), local_files_only=True, trust_remote_code=False,
use_safetensors=True,
)
model.eval()
model_labels = {
str(index): str(value)
for index, value in dict(model.config.id2label).items()
}
if model_labels != policy["id2label"] or model.config.num_labels != len(labels):
raise ValueError("model label semantics differ from signed policy")
model_limit = getattr(model.config, "max_position_embeddings", None)
tokenizer_limit = getattr(tokenizer, "model_max_length", None)
usable_limits = [
value for value in (model_limit, tokenizer_limit)
if isinstance(value, int) and not isinstance(value, bool) and 0 < value < 1_000_000
]
if usable_limits and policy["max_sequence_tokens"] > min(usable_limits):
raise ValueError("signed sequence limit exceeds the loaded model limit")
special = tokenizer.num_special_tokens_to_add(pair=False)
capacity = policy["max_sequence_tokens"] - special
overlap = policy["window_overlap_tokens"]
if capacity <= 0 or overlap >= capacity:
raise ValueError("signed window contract leaves no content capacity")
step = capacity - overlap
for record in records:
token_ids = tokenizer.encode(
record["text"], add_special_tokens=False, truncation=False,
)
if len(token_ids) > policy["maximum_chunk_tokens"]:
unmeasured_chunks += 1
continue
starts = window_starts(len(token_ids), capacity, overlap)
count = len(starts)
if scanned_windows + count > policy["maximum_total_windows"]:
unmeasured_chunks += 1
continue
maxima = [0.0] * len(labels)
for start in starts:
piece = token_ids[start:start + capacity]
encoded = tokenizer.prepare_for_model(
piece, add_special_tokens=True, truncation=False,
return_attention_mask=True, return_tensors="pt",
)
if int(encoded["input_ids"].shape[-1]) > policy["max_sequence_tokens"]:
raise ValueError("prepared classifier window exceeds policy")
batch = {name: value.to(model.device) for name, value in encoded.items()}
with torch.inference_mode():
logits = model(**batch).logits
if tuple(logits.shape) != (1, len(labels)) or not torch.isfinite(logits).all():
raise ValueError("classifier logits shape or values differ")
if policy["score_transform"] == "sigmoid":
values = torch.sigmoid(logits)[0].detach().cpu().tolist()
else:
values = torch.softmax(logits, dim=-1)[0].detach().cpu().tolist()
if len(values) != len(labels) or any(
not isinstance(value, (int, float))
or not math.isfinite(float(value))
or not 0 <= float(value) <= 1
for value in values
):
raise ValueError("classifier score domain differs")
maxima = [max(old, float(new)) for old, new in zip(maxima, values)]
scanned_windows += 1
scored = {
"chunk_id": record["chunk_id"], "window_count": count,
"max_scores": {label: maxima[index] for index, label in enumerate(labels)},
}
digest.update(canonical(scored) + b"\n")
for index, label in enumerate(labels):
if maxima[index] >= float(policy["category_thresholds"][label]):
finding_count += 1
if len(findings) < HARD_FINDING_SAMPLE_COUNT:
findings.append({
"chunk_id": record["chunk_id"], "label": label,
"score": maxima[index],
"threshold": float(policy["category_thresholds"][label]),
})
scanned_chunks += 1
scanned_tokens += len(token_ids)
return {
"scanned_chunks": scanned_chunks, "scanned_tokens": scanned_tokens,
"scanned_windows": scanned_windows, "unmeasured_chunks": unmeasured_chunks,
"finding_count": finding_count, "finding_sample": findings,
"score_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: captured.evidence() for name, captured in inputs["captures"].items()
}
base = {
"schema_version": "aidefend.d014001-post-retrieval-result.v3",
"control": CONTROL, "subject": receipt["subject"],
"route": receipt["route"], "policy_version": policy["policy_version"],
"model_revision": policy["model_revision"],
"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"),
}
key_digests = [bootstrap.evidence()["verification_key_sha256"]] + [
captured.evidence()["verification_key_sha256"]
for captured in inputs["captures"].values()
]
if inputs["status"] != "READY":
return {**base, "status": inputs["status"], "reason": inputs["reason"], "metrics": None}, key_digests
if not inputs["applicable"]:
return {**base, "status": "NOT_APPLICABLE", "reason": "signed route is outside the classifier policy", "metrics": None}, key_digests
if not inputs["records"]:
return {**base, "status": "INSUFFICIENT_DATA", "reason": "no retrieved chunk occurred in the complete snapshot", "metrics": None}, key_digests
metrics = scan(inputs["records"], inputs["captures"]["model"].raw, policy)
if metrics["finding_count"]:
status, reason = "FAIL", "one or more complete chunk windows met a signed category threshold"
elif metrics["unmeasured_chunks"]:
status, reason = "INSUFFICIENT_DATA", "one or more chunks could not be fully windowed within policy limits"
elif metrics["scanned_chunks"] != len(inputs["records"]):
raise ValueError("classifier population does not reconcile")
else:
status, reason = "PASS", "every chunk and token window remained below signed thresholds"
return {**base, "status": status, "reason": reason, "metrics": metrics}, key_digests
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.d014001-post-retrieval-result.v3",
"control": CONTROL, "profile_version": profile["profile_version"],
"profile_sha256": bootstrap.evidence()["payload_sha256"],
"status": "ERROR",
"reason": "input verification, model 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 torch, transformers, safetensors, 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/d014001-g001-post-retrieval
cosign sign-blob --yes --key env://D014_SNAPSHOT_SIGNING_KEY --bundle /var/lib/aidefend/d014/input/d014001-g001-post-retrieval-snapshot.jsonl.sigstore.json /var/lib/aidefend/d014/input/d014001-g001-post-retrieval-snapshot.jsonl
cosign sign-blob --yes --key env://D014_RECEIPT_SIGNING_KEY --bundle /var/lib/aidefend/d014/input/d014001-g001-post-retrieval-receipt.json.sigstore.json /var/lib/aidefend/d014/input/d014001-g001-post-retrieval-receipt.json
cosign sign-blob --yes --key env://D014_POLICY_SIGNING_KEY --bundle /etc/aidefend/d014/input/d014001-g001-post-retrieval-policy.json.sigstore.json /etc/aidefend/d014/input/d014001-g001-post-retrieval-policy.json
cosign sign-blob --yes --key env://D014_MODEL_SIGNING_KEY --bundle /var/lib/aidefend/d014/input/d014001-g001-post-retrieval-model.tar.sigstore.json /var/lib/aidefend/d014/input/d014001-g001-post-retrieval-model.tar
: "${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_d014001_g001_post_retrieval_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/d014001-g001-post-retrieval.sigstore.json /etc/aidefend/d014/profiles/d014001-g001-post-retrieval.json
python -m detection.evaluate_post_retrieval_classifierA policy-excluded route is NOT_APPLICABLE.
For an applicable route, a complete clean population is PASS; any threshold finding is FAIL; stale, incomplete, empty, or not-fully-windowed populations are INSUFFICIENT_DATA; integrity, schema, model, 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.
AID-D-014.001-G002
Compare every chunk in the exact ordered retrieval population with policy-owned malicious-concept embeddings and emit content findings before context assembly.
Bind the method to one exact retrieval population
Run this semantic method only on the complete ordered chunk population returned for one retrieval. The retrieval adapter exports a signed snapshot containing the request and retrieval IDs, index asset and version, route, expected count, ranks, chunk content digests, and vectors produced by the exact pinned embedding model. The detector owner separately signs the malicious-concept vectors and policy-owned thresholds. Do not enumerate the vector store or turn this guidance into an at-rest index scan.
Define signed policy and snapshot contracts
{
"schema_version": "aidefend.d014001-retrieved-concept-policy.v1",
"policy_version": "rag-content-prod-2026-08",
"embedding_model_id": "text-embedding-3-large",
"embedding_model_revision": "provider-revision-2026-08-15",
"maximum_chunks": 32,
"maximum_concepts": 64,
"vector_dimension": 3072,
"maximum_snapshot_age_seconds": 120,
"maximum_future_skew_seconds": 30,
"concept_thresholds": {
"ignore-prior-instructions": 0.84,
"credential-exfiltration": 0.82
}
}The counts, freshness windows, model revision, and thresholds above are illustrative policy values, not AIDEFEND defaults. The signed retrieval snapshot uses schema aidefend.d014001-retrieval-snapshot.v1 and contains request_id, retrieval_id, asset_id, asset_version, route_id, captured_at, collection_complete, expected_chunk_count, the embedding identity, and chunks. Every chunk contains exactly chunk_id, zero-based rank, lowercase SHA-256 of the canonical retrieved content, and its finite vector. The concept-set schema binds the same embedding identity to a unique concept population.
Evaluate every retrieved chunk against every governed concept
# File: detection/evaluate_retrieved_concepts.py
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import re
import stat
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
HARD_INPUT_BYTES = 64 * 1024 * 1024
HARD_CHUNKS = 512
HARD_CONCEPTS = 512
HARD_DIMENSION = 8192
HARD_FINDING_SAMPLE = 100
SHA256 = re.compile(r"^[a-f0-9]{64}$")
RFC3339_Z = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$")
class InputError(Exception):
pass
def read_stable(path: Path) -> bytes:
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 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 > HARD_INPUT_BYTES:
raise InputError("input is not a bounded regular file")
chunks, remaining = [], before.st_size
while remaining:
chunk = os.read(descriptor, min(1024 * 1024, remaining))
if not chunk:
raise InputError("input ended before its declared size")
chunks.append(chunk)
remaining -= len(chunk)
if os.read(descriptor, 1):
raise InputError("input grew while being captured")
after = os.fstat(descriptor)
identity_before = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns)
identity_after = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns)
if identity_before != identity_after:
raise InputError("input changed while being captured")
return b"".join(chunks)
finally:
os.close(descriptor)
def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
value: dict[str, Any] = {}
for key, item in pairs:
if key in value:
raise InputError("duplicate JSON key: " + key)
value[key] = item
return value
def strict_json(raw: bytes, label: str) -> dict[str, Any]:
try:
value = json.loads(
raw.decode("utf-8"), object_pairs_hook=unique_object,
parse_constant=lambda token: (_ for _ in ()).throw(InputError("non-finite JSON number: " + token)),
)
except (UnicodeError, json.JSONDecodeError) as error:
raise InputError(label + " is not strict UTF-8 JSON") from error
if not isinstance(value, dict):
raise InputError(label + " must be one JSON object")
return value
def exact_keys(value: dict[str, Any], expected: set[str], label: str) -> None:
if set(value) != expected:
raise InputError(label + " fields differ from the signed contract")
def integer(value: Any, label: str, minimum: int = 0) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
raise InputError(label + " must be an integer at least " + str(minimum))
return value
def finite_number(value: Any, label: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise InputError(label + " must be numeric")
result = float(value)
if not math.isfinite(result):
raise InputError(label + " must be finite")
return result
def parse_time(value: Any, label: str) -> datetime:
if not isinstance(value, str) or not RFC3339_Z.fullmatch(value):
raise InputError(label + " must be RFC 3339 UTC")
return datetime.fromisoformat(value[:-1] + "+00:00")
def vector(value: Any, dimension: int, label: str) -> list[float]:
if not isinstance(value, list) or len(value) != dimension:
raise InputError(label + " has the wrong vector dimension")
result = [finite_number(item, label) for item in value]
norm = math.sqrt(math.fsum(item * item for item in result))
if not math.isfinite(norm) or norm == 0:
raise InputError(label + " has a zero or non-finite norm")
return [item / norm for item in result]
def cosine(left: list[float], right: list[float]) -> float:
score = math.fsum(a * b for a, b in zip(left, right, strict=True))
if not math.isfinite(score):
raise InputError("cosine score is non-finite")
return max(-1.0, min(1.0, score))
def evaluate(policy_raw: bytes, retrieval_raw: bytes, concepts_raw: bytes) -> dict[str, Any]:
policy = strict_json(policy_raw, "policy")
retrieval = strict_json(retrieval_raw, "retrieval snapshot")
concept_set = strict_json(concepts_raw, "concept set")
exact_keys(policy, {
"schema_version", "policy_version", "embedding_model_id", "embedding_model_revision",
"maximum_chunks", "maximum_concepts", "vector_dimension",
"maximum_snapshot_age_seconds", "maximum_future_skew_seconds", "concept_thresholds",
}, "policy")
if policy["schema_version"] != "aidefend.d014001-retrieved-concept-policy.v1":
raise InputError("policy schema version differs")
for field in ("policy_version", "embedding_model_id", "embedding_model_revision"):
if not isinstance(policy[field], str) or not policy[field] or len(policy[field]) > 256:
raise InputError(field + " is invalid")
maximum_chunks = integer(policy["maximum_chunks"], "maximum_chunks", 1)
maximum_concepts = integer(policy["maximum_concepts"], "maximum_concepts", 1)
dimension = integer(policy["vector_dimension"], "vector_dimension", 1)
if maximum_chunks > HARD_CHUNKS or maximum_concepts > HARD_CONCEPTS or dimension > HARD_DIMENSION:
raise InputError("signed policy exceeds process safety ceilings")
max_age = finite_number(policy["maximum_snapshot_age_seconds"], "maximum_snapshot_age_seconds")
max_skew = finite_number(policy["maximum_future_skew_seconds"], "maximum_future_skew_seconds")
if max_age <= 0 or max_skew < 0:
raise InputError("freshness policy is invalid")
thresholds = policy["concept_thresholds"]
if not isinstance(thresholds, dict) or not thresholds or len(thresholds) > maximum_concepts:
raise InputError("concept threshold population is invalid")
normalized_thresholds = {}
for concept_id, value in thresholds.items():
if not isinstance(concept_id, str) or not concept_id or len(concept_id) > 128:
raise InputError("concept identifier is invalid")
threshold = finite_number(value, "threshold for " + concept_id)
if threshold < -1 or threshold > 1:
raise InputError("concept threshold is outside cosine range")
normalized_thresholds[concept_id] = threshold
exact_keys(retrieval, {
"schema_version", "request_id", "retrieval_id", "asset_id", "asset_version", "route_id",
"captured_at", "collection_complete", "expected_chunk_count", "embedding_model_id",
"embedding_model_revision", "chunks",
}, "retrieval snapshot")
if retrieval["schema_version"] != "aidefend.d014001-retrieval-snapshot.v1":
raise InputError("retrieval snapshot version differs")
for field in ("request_id", "retrieval_id", "asset_id", "asset_version", "route_id"):
if not isinstance(retrieval[field], str) or not retrieval[field] or len(retrieval[field]) > 256:
raise InputError(field + " is invalid")
if (
retrieval["embedding_model_id"] != policy["embedding_model_id"] or
retrieval["embedding_model_revision"] != policy["embedding_model_revision"]
):
raise InputError("retrieval embedding identity differs from policy")
if not isinstance(retrieval["collection_complete"], bool) or not isinstance(retrieval["chunks"], list):
raise InputError("retrieval completeness fields are invalid")
expected_count = integer(retrieval["expected_chunk_count"], "expected_chunk_count")
if expected_count > maximum_chunks or len(retrieval["chunks"]) > maximum_chunks:
raise InputError("retrieval population exceeds signed policy")
captured_at = parse_time(retrieval["captured_at"], "captured_at")
now = datetime.now(timezone.utc)
age = (now - captured_at).total_seconds()
incomplete_reasons = []
if age < -max_skew:
raise InputError("retrieval snapshot is future-dated beyond policy skew")
if age > max_age:
incomplete_reasons.append("retrieval snapshot is stale")
if retrieval["collection_complete"] is not True:
incomplete_reasons.append("retrieval adapter marked the population incomplete")
if len(retrieval["chunks"]) != expected_count:
incomplete_reasons.append("observed chunk count differs from expected count")
exact_keys(concept_set, {
"schema_version", "embedding_model_id", "embedding_model_revision", "concepts",
}, "concept set")
if concept_set["schema_version"] != "aidefend.d014001-concept-set.v1":
raise InputError("concept-set version differs")
if (
concept_set["embedding_model_id"] != policy["embedding_model_id"] or
concept_set["embedding_model_revision"] != policy["embedding_model_revision"] or
not isinstance(concept_set["concepts"], list) or
not 0 < len(concept_set["concepts"]) <= maximum_concepts
):
raise InputError("concept population or embedding identity differs")
concepts: dict[str, list[float]] = {}
for item in concept_set["concepts"]:
if not isinstance(item, dict):
raise InputError("concept row must be an object")
exact_keys(item, {"concept_id", "vector"}, "concept row")
concept_id = item["concept_id"]
if (
not isinstance(concept_id, str) or not concept_id or len(concept_id) > 128 or
concept_id in concepts or concept_id not in normalized_thresholds
):
raise InputError("concept identifier is duplicate or not governed")
concepts[concept_id] = vector(item["vector"], dimension, "concept vector " + concept_id)
if set(concepts) != set(normalized_thresholds):
raise InputError("concept vectors and signed thresholds differ")
chunks: list[tuple[str, int, list[float]]] = []
seen_ids: set[str] = set()
seen_ranks: set[int] = set()
for item in retrieval["chunks"]:
if not isinstance(item, dict):
raise InputError("retrieved chunk must be an object")
exact_keys(item, {"chunk_id", "rank", "content_sha256", "vector"}, "retrieved chunk")
chunk_id = item["chunk_id"]
rank = integer(item["rank"], "chunk rank")
if (
not isinstance(chunk_id, str) or not chunk_id or len(chunk_id) > 256 or
chunk_id in seen_ids or rank in seen_ranks or not SHA256.fullmatch(str(item["content_sha256"]))
):
raise InputError("chunk identity, rank, or content digest is invalid")
seen_ids.add(chunk_id)
seen_ranks.add(rank)
chunks.append((chunk_id, rank, vector(item["vector"], dimension, "chunk vector " + chunk_id)))
if seen_ranks != set(range(expected_count)):
incomplete_reasons.append("retrieval ranks are not the complete ordered population")
base = {
"schema_version": "aidefend.d014001-retrieved-concept-method-result.v1",
"control": "AID-D-014.001",
"guidance": "AID-D-014.001-G002",
"request_id": retrieval["request_id"],
"retrieval_id": retrieval["retrieval_id"],
"asset_id": retrieval["asset_id"],
"asset_version": retrieval["asset_version"],
"route_id": retrieval["route_id"],
"policy_version": policy["policy_version"],
"embedding_model_id": policy["embedding_model_id"],
"embedding_model_revision": policy["embedding_model_revision"],
"expected_chunk_count": expected_count,
"observed_chunk_count": len(chunks),
"concept_count": len(concepts),
"input_sha256": {
"policy": hashlib.sha256(policy_raw).hexdigest(),
"retrieval": hashlib.sha256(retrieval_raw).hexdigest(),
"concepts": hashlib.sha256(concepts_raw).hexdigest(),
},
"evaluated_at": now.isoformat().replace("+00:00", "Z"),
}
if incomplete_reasons:
return {**base, "method_status": "INCOMPLETE", "reasons": sorted(set(incomplete_reasons)), "metrics": None}
finding_count = 0
finding_sample = []
score_digest = hashlib.sha256()
for chunk_id, rank, chunk_vector in sorted(chunks, key=lambda row: row[1]):
for concept_id in sorted(concepts):
score = cosine(chunk_vector, concepts[concept_id])
record = {"chunk_id": chunk_id, "rank": rank, "concept_id": concept_id, "score": score}
score_digest.update(json.dumps(record, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n")
threshold = normalized_thresholds[concept_id]
if score >= threshold:
finding_count += 1
if len(finding_sample) < HARD_FINDING_SAMPLE:
finding_sample.append({**record, "threshold": threshold})
status = "FINDING" if finding_count else "CLEAR"
return {
**base,
"method_status": status,
"reasons": [],
"metrics": {
"scored_chunks": len(chunks),
"scored_pairs": len(chunks) * len(concepts),
"finding_count": finding_count,
"finding_sample": finding_sample,
"score_records_sha256": score_digest.hexdigest(),
},
}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--policy", type=Path, required=True)
parser.add_argument("--retrieval", type=Path, required=True)
parser.add_argument("--concepts", type=Path, required=True)
parser.add_argument("--out", type=Path, required=True)
args = parser.parse_args()
inputs = {}
try:
inputs = {
"policy": read_stable(args.policy),
"retrieval": read_stable(args.retrieval),
"concepts": read_stable(args.concepts),
}
result = evaluate(inputs["policy"], inputs["retrieval"], inputs["concepts"])
except Exception as error:
result = {
"schema_version": "aidefend.d014001-retrieved-concept-method-result.v1",
"control": "AID-D-014.001",
"guidance": "AID-D-014.001-G002",
"method_status": "ERROR",
"reason": "input verification or semantic evaluation failed",
"error_class": type(error).__name__,
"input_sha256": {name: hashlib.sha256(raw).hexdigest() for name, raw in inputs.items()},
"evaluated_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
}
payload = (json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n").encode("utf-8")
descriptor = os.open(args.out, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o400)
try:
with os.fdopen(descriptor, "wb", closefd=False) as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
finally:
os.close(descriptor)
if args.out.read_bytes() != payload:
raise RuntimeError("result readback differs")
return {"CLEAR": 0, "FINDING": 1, "INCOMPLETE": 3, "ERROR": 2}[result["method_status"]]
if __name__ == "__main__":
raise SystemExit(main())
Verify exact inputs, sign the method result, and roll it into the leaf outcome
set -euo pipefail
umask 077
STAGE="$(mktemp -d)"
trap 'rm -rf -- "$STAGE"' EXIT HUP INT TERM
for NAME in policy retrieval concepts; do
test -f "evidence/$NAME.json" && test ! -L "evidence/$NAME.json"
test -f "evidence/$NAME.sigstore.json" && test ! -L "evidence/$NAME.sigstore.json"
install -m 0400 -- "evidence/$NAME.json" "$STAGE/$NAME.json"
install -m 0400 -- "evidence/$NAME.sigstore.json" "$STAGE/$NAME.sigstore.json"
done
cosign verify-blob --key keys/d014-policy.pub --bundle "$STAGE/policy.sigstore.json" "$STAGE/policy.json"
cosign verify-blob --key keys/retrieval-adapter.pub --bundle "$STAGE/retrieval.sigstore.json" "$STAGE/retrieval.json"
cosign verify-blob --key keys/d014-concepts.pub --bundle "$STAGE/concepts.sigstore.json" "$STAGE/concepts.json"
set +e
python detection/evaluate_retrieved_concepts.py \
--policy "$STAGE/policy.json" \
--retrieval "$STAGE/retrieval.json" \
--concepts "$STAGE/concepts.json" \
--out "$STAGE/method-result.json"
STATUS=$?
set -e
cosign sign-blob --yes --key env://D014_RESULT_SIGNING_KEY \
--bundle "$STAGE/method-result.sigstore.json" "$STAGE/method-result.json"
install -m 0400 -- "$STAGE/method-result.json" "$STAGE/method-result.sigstore.json" evidence/
exit "$STATUS"CLEAR, FINDING, INCOMPLETE, and ERROR are method results, not additional AIDEFEND control outcomes. The AID-D-014.001 evaluator reconciles this result with the content-classifier method over the same retrieval ID and complete ordered chunk population, then emits the leaf's single outcome. A finding never removes or blocks a chunk; the context-admission or isolation owner consumes the signed finding.
Negative fixtures must cover one missing and one extra rank, duplicate chunk and concept IDs, changed content digest, wrong asset or embedding revision, stale and future timestamps, incomplete collection, zero and non-finite vectors, threshold boundary equality, altered signed input, symlink input, oversized files, and result-publication failure. An independent read-only identity re-verifies all signatures and recomputes every chunk-by-concept score from the retained exact bytes.
Open-source tools
- Guardrails AI
- NVIDIA NeMo Guardrails
- Hugging Face Transformers
- NumPy
- Cosign
Source-available / open-weight tools
- Granite Guardian 4.1 (Apache-2.0; open-weight)
- Qwen3Guard (Apache-2.0; open-weight)
- gpt-oss-safeguard (Apache-2.0; open-weight)
- Meta Llama Prompt Guard 2 (Llama 4 Community License; open-weight)
- Meta Llama Guard 4 (Llama 4 Community License; open-weight)
Commercial / hosted tools
- Check Point AI Guardrails
- Palo Alto Networks Prisma AIRS
- Amazon Bedrock Guardrails
- Azure AI Content Safety - Prompt Shields
Threat mappings
MITRE ATLAS
- AML.T0099 AI Agent Tool Data Poisoning
- AML.T0070 RAG Poisoning
- AML.T0051 LLM Prompt Injection
- AML.T0066 Retrieval Content Crafting (scanning detects crafted retrieval content)
- AML.T0051.001 LLM Prompt Injection: Indirect
- AML.T0130 AI Agent Response Biasing (retrieved-chunk prompt detectors identify source-preference instructions before context assembly)
MAESTRO
- Compromised RAG Pipelines (L2)
- Data Poisoning (L2)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection
- LLM05:2026 Data and Model Poisoning
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI01:2026 Agent Goal Hijack
- ASI06:2026 Memory & Context Poisoning
NIST Adversarial Machine Learning 2025
- NISTAML.015 Indirect Prompt Injection
- NISTAML.027 Misaligned Outputs (knowledge-base poisoning and indirect injection)
Cisco Integrated AI Security and Safety Framework
- AISubtech-6.1.1 Knowledge Base Poisoning
- AITech-1.2 Indirect Prompt Injection
- AITech-4.2 Context Boundary Attacks (malicious instructions in retrieved content before context assembly)
- AITech-7.3 Data Source Abuse and Manipulation
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection (findings identify indirect instructions in retrieved content)
Databricks AI Security Framework 3.0
- Model Serving - Inference requests 9.9: Input Resource Control
- Raw Data 1.7: Lack of data trustworthiness
- Raw Data 1.11: Compromised 3rd-party datasets
- Agents - Tools MCP Server 13.24: Context Spoofing and Manipulation (post-retrieval scanning detects spoofed context)