AID-D-003.008: Generated Media & Binary Output Policy-Violation Detection
Inspect every policy-designated non-text output produced by a model or agent, including images, audio, video, documents, archives, serialized objects, and executable or other binary artifacts, at the post-generation and pre-release observation point. Reconcile the complete output manifest to the actual bytes; detect the real media type, container, codec, and bounded-decode coverage; run the policy-required media classifiers, malware scanners, and binary rules; and emit version-bound findings. Unknown types, unsupported codecs, encrypted or unparseable containers, truncation, timeouts, skipped members, and incomplete detector coverage must remain explicit and must not be reported as clean output.
Pillars: app, data
Lifecycle: validation, operation
Scope boundary
Owns policy-violation findings in model-generated images, audio, video, documents, archives, serialized objects, executables, and other non-text artifacts, including real-type detection, bounded decoding, malware or binary rules, and explicit incomplete-coverage states.
AID-D-003.001: AID-D-003.008 inspects non-text media and binary output; AID-D-003.001 inspects generated text, tokens, and text-bearing fields. A document or archive may require both paths after bounded extraction, but each detector owns its actual content type.
AID-D-001.002: AID-D-003.008 detects whether generated media violates content or binary policy; AID-D-001.002 performs authenticity and manipulation forensics on media signals. A synthetic image may be policy-compliant, and prohibited media may be authentic.
AID-D-007: AID-D-003.008 analyzes policy violations inside each generated artifact; AID-D-007 analyzes contradiction or hidden payload relationships across modalities and sessions. Artifact-local policy classification does not establish cross-modal consistency.
Implementation guidance
- AID-D-003.008-G001: Reconcile the complete output manifest to actual bytes and perform bounded content-type, container, and codec probing before any policy detector reports a clean result.
- AID-D-003.008-G002: Run every policy-required malware, binary-rule, and modality classifier for every applicable artifact and preserve detector abstention, skip, and error states.
- AID-D-003.008-G003: Reconcile the complete artifact and detector population before rolling up findings, while preserving incomplete coverage and detector failures.
AID-D-003.008-G001
Reconcile the complete output manifest to actual bytes and perform bounded content-type, container, and codec probing before any policy detector reports a clean result.
Complete population, private snapshots and bounded decoding
Run this adapter after generation and before release. The caller verifies the policy and complete manifest signature before calling it. Limits and allowed types/codecs come from that versioned policy; there are no universal security thresholds. Use a dedicated low-privilege parser worker with OS-enforced CPU, memory, process and temporary-volume quotas, no production credentials or unrestricted network, and read-only snapshot mounts for parser subprocesses. A successful MIME or codec probe alone never proves complete decoding.
This runnable adapter enumerates ZIP and ZIP-based Office documents, including nested members, into opaque private filenames; it never uses extractall or a member name as a filesystem destination. Encrypted, truncated, unsupported or over-limit members remain named gaps. PDF, OLE, RAR and other document/container formats need an additional bounded format adapter and remain insufficient here. Text-bearing units also feed AID-D-003.001 after the appropriate bounded text extraction.
from __future__ import annotations
import hashlib
import json
import math
import os
import re
import stat
import subprocess
import tempfile
import time
import zipfile
from contextlib import contextmanager
from fractions import Fraction
from pathlib import Path, PurePosixPath
class CoverageGap(ValueError):
pass
def run(command: list[str], timeout: float, output_limit: int) -> subprocess.CompletedProcess:
# The worker also needs OS-enforced CPU, memory and temporary-volume limits.
with tempfile.TemporaryFile() as stdout, tempfile.TemporaryFile() as stderr:
process = subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=stdout, stderr=stderr)
deadline = time.monotonic() + timeout
try:
while process.poll() is None:
if time.monotonic() >= deadline:
raise CoverageGap("parser_timeout")
if os.fstat(stdout.fileno()).st_size + os.fstat(stderr.fileno()).st_size > output_limit:
raise CoverageGap("parser_output_limit")
time.sleep(0.02)
if time.monotonic() >= deadline:
raise CoverageGap("parser_timeout")
if os.fstat(stdout.fileno()).st_size + os.fstat(stderr.fileno()).st_size > output_limit:
raise CoverageGap("parser_output_limit")
stdout.seek(0)
stderr.seek(0)
return subprocess.CompletedProcess(command, process.returncode, stdout.read().decode("utf-8", "strict"), stderr.read().decode("utf-8", "strict"))
finally:
if process.poll() is None:
process.kill()
process.wait()
def media_coverage(path: Path, media_type: str, policy: dict, timeout: float) -> dict:
deadline = time.monotonic() + timeout
probe = run([
"ffprobe", "-v", "error", "-show_format", "-show_streams", "-of", "json", str(path),
], timeout, policy["maximum_tool_output_bytes"])
if probe.returncode != 0 or probe.stderr.strip():
raise CoverageGap("media_probe_failed")
metadata = json.loads(probe.stdout)
streams = metadata.get("streams", [])
if not streams or len(streams) > policy["maximum_media_streams"]:
raise CoverageGap("media_stream_population_outside_policy")
for stream in streams:
if stream.get("codec_type") not in {"video", "audio"} or stream.get("codec_name") not in policy["allowed_codecs"]:
raise CoverageGap("unsupported_media_stream_or_codec")
if stream["codec_type"] == "video":
width, height = stream.get("width", 0), stream.get("height", 0)
if not (type(width) is int and type(height) is int and 0 < width <= policy["maximum_width"] and 0 < height <= policy["maximum_height"] and width * height <= policy["maximum_pixels"]):
raise CoverageGap("media_dimensions_outside_policy")
duration = stream.get("duration", metadata.get("format", {}).get("duration"))
if duration is not None:
seconds = float(duration)
if not math.isfinite(seconds) or seconds < 0 or seconds > policy["maximum_duration_seconds"]:
raise CoverageGap("media_duration_outside_policy")
# Decode to EOF, not a preview or a capped number of frames. A limit/error
# is incomplete coverage. framecrc emits one row per decoded output packet.
remaining = deadline - time.monotonic()
if remaining <= 0:
raise CoverageGap("media_decode_timeout")
decoded = run([
"ffmpeg", "-nostdin", "-v", "error", "-xerror", "-err_detect", "explode",
"-i", str(path), "-map", "0", "-c:v", "rawvideo", "-c:a", "pcm_s16le",
"-fps_mode", "passthrough", "-f", "framecrc", "-",
], remaining, policy["maximum_tool_output_bytes"])
if decoded.returncode != 0 or decoded.stderr.strip():
raise CoverageGap("media_decode_failed_or_truncated")
time_bases, bounds, counts = {}, {}, {}
total_bytes = 0
for line in decoded.stdout.splitlines():
match = re.fullmatch(r"#tb (\d+):\s*(\d+/\d+)", line.strip())
if match:
if int(match[2].split("/")[1]) == 0:
raise CoverageGap("invalid_decode_receipt")
time_bases[int(match[1])] = Fraction(match[2])
continue
if not line.strip() or line.startswith("#"):
continue
fields = [part.strip() for part in line.split(",")]
if len(fields) != 6:
raise CoverageGap("invalid_decode_receipt")
index, _, pts, duration, size = map(int, fields[:5])
if index not in time_bases or duration < 0 or size < 0:
raise CoverageGap("invalid_decode_receipt")
counts[index] = counts.get(index, 0) + 1
total_bytes += size
start, end = pts * time_bases[index], (pts + duration) * time_bases[index]
previous = bounds.get(index, (start, end))
bounds[index] = (min(previous[0], start), max(previous[1], end))
if total_bytes > policy["maximum_decoded_bytes"] or sum(counts.values()) > policy["maximum_decoded_packets"]:
raise CoverageGap("decoded_population_outside_policy")
if not media_type.startswith("image/") and bounds[index][1] - bounds[index][0] > policy["maximum_duration_seconds"]:
raise CoverageGap("decoded_duration_outside_policy")
if set(counts) != set(range(len(streams))):
raise CoverageGap("not_every_media_stream_decoded")
return {"metadata": metadata, "decoded_packets": counts, "decoded_bytes": total_bytes, "decode_status": "COMPLETE"}
@contextmanager
def probe_population(root: Path, manifest: dict, policy: dict):
"""Keep the private byte snapshots alive through scanning and roll-up."""
numeric = ["maximum_artifact_bytes", "maximum_member_bytes", "maximum_expanded_bytes", "maximum_members", "maximum_depth", "maximum_expansion_ratio", "maximum_width", "maximum_height", "maximum_pixels", "maximum_media_streams", "maximum_duration_seconds", "maximum_decoded_bytes", "maximum_decoded_packets", "maximum_tool_output_bytes", "population_timeout_seconds", "probe_timeout_seconds"]
if any(isinstance(policy[key], bool) or not isinstance(policy[key], (int, float)) or not math.isfinite(policy[key]) or policy[key] <= 0 for key in numeric):
raise ValueError("signed probe limits must be finite and positive")
root = root.resolve(strict=True)
if set(manifest) != {"schema_version", "manifest_id", "expected_count", "artifacts"} or manifest["schema_version"] != "aidefend.generated-output-manifest.v1":
raise ValueError("output manifest schema mismatch")
artifacts = manifest["artifacts"]
if not isinstance(artifacts, list) or type(manifest["expected_count"]) is not int or manifest["expected_count"] != len(artifacts):
raise ValueError("manifest population count mismatch")
for item in artifacts:
if not isinstance(item, dict) or set(item) != {"artifact_id", "relative_path", "sha256", "declared_media_type"} or not isinstance(item["artifact_id"], str) or not item["artifact_id"] or not re.fullmatch(r"[0-9a-f]{64}", item["sha256"]):
raise ValueError("artifact manifest entry schema mismatch")
ids, paths = [item["artifact_id"] for item in artifacts], [item["relative_path"] for item in artifacts]
if len(ids) != len(set(ids)) or any(not isinstance(path, str) or not path or PurePosixPath(path).is_absolute() or ".." in PurePosixPath(path).parts or "\\" in path for path in paths) or len(paths) != len(set(paths)):
raise ValueError("manifest identity or path missing, unsafe or duplicated")
discovered = sorted(path.relative_to(root).as_posix() for path in root.rglob("*") if path.is_file())
if sorted(paths) != discovered:
raise ValueError("manifest does not cover the complete output directory")
units, member_count, expanded_bytes = [], 0, 0
deadline = time.monotonic() + policy["population_timeout_seconds"]
with tempfile.TemporaryDirectory(prefix="aidefend-probe-") as directory_name:
directory = Path(directory_name)
directory.chmod(0o700)
def remaining():
seconds = min(policy["probe_timeout_seconds"], deadline - time.monotonic())
if seconds <= 0:
raise CoverageGap("population_timeout")
return seconds
def copy_bytes(handle, maximum, is_member=False):
nonlocal expanded_bytes
target = directory / str(len(list(directory.iterdir())))
digest, size = hashlib.sha256(), 0
with target.open("xb") as out:
while chunk := handle.read(65536):
remaining()
size += len(chunk)
if is_member:
expanded_bytes += len(chunk)
if size > maximum or expanded_bytes > policy["maximum_expanded_bytes"]:
raise CoverageGap("decompression_or_size_limit")
digest.update(chunk)
out.write(chunk)
out.flush()
os.fsync(out.fileno())
target.chmod(0o400)
return target, size, digest.hexdigest()
def inspect(path, unit, depth):
nonlocal member_count
process = run(["java", "-jar", policy["tika_app_jar"], "--detect", str(path)], remaining(), policy["maximum_tool_output_bytes"])
detected = process.stdout.strip() if process.returncode == 0 else ""
unit["detected_media_type"] = detected
if not detected or "\n" in detected or detected not in policy["allowed_media_types"] or detected == "application/octet-stream":
raise CoverageGap("unknown_or_unsupported_media_type")
if detected.startswith(("image/", "audio/", "video/")):
unit["media"] = media_coverage(path, detected, policy, remaining())
remaining()
elif zipfile.is_zipfile(path):
if depth >= policy["maximum_depth"]:
raise CoverageGap("container_depth_limit")
with zipfile.ZipFile(path) as archive:
names = set()
first_child = len(units)
unit["members"] = []
for ordinal, info in enumerate(archive.infolist()):
remaining()
member_count += 1
if member_count > policy["maximum_members"]:
raise CoverageGap("container_member_limit")
name = PurePosixPath(info.filename)
mode = info.external_attr >> 16
if info.filename in names or name.is_absolute() or ".." in name.parts or "\\" in info.filename or stat.S_ISLNK(mode):
raise CoverageGap("unsafe_or_duplicate_container_member")
names.add(info.filename)
unit["members"].append({"ordinal": ordinal, "name": info.filename, "directory": info.is_dir(), "declared_bytes": info.file_size, "compressed_bytes": info.compress_size})
if info.is_dir():
continue
child = {"artifact_id": json.dumps([unit["artifact_id"], ordinal], separators=(",", ":")), "root_artifact_id": unit["root_artifact_id"], "root_sha256": unit["root_sha256"], "member_name": info.filename, "status": "INSUFFICIENT_DATA"}
units.append(child)
try:
if info.flag_bits & 1:
raise CoverageGap("encrypted_member")
if info.file_size > policy["maximum_member_bytes"] or info.file_size > max(1, info.compress_size) * policy["maximum_expansion_ratio"]:
raise CoverageGap("member_size_or_expansion_ratio_limit")
with archive.open(info) as handle:
snapshot, size, digest = copy_bytes(handle, policy["maximum_member_bytes"], True)
if size != info.file_size:
raise CoverageGap("member_size_mismatch")
child.update(snapshot_path=str(snapshot), sha256=digest, bytes=size)
inspect(snapshot, child, depth + 1)
child["status"] = "PASS"
except (ValueError, OSError, RuntimeError, zipfile.BadZipFile) as error:
child["reason"] = str(error)
if any(child["status"] != "PASS" for child in units[first_child:]):
raise CoverageGap("container_member_coverage_incomplete")
elif detected in {"application/x-executable", "application/x-sharedlib", "application/x-dosexec", "application/vnd.microsoft.portable-executable"} or detected.startswith("text/") or detected in {"application/xml", "application/json"}:
pass
else:
# ZIP/OOXML and media are implemented here. Other document/container
# formats require a bounded adapter; type detection alone is not PASS.
raise CoverageGap("bounded_document_or_container_adapter_required")
for item in artifacts:
unit = {"artifact_id": json.dumps([item["artifact_id"]], separators=(",", ":")), "root_artifact_id": item["artifact_id"], "root_sha256": item["sha256"], "status": "INSUFFICIENT_DATA"}
units.append(unit)
try:
path = root / item["relative_path"]
if path.is_symlink() or root not in path.resolve(strict=True).parents:
raise CoverageGap("artifact_symlink_or_escape")
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
with os.fdopen(descriptor, "rb") as handle:
if not stat.S_ISREG(os.fstat(handle.fileno()).st_mode):
raise CoverageGap("artifact_not_regular")
snapshot, size, digest = copy_bytes(handle, policy["maximum_artifact_bytes"])
if digest != item["sha256"]:
raise CoverageGap("artifact_digest_mismatch")
if size == 0:
raise CoverageGap("empty_artifact")
unit.update(snapshot_path=str(snapshot), sha256=digest, bytes=size)
inspect(snapshot, unit, 0)
unit["status"] = "PASS"
except (ValueError, OSError, RuntimeError, zipfile.BadZipFile) as error:
unit["reason"] = str(error)
yield {"manifest_id": manifest["manifest_id"], "policy_version": policy["policy_version"], "probe_revision": policy["probe_revision"], "expected_count": len(artifacts), "measured_count": len(artifacts), "status": "PASS" if all(unit["status"] == "PASS" for unit in units) else "INSUFFICIENT_DATA", "units": units}
Connect the probe and scanners
Save the code as generated_artifact_probe.py. Supply policy_version, probe_revision, tika_app_jar, allowed_media_types, allowed_codecs and every named limit. Keep the with probe_population(root, manifest, policy) as population: context open until scanning and aggregation finish. Scan each unit with a snapshot, including each extracted member; preserve all root and member probe gaps. Do not build the evidence population by filtering only successful probes. Snapshots are ephemeral staging, not independent evidence storage.
Verification: Replay missing/extra files, duplicate artifact IDs, MIME spoofing, symlink escape and changed digests; benign ZIP/Office ZIP and nested archives; traversal, duplicate, encrypted, truncated, oversized, excessive-depth and high-expansion members; huge dimensions/duration, missing streams, decoder failure, output limits and timeout. Confirm incomplete cases cannot produce PASS and every readable member is scanned.
AID-D-003.008-G002
Run every policy-required malware, binary-rule, and modality classifier for every applicable artifact and preserve detector abstention, skip, and error states.
One scan per immutable root or member
Save G001 as generated_artifact_probe.py in the same package. Pass its unit ID, digest, detected type, probe revision and coverage status as a VerifiedProbe, while its private snapshot remains available. Invoke this adapter for each root and each bounded extracted member. ClamAV recursion does not replace member-level YARA rules. ClamAV and YARA are this published example’s local scanner choices; another implementation may use different policy-required detectors with the same coverage and byte-binding properties.
Verified daemon configuration and non-clean states
Enable AlertEncrypted yes and AlertExceedsMax yes in the running ClamAV daemon. Set clamav_coverage_alerts_verified only after trusted supervisor readback binds the actual daemon, active configuration and database version; an edited but unused config file or caller assertion is not enough. Without that verification an OK response remains insufficient. Encrypted/limit heuristic alerts describe coverage gaps, not proof of malware; keep a separate positive finding if one is also reported. An encrypted-content policy violation may be recorded separately without removing the coverage gap.
from __future__ import annotations
import hashlib
import math
import os
import re
import stat
import subprocess
import tempfile
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
# Save G001's code as generated_artifact_probe.py in the same package.
from generated_artifact_probe import CoverageGap, run
@dataclass(frozen=True)
class VerifiedProbe:
manifest_id: str
artifact_id: str
artifact_sha256: str
detected_media_type: str
probe_revision: str
coverage_status: str
@dataclass(frozen=True)
class ScannerPolicy:
policy_version: str
yara_rules_path: str
yara_rules_sha256: str
maximum_artifact_bytes: int
maximum_yara_rules_bytes: int
maximum_tool_output_bytes: int
timeout_seconds: float
clamav_version: str
clamav_database_version: str
yara_version: str
moderation_policy_version: str | None
moderation_detector_version: str | None
# Trusted supervisor readback of the RUNNING daemon, not a caller assertion
# or an unused configuration file: AlertEncrypted and AlertExceedsMax enabled.
clamav_coverage_alerts_verified: bool
def _snapshot(path: Path, expected_sha256: str, maximum: int, directory: Path, name: str) -> Path:
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
target, size, digest = directory / name, 0, hashlib.sha256()
with os.fdopen(descriptor, "rb") as source, target.open("xb") as out:
if path.is_symlink() or not stat.S_ISREG(os.fstat(source.fileno()).st_mode):
raise ValueError("snapshot source is not a regular file")
while chunk := source.read(65536):
size += len(chunk)
if size > maximum:
raise ValueError("snapshot source exceeds its signed bound")
digest.update(chunk)
out.write(chunk)
out.flush()
os.fsync(out.fileno())
if digest.hexdigest() != expected_sha256:
raise ValueError("snapshot bytes differ from the verified binding")
target.chmod(0o400)
return target
def clamav_result(process: subprocess.CompletedProcess, alerts_verified: bool) -> dict:
signatures = [line.rsplit(": ", 1)[-1][:-6] for line in process.stdout.splitlines() if line.endswith(" FOUND")]
limited = [name for name in signatures if name.startswith(("Heuristics.Encrypted", "Heuristics.Limits.Exceeded"))]
findings = [name for name in signatures if name not in limited]
if limited:
return {"control_status": "INSUFFICIENT_DATA", "finding_status": "FINDING" if findings else "UNKNOWN", "coverage_reasons": limited, "signatures": findings}
if process.returncode == 0 and not signatures and not process.stderr.strip() and process.stdout.rstrip().endswith(": OK"):
return {"control_status": "PASS" if alerts_verified else "INSUFFICIENT_DATA", "finding_status": "NO_FINDING" if alerts_verified else "UNKNOWN"}
if process.returncode == 1 and signatures and not process.stderr.strip():
return {"control_status": "PASS" if alerts_verified else "INSUFFICIENT_DATA", "finding_status": "FINDING"}
return {"control_status": "ERROR", "finding_status": "FINDING" if findings else "UNKNOWN"}
def scan_artifact(*, ingress_path: Path, probe: VerifiedProbe, policy: ScannerPolicy,
moderate_media: Callable[[Path, str, str, str], bytes],
verify_moderation_receipt: Callable[[bytes], dict]) -> list[dict]:
# One invocation per G001 unit, including each bounded extracted member.
# A ZIP's outer-byte YARA result never covers its decompressed members.
if not all((probe.manifest_id, probe.artifact_id, probe.detected_media_type, probe.probe_revision, policy.policy_version, policy.clamav_version, policy.clamav_database_version, policy.yara_version)) or not re.fullmatch(r"[0-9a-f]{64}", probe.artifact_sha256) or probe.coverage_status not in {"PASS", "INSUFFICIENT_DATA", "ERROR"}:
raise ValueError("verified scanner bindings are incomplete")
if any(type(value) is not int or value <= 0 for value in (policy.maximum_artifact_bytes, policy.maximum_yara_rules_bytes, policy.maximum_tool_output_bytes)) or isinstance(policy.timeout_seconds, bool) or not math.isfinite(policy.timeout_seconds) or policy.timeout_seconds <= 0 or type(policy.clamav_coverage_alerts_verified) is not bool:
raise ValueError("signed scanner policy is invalid")
bindings = {"manifest_id": probe.manifest_id, "artifact_id": probe.artifact_id, "artifact_sha256": probe.artifact_sha256, "probe_revision": probe.probe_revision, "policy_version": policy.policy_version}
results = []
with tempfile.TemporaryDirectory(prefix="aidefend-scan-") as temporary:
directory = Path(temporary)
directory.chmod(0o700)
artifact = _snapshot(ingress_path, probe.artifact_sha256, policy.maximum_artifact_bytes, directory, "artifact.bin")
rules = _snapshot(Path(policy.yara_rules_path), policy.yara_rules_sha256, policy.maximum_yara_rules_bytes, directory, "rules.yar")
detectors = [
("clamav", policy.clamav_version, policy.clamav_database_version, ["clamdscan", "--fdpass", "--no-summary", str(artifact)]),
("yara", policy.yara_version, policy.yara_rules_sha256, ["yara", "--fail-on-warnings", str(rules), str(artifact)]),
]
for detector, version, rule_version, command in detectors:
try:
process = run(command, policy.timeout_seconds, policy.maximum_tool_output_bytes)
if detector == "clamav":
result = clamav_result(process, policy.clamav_coverage_alerts_verified)
elif process.returncode == 0:
matches = sorted({line.split(maxsplit=1)[0] for line in process.stdout.splitlines() if line.strip()})
result = {"control_status": "PASS", "finding_status": "FINDING" if matches else "NO_FINDING", "matches": matches}
else:
result = {"control_status": "ERROR", "finding_status": "UNKNOWN"}
except (CoverageGap, OSError, UnicodeError) as error:
result = {"control_status": "INSUFFICIENT_DATA", "finding_status": "UNKNOWN", "coverage_reason": str(error)}
if probe.coverage_status != "PASS":
result["control_status"] = "INSUFFICIENT_DATA"
result["coverage_reason"] = "probe_coverage_incomplete"
if result["finding_status"] != "FINDING":
result["finding_status"] = "UNKNOWN"
results.append({**bindings, "detector": detector, "detector_version": version, "rule_version": rule_version, **result})
if probe.detected_media_type.startswith(("image/", "audio/", "video/")):
result = {"control_status": "INSUFFICIENT_DATA", "finding_status": "UNKNOWN"}
if policy.moderation_policy_version and policy.moderation_detector_version:
try:
receipt = verify_moderation_receipt(moderate_media(artifact, probe.artifact_sha256, probe.detected_media_type, policy.moderation_policy_version))
expected = {"detector": "media_moderation", "detector_version": policy.moderation_detector_version, "policy_version": policy.moderation_policy_version, "artifact_sha256": probe.artifact_sha256}
if any(receipt.get(key) != value for key, value in expected.items()) or receipt.get("control_status") not in {"PASS", "INSUFFICIENT_DATA", "ERROR"} or receipt.get("finding_status") not in {"FINDING", "NO_FINDING", "UNKNOWN"} or (receipt["control_status"] == "PASS" and receipt["finding_status"] == "UNKNOWN") or (receipt["control_status"] != "PASS" and receipt["finding_status"] == "NO_FINDING"):
raise ValueError("provider receipt malformed or misbound")
result = {key: receipt[key] for key in ("control_status", "finding_status")}
result["categories"] = receipt.get("categories", [])
except (ValueError, OSError, TimeoutError) as error:
result = {"control_status": "ERROR", "finding_status": "UNKNOWN", "coverage_reason": str(error)}
if probe.coverage_status != "PASS":
result["control_status"] = "INSUFFICIENT_DATA"
if result["finding_status"] != "FINDING":
result["finding_status"] = "UNKNOWN"
results.append({**bindings, "detector": "media_moderation", "detector_version": policy.moderation_detector_version, "rule_version": policy.moderation_policy_version, **result})
return results
Verification and operating limits
Run scanners and provider adapters under a bounded low-privilege runtime profile. Verify deployed scanner/database versions against the supplied policy, and bound a provider call’s time/output in its adapter. Wrong-digest snapshots or invalid configuration are visible errors, never discarded. Root/member units without readable bytes retain their probe gap; record failures and reconcile them in G003. Replay clean files with alerts enabled and unverified, encryption/limit alerts, a simultaneous malware finding, benign and matched YARA members inside ZIP/Office ZIP, scanner failure/timeout, wrong-digest or wrong-version provider receipts, and source replacement. Results and findings always refer to the same immutable unit digest.
AID-D-003.008-G003
Reconcile the complete artifact and detector population before rolling up findings, while preserving incomplete coverage and detector failures.
Population and binding validation before roll-up
Build one evidence record per manifest from an independently verified complete manifest and policy. The expected mapping contains one binding for every required detector/root/member pair, including probe receipts, and supplies the manifest ID, unit ID, digest, probe revision, policy version, detector version and rule/database version. It must not be derived from the incoming receipts. Include a probe receipt for every manifest root even if probing fails; unreadable members stay named gaps in that root’s probe evidence. Never silently omit a failed root or manufacture a member digest.
Only byte-for-byte equivalent logical receipts may be deduplicated. Unexpected artifacts/detectors, conflicting receipts, malformed states or a different digest/version prevent PASS. Findings and completeness are separate: a known finding can coexist with incomplete observation. Only valid bound receipts contribute factual findings; rejected receipts remain visible as integrity errors.
import json
import re
def roll_up(manifest_id: str, expected: dict[tuple[str, str], dict], receipts: list[dict]) -> dict:
# Build expected from the independently verified complete G001 manifest and
# policy, NEVER from the receipts being checked. It includes root/member IDs.
binding_fields = {"manifest_id", "artifact_id", "artifact_sha256", "probe_revision", "policy_version", "detector", "detector_version", "rule_version"}
for key, binding in expected.items():
if set(binding) != binding_fields or key != (binding["artifact_id"], binding["detector"]) or binding["manifest_id"] != manifest_id or any(not isinstance(value, str) or not value for value in binding.values()) or not re.fullmatch(r"[0-9a-f]{64}", binding["artifact_sha256"]):
raise ValueError("expected population binding is incomplete")
accepted, invalid, unexpected, conflicts, replays = {}, [], [], [], 0
for receipt in receipts:
if not isinstance(receipt, dict):
invalid.append({"reason": "malformed_receipt"})
continue
key = (receipt.get("artifact_id"), receipt.get("detector"))
if not all(isinstance(value, str) for value in key):
invalid.append({"reason": "missing_receipt_identity", "receipt": receipt})
continue
if key not in expected:
unexpected.append(receipt)
continue
if any(receipt.get(field) != value for field, value in expected[key].items()):
invalid.append({"reason": "receipt_binding_mismatch", "receipt": receipt})
continue
control, finding = receipt.get("control_status"), receipt.get("finding_status")
if control not in {"PASS", "FAIL", "INSUFFICIENT_DATA", "ERROR"} or finding not in {"FINDING", "NO_FINDING", "UNKNOWN"} or (control == "PASS" and finding == "UNKNOWN") or (control == "FAIL" and finding != "FINDING") or (control in {"ERROR", "INSUFFICIENT_DATA"} and finding == "NO_FINDING"):
invalid.append({"reason": "invalid_receipt_status", "receipt": receipt})
continue
if key in accepted:
if json.dumps(accepted[key], sort_keys=True) == json.dumps(receipt, sort_keys=True):
replays += 1
else:
conflicts.append(receipt)
else:
accepted[key] = receipt
missing = sorted(set(expected) - set(accepted))
valid = list(accepted.values())
# Keep findings from conflicting but individually bound receipts, too.
findings = [item for item in valid + conflicts if item["finding_status"] == "FINDING"]
errors = [item for item in valid if item["control_status"] == "ERROR"]
insufficient = [item for item in valid if item["control_status"] == "INSUFFICIENT_DATA"]
incomplete = bool(missing or invalid or unexpected or conflicts or errors or insufficient)
outcome = "FAIL" if findings else "ERROR" if (invalid or unexpected or conflicts or errors) else "INSUFFICIENT_DATA" if incomplete else "PASS" if expected else "NOT_APPLICABLE"
return {
"schema_version": "aidefend.generated-output-evidence.v2",
"control": "AID-D-003.008", "manifest_id": manifest_id,
"outcome": outcome,
"coverage_status": "INCOMPLETE" if incomplete else "COMPLETE",
"finding_status": "FINDING" if findings else "NO_FINDING" if outcome == "PASS" else "UNKNOWN",
"expected_detector_population": sorted(expected),
"observed_detector_population": sorted(accepted),
"missing_detector_population": missing,
"finding_receipts": findings, "error_receipts": errors,
"insufficient_receipts": insufficient, "invalid_receipts": invalid,
"unexpected_receipts": unexpected, "conflicting_receipts": conflicts,
"identical_replays": replays, "receipts": valid,
}
Independent verification: A verifier outside generation and scanner identities reads back the immutable manifest, complete probe population, policy, version bindings and receipts from evidence storage, rebuilds the expected population and recomputes this result. Replay a policy-selected sample and every finding, error, abstention and unsupported case. Test identical replays, duplicate conflicts, extra artifact receipts, missing results, wrong hashes/versions and findings alongside gaps. The v2 evidence record carries explicit completeness separately from findings. These are detection results; AID-H-006.004 owns holding and release. After-delivery checks identify the scanned copy and their timing; without independently established delivery identity they do not prove what was delivered or protection before delivery.
Open-source tools
- FFmpeg
- Apache Tika
- ClamAV
- YARA
Commercial / hosted tools
- Azure AI Content Safety
- Hive Visual Moderation API
- Google Cloud Vision SafeSearch Detection
- VirusTotal Private Scanning
Threat mappings
MITRE ATLAS
- AML.T0048 External Harms
- AML.T0067 LLM Trusted Output Components Manipulation
- AML.T0123 Obfuscated Files or Information (real media-type and container detection exposes disguised binary outputs)
MAESTRO
- N/A
OWASP LLM Top 10 2026
- LLM10:2026 Improper Output Handling
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI05:2026 Unexpected Code Execution (RCE)
NIST Adversarial Machine Learning 2025
- NISTAML.027 Misaligned Outputs
Cisco Integrated AI Security and Safety Framework
- AISubtech-12.2.1 Code Detection / Malicious Code Output (known malware and payload signatures in generated binary or media artifacts)
- AISubtech-15.1.1 Cybersecurity and Hacking: Malware / Exploits (known malware and payload signatures in generated binary or media artifacts)
- AISubtech-15.1.25 General Harms (known malware and payload signatures in generated binary or media artifacts)
- AITech-12.2 Insecure Output Handling (known malware and payload signatures in generated binary or media artifacts)
- AITech-15.1 Harmful Content (known malware and payload signatures in generated binary or media artifacts)
Google Secure AI Framework 2.0 - Risks
- IMO: Insecure Model Output
Databricks AI Security Framework 3.0
- Model Serving - Inference response 10.1: Lack of audit and monitoring inference quality
- Model Serving - Inference response 10.2: Output manipulation
- Agents - Tools MCP Client 13.32: Client-Side Code Execution