AID-D-007.001: Batch Cross-Modal Consistency & Artifact Anomaly Detection
Pillars: data, model
Lifecycle: operation
Scope boundary
Owns cross-modal semantic contradiction, hidden payload, expert disagreement, attention anomaly, and out-of-context modality findings for complete non-realtime files and artifact batches. It does not sanitize or admit the artifact.
AID-D-007.002: AID-D-007.001 analyzes complete artifacts that can be decoded, rescanned, and replayed; AID-D-007.002 analyzes ephemeral realtime streams where timing, incremental deltas, source changes, and action timing carry security meaning. A recorded stream may support later batch analysis, but it cannot fully reconstruct every live decision-time state.
AID-D-001.002: AID-D-007.001 owns cross-modal consistency and hidden-payload findings; AID-D-001.002 owns forensic authenticity and synthetic-media analysis within the signal. A deepfake detector and a cross-modal contradiction detector answer different forensic questions.
AID-H-002.003: AID-D-007.001 emits findings for complete multimodal artifacts; AID-H-002.003 sanitizes and makes the admission decision for those artifacts. Detection describes the risk, while Harden transforms, rejects, or admits the input.
Implementation guidance
- AID-D-007.001-G001: Implement semantic consistency checks between information extracted from different modalities (e.g., compare text captions and image content; detect when audio commands contradict visual cues) and emit structured inconsistency findings for downstream admission, review, or hardening controls.
- AID-D-007.001-G002: Scan non-primary modalities for embedded instructions or payloads intended for other modalities (e.g., steganographically hidden text in images, QR codes containing malicious prompts, audio watermarks with commands).
- AID-D-007.001-G003: Aggregate modality-specific detector findings and preprocessing facts into one evidence record without creating an admission decision.
- AID-D-007.001-G004: Monitor the AI model's internal attention mechanisms (if accessible and interpretable) for unusual or forced cross-modal attention patterns that might indicate manipulation.
- AID-D-007.001-G005: Develop and maintain a library of known cross-modal attack patterns and use this knowledge to inform detection rules and defensive transformations.
- AID-D-007.001-G006: Score multimodal output faithfulness and emit a finding when the generated response appears dominated by one compromised modality.
- AID-D-007.001-G007: Detect expert-model disagreement across modalities and publish the discrepancy as a multimodal risk signal.
- AID-D-007.001-G008: Detect out-of-context modalities for task-specific workflows and emit context-mismatch findings.
AID-D-007.001-G001
Production boundary
Run pinned modality adapters in isolated workers and pass only authenticated observation records into this pre-fusion detector. The signed policy declares the complete modality-pair population, exact adapter and ontology digests, and calibrated thresholds. Missing pairs or excessive uncertainty are INSUFFICIENT_DATA; malformed, extra, duplicate, or misbound records are ERROR; a measured contradiction is FAIL; only a complete measured population with no finding is PASS. This detector emits evidence and never admits, blocks, or rewrites content.
Strict pair-population evaluator
# File: multimodal/score_semantic_consistency.py
from __future__ import annotations
import argparse
import hashlib
import json
import math
from pathlib import Path
PAIR_FIELDS = {"pair_id", "left_id", "right_id", "adapter_id", "adapter_sha256", "ontology_sha256"}
OBJECT_FIELDS = {"modality_id", "media_type", "content_sha256"}
OBS_FIELDS = {"request_id", "pair_id", "left_sha256", "right_sha256", "adapter_id", "adapter_sha256", "ontology_sha256", "scores", "uncertainty"}
SCORE_FIELDS = {"entailment", "contradiction", "similarity"}
def strict_json(raw: bytes) -> object:
def unique(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate_json_key")
value[key] = item
return value
return json.loads(raw.decode("utf-8", errors="strict"), object_pairs_hook=unique,
parse_constant=lambda value: (_ for _ in ()).throw(ValueError("nonfinite_json")))
def exact(value: object, fields: set[str], label: str) -> dict:
if not isinstance(value, dict) or set(value) != fields:
raise ValueError(label + "_schema")
return value
def unit(value: object, label: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(label + "_type")
number = float(value)
if not math.isfinite(number) or not 0.0 <= number <= 1.0:
raise ValueError(label + "_range")
return number
def evaluate(policy: dict, ledger: dict, rows: object) -> dict:
exact(policy, {"schema_version", "policy_version", "required_pairs", "thresholds"}, "policy")
exact(ledger, {"schema_version", "request_id", "asset_id", "request_sha256", "modality_objects"}, "ledger")
if policy["schema_version"] != "aidefend.cross-modal-policy.v1" or ledger["schema_version"] != "aidefend.multimodal-request-ledger.v1":
raise ValueError("schema_version")
thresholds = exact(policy["thresholds"], {"contradiction_min", "similarity_min", "max_uncertainty"}, "thresholds")
contradiction_min = unit(thresholds["contradiction_min"], "contradiction_min")
similarity_min = unit(thresholds["similarity_min"], "similarity_min")
max_uncertainty = unit(thresholds["max_uncertainty"], "max_uncertainty")
objects = {}
if not isinstance(ledger["modality_objects"], list) or not ledger["modality_objects"]:
raise ValueError("empty_modality_population")
for item in ledger["modality_objects"]:
item = exact(item, OBJECT_FIELDS, "modality")
identity = item["modality_id"]
if not isinstance(identity, str) or not identity or identity in objects:
raise ValueError("modality_identity")
if not isinstance(item["content_sha256"], str) or len(item["content_sha256"]) != 64:
raise ValueError("modality_digest")
objects[identity] = item
expected = {}
if not isinstance(policy["required_pairs"], list) or not policy["required_pairs"]:
raise ValueError("empty_pair_population")
for pair in policy["required_pairs"]:
pair = exact(pair, PAIR_FIELDS, "pair")
pair_id = pair["pair_id"]
if not isinstance(pair_id, str) or not pair_id or pair_id in expected:
raise ValueError("pair_identity")
if pair["left_id"] not in objects or pair["right_id"] not in objects:
raise ValueError("pair_modality")
if any(not isinstance(pair[name], str) or len(pair[name]) != 64 for name in ("adapter_sha256", "ontology_sha256")):
raise ValueError("pair_digest")
expected[pair_id] = pair
if not isinstance(rows, list):
raise ValueError("observation_population")
measured = {}
for row in rows:
row = exact(row, OBS_FIELDS, "observation")
pair_id = row["pair_id"]
if pair_id in measured or pair_id not in expected:
raise ValueError("duplicate_or_extra_pair")
pair = expected[pair_id]
if (row["request_id"] != ledger["request_id"]
or row["left_sha256"] != objects[pair["left_id"]]["content_sha256"]
or row["right_sha256"] != objects[pair["right_id"]]["content_sha256"]
or any(row[name] != pair[name] for name in ("adapter_id", "adapter_sha256", "ontology_sha256"))):
raise ValueError("observation_binding")
scores = exact(row["scores"], SCORE_FIELDS, "scores")
measured[pair_id] = {"scores": {name: unit(scores[name], name) for name in sorted(SCORE_FIELDS)},
"uncertainty": unit(row["uncertainty"], "uncertainty")}
missing = sorted(set(expected) - set(measured))
uncertain = sorted(pair_id for pair_id, row in measured.items() if row["uncertainty"] > max_uncertainty)
findings = sorted(pair_id for pair_id, row in measured.items()
if pair_id not in uncertain and (row["scores"]["contradiction"] >= contradiction_min
or row["scores"]["similarity"] < similarity_min))
outcome = "FAIL" if findings else "INSUFFICIENT_DATA" if missing or uncertain else "PASS"
return {"schema_version": "aidefend.cross-modal-finding.v1", "control_id": "AID-D-007.001",
"request_id": ledger["request_id"], "asset_id": ledger["asset_id"],
"request_sha256": ledger["request_sha256"], "policy_version": policy["policy_version"],
"required_pair_ids": sorted(expected), "measured_pair_ids": sorted(measured),
"missing_pair_ids": missing, "uncertain_pair_ids": uncertain,
"finding_pair_ids": findings, "score_vector_by_pair": measured, "outcome": outcome}
def main() -> None:
parser = argparse.ArgumentParser()
for name in ("policy", "ledger", "observations", "out"):
parser.add_argument("--" + name, type=Path, required=True)
args = parser.parse_args()
raw = {name: getattr(args, name).read_bytes() for name in ("policy", "ledger", "observations")}
try:
result = evaluate(strict_json(raw["policy"]), strict_json(raw["ledger"]), strict_json(raw["observations"]))
except Exception as error:
result = {"schema_version": "aidefend.cross-modal-finding.v1", "control_id": "AID-D-007.001",
"input_sha256": {name: hashlib.sha256(value).hexdigest() for name, value in sorted(raw.items())},
"outcome": "ERROR", "reason_code": type(error).__name__}
temporary = args.out.with_suffix(args.out.suffix + ".tmp")
temporary.write_text(json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False) + chr(10), encoding="utf-8")
temporary.replace(args.out)
if result["outcome"] == "ERROR":
raise SystemExit(2)
if __name__ == "__main__":
main()Authenticated execution and evidence
set -euo pipefail
umask 077
cosign verify-blob --key keys/multimodal-policy.pub --bundle evidence/policy.bundle evidence/policy.json
cosign verify-blob --key keys/request-ledger.pub --bundle evidence/ledger.bundle evidence/ledger.json
cosign verify-blob --key keys/adapter-observations.pub --bundle evidence/observations.bundle evidence/observations.json
set +e
python multimodal/score_semantic_consistency.py --policy evidence/policy.json --ledger evidence/ledger.json --observations evidence/observations.json --out evidence/cross-modal-finding.json
STATUS=$?
set -e
cosign sign-blob --yes --key env://D007001_SIGNING_KEY --bundle evidence/cross-modal-finding.bundle evidence/cross-modal-finding.json
exit "$STATUS"Stage private immutable snapshots after signature verification so there is no verifier-then-reread race. A separately credentialed runner recomputes adapter observations from the exact modality bytes and replays consistent, contradictory, ambiguous, missing, extra, duplicate, adapter-drift, ontology-drift, duplicate-key, and non-finite fixtures. Reconcile every eligible request in the authoritative ingestion ledger; threshold values come only from the signed deployment policy.
AID-D-007.001-G002
Production implementation
Decode untrusted carriers in a resource-capped sandbox with no network or host-file access. Run the complete policy-selected scanner population, distinguish NO_PAYLOAD, PAYLOAD_FOUND, NOT_APPLICABLE, and ERROR, and emit only payload type, location, length, digest, and an encrypted quarantine reference. Never print extracted bytes. If threat classification is required, send the exact extracted-byte digest through the authenticated D-001 contract rather than assuming every QR, OCR, or watermark payload is malicious.
# File: multimodal/run_carrier_scanners.py
from __future__ import annotations
import hashlib
import json
import os
import re
import stat
import subprocess
import tempfile
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
IMAGE = re.compile(r"^[a-z0-9][a-z0-9._/:~-]*@sha256:[a-f0-9]{64}$")
SHA256 = re.compile(r"^[a-f0-9]{64}$")
RESULT_FIELDS = {"schema_version", "scanner_id", "status", "payloads"}
PAYLOAD_FIELDS = {"payload_type", "location", "length", "sha256", "quarantine_ref"}
@dataclass(frozen=True)
class VerifiedScannerSpec:
scanner_id: str
image_digest: str
command: tuple[str, ...]
supported_mime_types: frozenset[str]
@dataclass(frozen=True)
class VerifiedScannerPolicy:
policy_sha256: str
scanner_specs: tuple[VerifiedScannerSpec, ...]
timeout_seconds: float
maximum_output_bytes: int
memory_limit: str
cpu_limit: str
pids_limit: int
tmpfs_limit: str
def regular_sha256(path: Path) -> str:
digest = hashlib.sha256()
descriptor = os.open(
path, os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
)
with os.fdopen(descriptor, "rb") as handle:
if not stat.S_ISREG(os.fstat(handle.fileno()).st_mode):
raise ValueError("artifact is not a regular file")
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
@contextmanager
def private_snapshot(source: Path):
source = source.resolve(strict=True)
with tempfile.TemporaryDirectory(prefix="aidefend-carrier-scan-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
snapshot = root / "artifact"
source_fd = os.open(
source,
os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0),
)
digest = hashlib.sha256()
try:
before = os.fstat(source_fd)
if not stat.S_ISREG(before.st_mode):
raise ValueError("artifact is not a regular file")
snapshot_fd = os.open(
snapshot,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0),
0o600,
)
with (
os.fdopen(source_fd, "rb", closefd=False) as reader,
os.fdopen(snapshot_fd, "wb") as writer,
):
for chunk in iter(lambda: reader.read(1024 * 1024), b""):
digest.update(chunk)
writer.write(chunk)
writer.flush()
os.fsync(writer.fileno())
after = os.fstat(source_fd)
finally:
os.close(source_fd)
identity = lambda value: (
value.st_dev, value.st_ino, value.st_size,
value.st_mtime_ns, value.st_ctime_ns,
)
if identity(before) != identity(after):
raise OSError("source artifact changed while creating private snapshot")
snapshot_sha256 = digest.hexdigest()
if regular_sha256(snapshot) != snapshot_sha256:
raise OSError("private scan snapshot readback mismatch")
os.chmod(snapshot, 0o444)
yield snapshot, snapshot_sha256
def _run_scanners_snapshot(
*,
artifact: Path,
artifact_mime_type: str,
expected_artifact_sha256: str,
policy: VerifiedScannerPolicy,
) -> dict[str, object]:
artifact = artifact.resolve(strict=True)
scanner_ids = [spec.scanner_id for spec in policy.scanner_specs]
if (
not SHA256.fullmatch(policy.policy_sha256)
or not SHA256.fullmatch(expected_artifact_sha256)
or not artifact_mime_type
or not policy.scanner_specs
or len(scanner_ids) != len(set(scanner_ids))
or policy.timeout_seconds <= 0
or policy.maximum_output_bytes < 1
or policy.pids_limit < 1
or any(
not spec.scanner_id or not IMAGE.fullmatch(spec.image_digest)
or not spec.command or not spec.supported_mime_types
for spec in policy.scanner_specs
)
):
raise ValueError("verified scanner policy is invalid")
if "," in str(artifact):
raise ValueError("artifact path cannot be encoded safely for OCI --mount")
artifact_sha256 = regular_sha256(artifact)
if artifact_sha256 != expected_artifact_sha256:
raise ValueError("artifact digest differs from signed scan request")
results = []
applicable_count = 0
for spec in policy.scanner_specs:
if artifact_mime_type not in spec.supported_mime_types:
results.append({
"scanner_id": spec.scanner_id,
"status": "NOT_APPLICABLE",
"payloads": [],
})
continue
applicable_count += 1
command = [
"podman", "run", "--rm", "--network=none", "--read-only",
"--cap-drop=ALL", "--security-opt=no-new-privileges",
f"--pids-limit={policy.pids_limit}",
f"--memory={policy.memory_limit}", f"--cpus={policy.cpu_limit}",
"--user=65532:65532",
"--mount", f"type=bind,src={artifact},dst=/input/artifact,ro=true",
"--tmpfs", f"/tmp:rw,noexec,nosuid,size={policy.tmpfs_limit}",
spec.image_digest, *spec.command, "/input/artifact",
]
try:
with tempfile.TemporaryFile() as stdout, tempfile.TemporaryFile() as stderr:
completed = subprocess.run(
command, stdin=subprocess.DEVNULL, stdout=stdout, stderr=stderr,
timeout=policy.timeout_seconds, check=False,
)
stdout_size = stdout.tell()
if completed.returncode != 0:
results.append({
"scanner_id": spec.scanner_id,
"status": "ERROR",
"payloads": [],
"error_type": "SCANNER_NONZERO_EXIT",
})
continue
if stdout_size > policy.maximum_output_bytes:
results.append({
"scanner_id": spec.scanner_id,
"status": "ERROR",
"payloads": [],
"error_type": "SCANNER_OUTPUT_LIMIT",
})
continue
stdout.seek(0)
value = json.loads(stdout.read().decode("utf-8", errors="strict"))
except subprocess.TimeoutExpired:
results.append({
"scanner_id": spec.scanner_id,
"status": "ERROR",
"payloads": [],
"error_type": "SCANNER_TIMEOUT",
})
continue
except (OSError, UnicodeError, json.JSONDecodeError):
results.append({
"scanner_id": spec.scanner_id,
"status": "ERROR",
"payloads": [],
"error_type": "SCANNER_RUNTIME_OR_SCHEMA",
})
continue
if (
not isinstance(value, dict) or set(value) != RESULT_FIELDS
or value["schema_version"] != "aidefend.carrier-scanner-result.v1"
or value["scanner_id"] != spec.scanner_id
or value["status"] not in {"NO_PAYLOAD", "PAYLOAD_FOUND"}
or not isinstance(value["payloads"], list)
or (value["status"] == "NO_PAYLOAD" and value["payloads"])
or (value["status"] == "PAYLOAD_FOUND" and not value["payloads"])
):
results.append({
"scanner_id": spec.scanner_id,
"status": "ERROR",
"payloads": [],
"error_type": "SCANNER_RESULT_SCHEMA",
})
continue
valid_payloads = True
for payload in value["payloads"]:
valid_payloads = valid_payloads and (
isinstance(payload, dict) and set(payload) == PAYLOAD_FIELDS
and isinstance(payload["length"], int) and not isinstance(payload["length"], bool)
and payload["length"] >= 0
and SHA256.fullmatch(str(payload["sha256"])) is not None
and all(
isinstance(payload[field], str) and payload[field]
for field in ("payload_type", "location", "quarantine_ref")
)
)
if not valid_payloads:
results.append({
"scanner_id": spec.scanner_id,
"status": "ERROR",
"payloads": [],
"error_type": "PAYLOAD_EVIDENCE_SCHEMA",
})
continue
results.append({
"scanner_id": spec.scanner_id,
"status": value["status"],
"payloads": value["payloads"],
})
statuses = {item["status"] for item in results}
outcome = (
"INSUFFICIENT_DATA" if applicable_count == 0
else "ERROR" if "ERROR" in statuses
else "FAIL" if "PAYLOAD_FOUND" in statuses
else "PASS"
)
return {
"schema_version": "aidefend.carrier-scan-aggregate.v1",
"control": "AID-D-007.001",
"artifact_sha256": artifact_sha256,
"artifact_mime_type": artifact_mime_type,
"policy_sha256": policy.policy_sha256,
"results": results,
"outcome": outcome,
}
def run_scanners(
*,
artifact: Path,
artifact_mime_type: str,
expected_artifact_sha256: str,
policy: VerifiedScannerPolicy,
) -> dict[str, object]:
if SHA256.fullmatch(expected_artifact_sha256) is None:
raise ValueError("signed artifact digest is invalid")
with private_snapshot(artifact) as (snapshot, snapshot_sha256):
if snapshot_sha256 != expected_artifact_sha256:
raise ValueError("private snapshot differs from signed scan request")
return _run_scanners_snapshot(
artifact=snapshot,
artifact_mime_type=artifact_mime_type,
expected_artifact_sha256=snapshot_sha256,
policy=policy,
)Independent verification
An independent runner replays benign, embedded-payload, malformed, decompression-bomb, unsupported-family, output-limit, timeout, nonzero-exit, and schema-failure fixtures with the same digest-pinned images. It reconciles the expected scanner set against the aggregate and verifies each emitted digest against encrypted quarantined bytes without exposing content. The organizational policy supplies all resource limits; scanner or runtime failure remains ERROR, never NO_PAYLOAD.
Document carrier adapter
Consume the exact source and projection digests from AID-H-002.003-G005. Inspect extracted text and bounded metadata evidence for instruction-bearing content that a rendered-page review might miss. Report PDF nonpainting text, optional-content layers, and DOCX hidden-run/style declarations with source locations where the adapter supports them; otherwise label location coverage as unavailable. Compare inherited style behavior, not only a direct run flag. White or small text alone is a candidate feature, not proof of a malicious instruction. Use the deployment's calibrated semantic detector for instruction content and retain benign OCR, accessibility, white-on-dark, and unused-style cases in the false-positive suite. Detection emits a finding bound to the projection; the Harden input gate owns admission/quarantine. If the model consumes raw files, this text-only adapter is insufficient unless every additional model-visible representation is inspected.
AID-D-007.001-G003
Verify complete coverage
Verify a signed aggregation policy binding tenant/workflow, request/content IDs, expected detector control IDs and producer workload identities, finding schema versions, detector/policy/artifact versions, required modality population, freshness/sequence rules, preprocessing-fact schema, duplicate handling, and aggregate event/evidence retention.
Production implementation
Consume findings over authenticated channels, verify each signature, producer, schema, scope, freshness, and evidence digest, require exact equality with the expected detector population, and preserve each source state. Preprocessing facts remain facts, never detector verdicts, and the aggregate has no allow/deny field.
# File: multimodal/aggregate_findings.py
from __future__ import annotations
import hashlib
import json
import re
from dataclasses import dataclass
from datetime import datetime, timezone
SHA256 = re.compile(r"^[a-f0-9]{64}$")
OUTCOMES = {"PASS", "FAIL", "INSUFFICIENT_DATA", "ERROR"}
FINDING_FIELDS = {
"schema_version", "request_id_sha256", "detector_control_id",
"producer_identity", "sequence", "observed_at", "outcome",
"finding_count", "evidence_sha256",
}
@dataclass(frozen=True)
class VerifiedAggregationPolicy:
policy_sha256: str
request_id_sha256: str
expected_producers: dict[str, str]
not_before: datetime
expires_at: datetime
def canonical_sha256(value: object) -> str:
return hashlib.sha256(
json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
).hexdigest()
def timestamp(value: str) -> datetime:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError("finding timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
def aggregate(
policy: VerifiedAggregationPolicy,
verified_findings: list[dict[str, object]],
) -> dict[str, object]:
expected = set(policy.expected_producers)
if (
not SHA256.fullmatch(policy.policy_sha256)
or not SHA256.fullmatch(policy.request_id_sha256)
or not expected
or any(not detector or not producer for detector, producer in policy.expected_producers.items())
or policy.not_before.tzinfo is None or policy.expires_at.tzinfo is None
or policy.not_before >= policy.expires_at
):
raise ValueError("verified aggregation policy is invalid")
by_detector = {}
for finding in verified_findings:
if not isinstance(finding, dict) or set(finding) != FINDING_FIELDS:
raise ValueError("finding schema differs")
detector = finding["detector_control_id"]
sequence = finding["sequence"]
finding_count = finding["finding_count"]
if (
detector not in expected or detector in by_detector
or finding["schema_version"] != "aidefend.detector-finding.v1"
or finding["request_id_sha256"] != policy.request_id_sha256
or finding["producer_identity"] != policy.expected_producers[detector]
or finding["outcome"] not in OUTCOMES
or isinstance(sequence, bool) or not isinstance(sequence, int) or sequence < 0
or isinstance(finding_count, bool) or not isinstance(finding_count, int)
or finding_count < 0
or not SHA256.fullmatch(str(finding["evidence_sha256"]))
):
raise ValueError("finding binding or value is invalid")
observed = timestamp(str(finding["observed_at"]))
if not policy.not_before <= observed < policy.expires_at:
raise ValueError("finding is outside the signed aggregation window")
if (finding["outcome"] == "FAIL") != (finding_count > 0):
raise ValueError("finding count and outcome disagree")
by_detector[detector] = finding
missing = sorted(expected - set(by_detector))
sources = [
{
"detector_control_id": detector,
"producer_identity": finding["producer_identity"],
"sequence": finding["sequence"],
"outcome": finding["outcome"],
"finding_count": finding["finding_count"],
"evidence_sha256": finding["evidence_sha256"],
}
for detector, finding in sorted(by_detector.items())
]
outcomes = {finding["outcome"] for finding in by_detector.values()}
if "FAIL" in outcomes:
outcome = "FAIL"
elif "ERROR" in outcomes:
outcome = "ERROR"
elif missing or "INSUFFICIENT_DATA" in outcomes:
outcome = "INSUFFICIENT_DATA"
else:
outcome = "PASS"
aggregate_event = {
"schema_version": "aidefend.multimodal-aggregate.v1",
"control": "AID-D-007.001",
"request_id_sha256": policy.request_id_sha256,
"policy_sha256": policy.policy_sha256,
"expected_detector_ids": sorted(expected),
"missing_detector_ids": missing,
"sources": sources,
"finding_count": sum(item["finding_count"] for item in sources),
"outcome": outcome,
}
return {
**aggregate_event,
"aggregate_sha256": canonical_sha256(aggregate_event),
}Independent verification
A separate verifier reconstructs the expected population from the signed policy and authoritative request ledger, revalidates source receipts, recomputes the canonical aggregate bytes and state, and checks the event bus plus immutable evidence store for exact set and digest equality. Negative fixtures cover duplicate, unknown, missing, stale, misbound, ERROR, and mixed finding populations; none may produce an allow/deny field.
AID-D-007.001-G004
Production implementation
At the instrumented serving observation point, capture only the declared normalized attention statistics for each sampled eligible inference, validate tensor shape/masks/finite values, calculate policy-defined concentration/entropy/deviation features, and emit a finding without changing output or routing. Unsupported layers, fused kernels, quantization paths or missing hooks are surfaced rather than assumed normal.
Independent verification
An independent pinned runner replays signed benign and manipulation fixtures plus policy boundary cases, recomputes tensors/features, checks sampled population selection against serving logs, and compares emitted event digests and thresholds. Model-release or hook drift invalidates the profile until recalibrated.
AID-D-007.001-G005
Detect complementary fragments after semantic extraction
Run pinned OCR, QR/barcode, metadata, container, audio, and extracted-text adapters inside resource-capped sandboxes. Each signature-verified adapter receipt must bind the artifact, source, modality, adapter digest, and normalized ontology concepts. Evaluate a signed composition-rule library over the complete receipt population. A chained-payload finding requires complementary semantic fragments from distinct observations and at least the policy-required number of modalities; a single carrier containing the whole pattern is handled by the ordinary payload rules in G002, not double-counted here.
from __future__ import annotations
# File: detection/complementary_fragment_detector.py
from dataclasses import dataclass
from itertools import product
from typing import Sequence
@dataclass(frozen=True)
class VerifiedSemanticObservation:
artifact_id: str
source_id: str
modality: str
adapter_sha256: str
observation_sha256: str
concept_ids: frozenset[str]
@dataclass(frozen=True)
class CompositionRule:
rule_id: str
required_fragments: tuple[frozenset[str], ...]
minimum_modalities: int
def detect_complementary_composition(
observations: Sequence[VerifiedSemanticObservation],
rules: Sequence[CompositionRule],
) -> list[dict]:
if not observations or not rules:
raise ValueError("verified observations and signed rules are required")
identities = {(item.artifact_id, item.observation_sha256) for item in observations}
if len(identities) != len(observations):
raise ValueError("semantic observation population contains duplicates")
for item in observations:
if (
not item.artifact_id or not item.source_id or not item.modality
or len(item.adapter_sha256) != 64 or len(item.observation_sha256) != 64
or not item.concept_ids or any(not value for value in item.concept_ids)
):
raise ValueError("verified semantic observation is malformed")
findings = []
for rule in rules:
fragments = tuple(frozenset(fragment) for fragment in rule.required_fragments)
if (
not rule.rule_id or len(fragments) < 2
or any(not fragment for fragment in fragments)
or isinstance(rule.minimum_modalities, bool)
or rule.minimum_modalities < 2
):
raise ValueError("signed composition rule is invalid")
carriers = [
[item for item in observations if fragment.issubset(item.concept_ids)]
for fragment in fragments
]
if any(not candidates for candidates in carriers):
continue
complete_pattern = frozenset().union(*fragments)
for candidate_tuple in product(*carriers):
if len({item.observation_sha256 for item in candidate_tuple}) != len(fragments):
continue
if len({item.modality for item in candidate_tuple}) < rule.minimum_modalities:
continue
if any(complete_pattern.issubset(item.concept_ids) for item in candidate_tuple):
continue
findings.append({
"finding_type": "complementary_cross_modal_payload",
"rule_id": rule.rule_id,
"artifact_ids": sorted({item.artifact_id for item in candidate_tuple}),
"source_ids": sorted({item.source_id for item in candidate_tuple}),
"modalities": sorted({item.modality for item in candidate_tuple}),
"observation_sha256": sorted(item.observation_sha256 for item in candidate_tuple),
"matched_fragments": [sorted(fragment) for fragment in fragments],
})
break
return findings
Keep fixtures that prove composition, not keyword coincidence
from complementary_fragment_detector import (
CompositionRule, VerifiedSemanticObservation,
detect_complementary_composition,
)
RULE = CompositionRule(
rule_id="CISCO-19.2.2-CREDENTIAL-EXFIL",
required_fragments=(
frozenset({"credential_secret_target"}),
frozenset({"external_exfiltration_action"}),
),
minimum_modalities=2,
)
def observation(identifier: str, modality: str, concepts: set[str]):
return VerifiedSemanticObservation(
artifact_id="artifact-42", source_id="upload-42", modality=modality,
adapter_sha256="a" * 64, observation_sha256=identifier * 64,
concept_ids=frozenset(concepts),
)
positive = [
observation("1", "image", {"credential_secret_target"}),
observation("2", "audio", {"external_exfiltration_action"}),
]
missing_fragment = [positive[0]]
single_carrier = [
observation("3", "image", {
"credential_secret_target", "external_exfiltration_action",
})
]
benign_pair = [
observation("4", "image", {"credential_secret_target"}),
observation("5", "audio", {"accessibility_narration"}),
]
assert len(detect_complementary_composition(positive, [RULE])) == 1
assert detect_complementary_composition(missing_fragment, [RULE]) == []
assert detect_complementary_composition(single_carrier, [RULE]) == []
assert detect_complementary_composition(benign_pair, [RULE]) == []
Evidence and independent verification
Bind the signed rule-library digest, ontology version, adapter digests, expected artifact/modality population, and every observation digest into the detector result. Missing, extra, stale, unsupported, or failed adapters produce INSUFFICIENT_DATA or ERROR, never NO_FINDING. Do not log raw concealed content or execute extracted payloads. An independent runner verifies signatures, reconciles the authoritative ingestion ledger, recomputes every composition from the same immutable receipts, and replays positive complementary-fragment, missing-fragment, same-carrier, same-modality, benign-pair, near-match, malformed, and adapter-failure fixtures. This Detect guidance emits a finding for runtime fusion effects; a separate Harden control owns transformation or rejection.
AID-D-007.001-G006
Production boundary
Run proposition extraction and expert inference in pinned isolated adapters, then evaluate their authenticated numeric outputs with one shared module. This guidance owns post-generation faithfulness and modality dominance; the same module also computes the expert-disagreement signal consumed by G007 so the framework does not publish two subtly different normalization or missing-population implementations. Missing required modalities/experts or excessive uncertainty is INSUFFICIENT_DATA, not agreement.
Shared faithfulness and disagreement scorer
# File: multimodal/score_faithfulness_and_disagreement.py
from __future__ import annotations
import argparse
import json
import math
from itertools import combinations
from pathlib import Path
def strict_json(raw: bytes) -> object:
def unique(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate_json_key")
value[key] = item
return value
return json.loads(raw.decode("utf-8", errors="strict"), object_pairs_hook=unique,
parse_constant=lambda value: (_ for _ in ()).throw(ValueError("nonfinite_json")))
def unit(value: object, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(field + "_type")
number = float(value)
if not math.isfinite(number) or not 0.0 <= number <= 1.0:
raise ValueError(field + "_range")
return number
def distribution(value: object, size: int) -> tuple[float, ...]:
if not isinstance(value, list) or len(value) != size:
raise ValueError("distribution_shape")
result = tuple(unit(item, "probability") for item in value)
if not math.isclose(sum(result), 1.0, rel_tol=0.0, abs_tol=0.000001):
raise ValueError("distribution_sum")
return result
def js_divergence(left: tuple[float, ...], right: tuple[float, ...]) -> float:
middle = tuple((a + b) / 2.0 for a, b in zip(left, right))
def kl(source, target):
return sum(a * math.log2(a / b) for a, b in zip(source, target) if a > 0.0)
return (kl(left, middle) + kl(right, middle)) / 2.0
def evaluate(policy: dict, evidence: dict) -> dict:
if set(policy) != {"schema_version", "policy_version", "ontology_sha256", "ontology_labels", "required_modalities", "required_experts", "thresholds"}:
raise ValueError("policy_schema")
if set(evidence) != {"schema_version", "request_id", "response_sha256", "modality_support", "expert_distributions"}:
raise ValueError("evidence_schema")
if policy["schema_version"] != "aidefend.multimodal-score-policy.v1" or evidence["schema_version"] != "aidefend.multimodal-score-input.v1":
raise ValueError("schema_version")
labels = policy["ontology_labels"]
if not isinstance(labels, list) or not labels or any(not isinstance(label, str) or not label for label in labels) or len(labels) != len(set(labels)):
raise ValueError("ontology_labels")
thresholds = policy["thresholds"]
required_thresholds = {"min_faithfulness", "max_modality_dominance", "max_uncertainty", "max_js_divergence"}
if not isinstance(thresholds, dict) or set(thresholds) != required_thresholds:
raise ValueError("threshold_schema")
limit = {name: unit(thresholds[name], name) for name in required_thresholds}
required_modalities = set(policy["required_modalities"])
required_experts = set(policy["required_experts"])
if not required_modalities or not required_experts or len(required_modalities) != len(policy["required_modalities"]) or len(required_experts) != len(policy["required_experts"]):
raise ValueError("required_population")
modalities = {}
if not isinstance(evidence["modality_support"], list):
raise ValueError("modality_population")
for row in evidence["modality_support"]:
if not isinstance(row, dict) or set(row) != {"modality_id", "source_sha256", "support", "uncertainty"}:
raise ValueError("modality_schema")
identity = row["modality_id"]
if identity in modalities or identity not in required_modalities:
raise ValueError("duplicate_or_extra_modality")
modalities[identity] = {"source_sha256": row["source_sha256"], "support": unit(row["support"], "support"),
"uncertainty": unit(row["uncertainty"], "uncertainty")}
experts = {}
if not isinstance(evidence["expert_distributions"], list):
raise ValueError("expert_population")
for row in evidence["expert_distributions"]:
if not isinstance(row, dict) or set(row) != {"expert_id", "artifact_sha256", "ontology_sha256", "probabilities", "uncertainty"}:
raise ValueError("expert_schema")
identity = row["expert_id"]
if identity in experts or identity not in required_experts or row["ontology_sha256"] != policy["ontology_sha256"]:
raise ValueError("expert_identity_or_ontology")
experts[identity] = {"artifact_sha256": row["artifact_sha256"],
"probabilities": distribution(row["probabilities"], len(labels)),
"uncertainty": unit(row["uncertainty"], "uncertainty")}
missing_modalities = sorted(required_modalities - set(modalities))
missing_experts = sorted(required_experts - set(experts))
uncertain_modalities = sorted(key for key, row in modalities.items() if row["uncertainty"] > limit["max_uncertainty"])
uncertain_experts = sorted(key for key, row in experts.items() if row["uncertainty"] > limit["max_uncertainty"])
usable_support = [row["support"] for key, row in modalities.items() if key not in uncertain_modalities]
faithfulness = sum(usable_support) / len(usable_support) if usable_support else None
support_total = sum(usable_support)
dominance = max(usable_support) / support_total if support_total > 0.0 else 0.0
usable_experts = [row["probabilities"] for key, row in experts.items() if key not in uncertain_experts]
divergences = [js_divergence(left, right) for left, right in combinations(usable_experts, 2)]
max_divergence = max(divergences) if divergences else None
insufficient = bool(missing_modalities or missing_experts or uncertain_modalities or uncertain_experts or faithfulness is None or max_divergence is None)
findings = []
if faithfulness is not None and faithfulness < limit["min_faithfulness"]:
findings.append("low_faithfulness")
if usable_support and dominance > limit["max_modality_dominance"]:
findings.append("single_modality_dominance")
if max_divergence is not None and max_divergence > limit["max_js_divergence"]:
findings.append("expert_disagreement")
outcome = "FAIL" if findings else "INSUFFICIENT_DATA" if insufficient else "PASS"
return {"schema_version": "aidefend.multimodal-score-finding.v1", "control_id": "AID-D-007.001",
"request_id": evidence["request_id"], "response_sha256": evidence["response_sha256"],
"policy_version": policy["policy_version"], "ontology_sha256": policy["ontology_sha256"],
"missing_modality_ids": missing_modalities, "missing_expert_ids": missing_experts,
"uncertain_modality_ids": uncertain_modalities, "uncertain_expert_ids": uncertain_experts,
"faithfulness": faithfulness, "modality_dominance": dominance,
"max_expert_js_divergence": max_divergence, "findings": findings, "outcome": outcome}
def main() -> None:
parser = argparse.ArgumentParser()
for name in ("policy", "evidence", "out"):
parser.add_argument("--" + name, type=Path, required=True)
args = parser.parse_args()
try:
result = evaluate(strict_json(args.policy.read_bytes()), strict_json(args.evidence.read_bytes()))
except Exception as error:
result = {"schema_version": "aidefend.multimodal-score-finding.v1", "control_id": "AID-D-007.001",
"outcome": "ERROR", "reason_code": type(error).__name__}
temporary = args.out.with_suffix(args.out.suffix + ".tmp")
temporary.write_text(json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False) + chr(10), encoding="utf-8")
temporary.replace(args.out)
if result["outcome"] == "ERROR":
raise SystemExit(2)
if __name__ == "__main__":
main()Independent verification
Verify signatures and privately snapshot the policy, immutable response/source evidence, and every adapter result before execution; sign the result even for non-PASS outcomes. Replay faithful, unfaithful, single-modality-dominated, agreement, disagreement, low-confidence, missing-expert, missing-modality, ontology-mismatch, malformed-vector, and adapter-failure fixtures. Recompute propositions and expert vectors from the exact source and response bytes, then reconcile the complete response population. Thresholds and required populations come only from the signed deployment policy.
AID-D-007.001-G007
Production implementation
Use the shared executable scorer in AID-D-007.001-G006; do not copy or fork its ontology validation, probability normalization, missing-expert handling, uncertainty limits, or Jensen-Shannon divergence calculation. Submit the exact same immutable modality-object digests to every required expert over authenticated channels and require the scorer's signed expert_disagreement finding. Low-confidence, missing, stale, extra, or differently-versioned expert output remains INSUFFICIENT_DATA or ERROR, never agreement.
Independent verification
An independently credentialed runner validates expert identity and artifact provenance, recomputes every vector from the exact inputs, and replays signed agreement, disagreement, low-confidence, missing/extra expert, invalid distribution, and ontology-mismatch fixtures through the G006 module. Reconcile the required expert population for every eligible request and verify the signed result digest before consuming the signal.
AID-D-007.001-G008
Production boundary
Compare an authenticated observed modality graph against a separately signed workflow contract. This is a structural context check only: it reports missing, extra, forbidden, or misbound nodes and edges, without classifying semantic safety or making an admission decision.
Exact workflow-graph comparator
# File: multimodal/compare_workflow_graph.py
from __future__ import annotations
import argparse
import json
from pathlib import Path
def strict_json(raw: bytes) -> object:
def unique(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate_json_key")
value[key] = item
return value
return json.loads(raw.decode("utf-8", errors="strict"), object_pairs_hook=unique,
parse_constant=lambda value: (_ for _ in ()).throw(ValueError("nonfinite_json")))
def keyed(rows: object, fields: set[str], key: str, label: str) -> dict:
if not isinstance(rows, list):
raise ValueError(label + "_population")
result = {}
for row in rows:
if not isinstance(row, dict) or set(row) != fields:
raise ValueError(label + "_schema")
identity = row[key]
if not isinstance(identity, str) or not identity or identity in result:
raise ValueError(label + "_identity")
result[identity] = row
return result
def evaluate(contract: dict, observed: dict) -> dict:
top = {"schema_version", "workflow_id", "policy_version", "nodes", "edges"}
if not isinstance(contract, dict) or set(contract) != top or not isinstance(observed, dict) or set(observed) != top:
raise ValueError("graph_schema")
if contract["schema_version"] != "aidefend.workflow-modality-graph.v1" or observed["schema_version"] != "aidefend.observed-modality-graph.v1":
raise ValueError("schema_version")
if contract["workflow_id"] != observed["workflow_id"] or contract["policy_version"] != observed["policy_version"]:
raise ValueError("graph_binding")
node_fields = {"node_id", "media_type", "source", "role", "provenance_sha256"}
edge_fields = {"edge_id", "from_node", "to_node", "relationship"}
expected_nodes = keyed(contract["nodes"], node_fields, "node_id", "node")
observed_nodes = keyed(observed["nodes"], node_fields, "node_id", "node")
expected_edges = keyed(contract["edges"], edge_fields, "edge_id", "edge")
observed_edges = keyed(observed["edges"], edge_fields, "edge_id", "edge")
missing_nodes = sorted(set(expected_nodes) - set(observed_nodes))
extra_nodes = sorted(set(observed_nodes) - set(expected_nodes))
mismatched_nodes = sorted(key for key in set(expected_nodes) & set(observed_nodes) if expected_nodes[key] != observed_nodes[key])
missing_edges = sorted(set(expected_edges) - set(observed_edges))
extra_edges = sorted(set(observed_edges) - set(expected_edges))
mismatched_edges = sorted(key for key in set(expected_edges) & set(observed_edges) if expected_edges[key] != observed_edges[key])
findings = {"missing_nodes": missing_nodes, "extra_nodes": extra_nodes,
"mismatched_nodes": mismatched_nodes, "missing_edges": missing_edges,
"extra_edges": extra_edges, "mismatched_edges": mismatched_edges}
outcome = "FAIL" if any(findings.values()) else "PASS"
return {"schema_version": "aidefend.workflow-context-finding.v1", "control_id": "AID-D-007.001",
"workflow_id": contract["workflow_id"], "policy_version": contract["policy_version"],
"expected_node_count": len(expected_nodes), "observed_node_count": len(observed_nodes),
"expected_edge_count": len(expected_edges), "observed_edge_count": len(observed_edges),
"findings": findings, "outcome": outcome}
def main() -> None:
parser = argparse.ArgumentParser()
for name in ("contract", "observed", "out"):
parser.add_argument("--" + name, type=Path, required=True)
args = parser.parse_args()
try:
result = evaluate(strict_json(args.contract.read_bytes()), strict_json(args.observed.read_bytes()))
except Exception as error:
result = {"schema_version": "aidefend.workflow-context-finding.v1", "control_id": "AID-D-007.001",
"outcome": "ERROR", "reason_code": type(error).__name__}
temporary = args.out.with_suffix(args.out.suffix + ".tmp")
temporary.write_text(json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False) + chr(10), encoding="utf-8")
temporary.replace(args.out)
if result["outcome"] == "ERROR":
raise SystemExit(2)
if __name__ == "__main__":
main()Independent verification
Verify and privately snapshot both signed inputs, run the comparator, and sign every outcome. An independent verifier reconstructs the observed graph from the authoritative ingestion ledger and replays valid, missing, extra, reordered, forged-provenance, relationship-mismatch, duplicate-key, and exception fixtures. Reconcile every eligible request and modality object; contract thresholds, exceptions, and policy versions must never come from the submitted artifact.
Open-source tools
- OpenCV
- Pillow
- ExifTool
- binwalk
- Tesseract OCR
- spaCy
- Hugging Face Transformers
- librosa
- pyzbar
- stegano
- Aletheia
- zsteg
Source-available / open-weight tools
- OpenAI Whisper (MIT; open-weight)
Commercial / hosted tools
- Azure AI Content Safety
- Amazon Rekognition
- Google Cloud Vision API
- Sensity AI Deepfake Detection
Threat mappings
MITRE ATLAS
- AML.T0051 LLM Prompt Injection (cross-modal injection variants)
- AML.T0051.000 LLM Prompt Injection: Direct
- AML.T0051.001 LLM Prompt Injection: Indirect
- AML.T0015 Evade AI Model (multimodal evasion)
- AML.T0043 Craft Adversarial Data (multimodal adversarial examples)
- AML.T0043.000 Craft Adversarial Data: White-Box Optimization
- AML.T0043.003 Craft Adversarial Data: Manual Modification
- AML.T0052.001 Phishing: Deepfake-Assisted Phishing (cross-modal inconsistency detection catches spoofed media evidence)
- AML.T0123 Obfuscated Files or Information (detects hidden payloads embedded in multimodal artifacts)
- AML.T0129 Triggers in Multimodal Inputs (carrier and modality findings expose instructions outside the primary visible text channel)
- AML.T0068 LLM Prompt Obfuscation (document carrier features and semantic inspection expose hidden-text instruction candidates)
MAESTRO
- Adversarial Examples (L1)
- Input Validation Attacks (L3)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection
- LLM07:2026 Misinformation (cross-modal faithfulness and expert-disagreement findings expose manipulated evidence or misleading output)
OWASP ML Top 10 2023
- ML01:2023 Input Manipulation Attack (multimodal inputs)
OWASP Top 10 for Agentic Applications 2026
- ASI01:2026 Agent Goal Hijack (cross-modal hidden instructions redirect agent goals)
NIST Adversarial Machine Learning 2025
- NISTAML.018 Prompt Injection
- NISTAML.022 Evasion
- NISTAML.015 Indirect Prompt Injection (hidden instructions in media are indirect injection)
Cisco Integrated AI Security and Safety Framework
- AISubtech-1.3.1 Image-Text Injection (instructions recovered from supported carriers and classified under the detector contract)
- AISubtech-1.3.3 Audio Command Injection (instructions recovered from supported carriers and classified under the detector contract)
- AISubtech-1.3.4 Video Overlay Manipulation (instructions recovered from supported carriers and classified under the detector contract)
- AISubtech-19.1.1 Contradictory Inputs Attack
- AISubtech-19.1.2 Modality Skewing
- AISubtech-19.2.2 Chained Payload Execution
- AITech-1.3 Multimodal Injection and Manipulation (instructions recovered from supported carriers and classified under the detector contract)
- AITech-19.1 Cross-Modal Inconsistency Exploits
- AITech-19.2 Fusion Payload Split
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection (cross-modal prompt injection detection)
- MEV: Model Evasion (multimodal evasion attacks)
Databricks AI Security Framework 3.0
- Model Serving - Inference requests 9.1: Prompt inject
- Agents - Tools MCP Server 13.16: Prompt Injection
- Agents - Core 13.6: Intent Breaking & Goal Manipulation (cross-modal hidden instructions break agent intent)