AID-H-002.003: Multimodal Input Sanitization
Pillars: data
Lifecycle: building, operation
Scope boundary
Owns deterministic byte and type limits, safe decoding, metadata handling, canonical re-encoding, bounded purification, and the final admit, block, or quarantine decision for complete non-realtime multimodal artifacts.
AID-H-002.002: AID-H-002.003 owns binary and media artifact decoding and sanitization; AID-H-002.002 owns text and structured-input validation at live inference ingress. OCR or extracted text may need the text gate after the artifact has first passed safe decoding.
AID-H-002.007: AID-H-002.003 sanitizes and may rebuild external media bytes; AID-H-002.007 extracts and verifies source provenance before metadata stripping or rebuilding. Provenance must be captured first because sanitization can intentionally remove the original manifest.
AID-H-002.008: AID-H-002.003 handles complete uploaded files and artifacts; AID-H-002.008 handles realtime streams whose timing and incremental state affect admission. Batch sanitization and live stream gating have different populations and safe-failure behavior.
AID-D-007.001: AID-H-002.003 executes preventive artifact transformation and admission; AID-D-007.001 emits cross-modal inconsistency and hidden-payload findings for complete artifacts. The detector supplies evidence, while this control owns the final artifact gate.
AID-H-002.009: AID-H-002.003 safely decodes and sanitizes complete image, audio, video, document, and OCR artifacts; AID-H-002.009 governs dataset containers, reference descriptors, and the parser result. A dataset whose records include media may require both controls: parser admission for the container and media admission for each extracted artifact.
Implementation guidance
- AID-H-002.003-G001: Rebuild images from decoded pixels after byte-level admission checks, strip metadata, and re-encode to a safe baseline format.
- AID-H-002.003-G002: Purify images via a denoising diffusion step for high-security use cases.
- AID-H-002.003-G003: Re-encode audio to standardized lossy or PCM format to remove hidden commands.
- AID-H-002.003-G004: File-type and content safety gates for all uploads.
- AID-H-002.003-G005: Build a bounded PDF/OOXML text projection and gate the exact bytes used for model context or embeddings, including hidden-text candidates.
AID-H-002.003-G001
Concept:
Image sanitization should behave like a deterministic intake gate, not a best-effort image edit. Verify provenance first when you also deploy AID-H-002.007, then apply one bounded image sanitizer that enforces type, size, and frame limits before rebuilding the image from decoded pixels and stripping metadata.
Step 1: Enforce byte-level admission checks before decode
# File: multimodal_guards/image_sanitizer.py
from __future__ import annotations
import os
import io
from dataclasses import dataclass
from pathlib import Path
import magic
from PIL import Image, ImageFile, ImageOps, UnidentifiedImageError
def _required_positive_int(name: str) -> int:
value = int(os.environ[name])
if value <= 0:
raise RuntimeError(f"{name} must be a positive integer")
return value
MAX_BYTES = _required_positive_int("VERIFIED_H002_IMAGE_UPLOAD_MAX_BYTES")
IMAGE_DECODE_MAX_PIXELS = _required_positive_int("VERIFIED_H002_IMAGE_DECODE_MAX_PIXELS")
ImageFile.LOAD_TRUNCATED_IMAGES = False
Image.MAX_IMAGE_PIXELS = IMAGE_DECODE_MAX_PIXELS
ALLOWED_MIME_EXTENSIONS = {
"image/jpeg": {".jpg", ".jpeg"},
"image/png": {".png"},
"image/webp": {".webp"},
}
OUTPUT_FORMAT_BY_MIME = {
"image/jpeg": "JPEG",
"image/png": "PNG",
"image/webp": "PNG",
}
@dataclass(frozen=True)
class SanitizedImage:
mime: str
width: int
height: int
sanitized_bytes: bytes
def sanitize_image_upload(filename: str, blob: bytes) -> SanitizedImage:
if not blob:
raise ValueError("empty_upload")
if len(blob) > MAX_BYTES:
raise ValueError("file_too_large")
sniffed_mime = magic.from_buffer(blob, mime=True)
allowed_extensions = ALLOWED_MIME_EXTENSIONS.get(sniffed_mime)
if allowed_extensions is None:
raise ValueError("mime_not_allowed")
if Path(filename).suffix.lower() not in allowed_extensions:
raise ValueError("extension_mime_mismatch")
try:
with Image.open(io.BytesIO(blob)) as probe:
probe.verify()
with Image.open(io.BytesIO(blob)) as source:
if getattr(source, "n_frames", 1) != 1:
raise ValueError("animated_images_not_allowed")
normalized = ImageOps.exif_transpose(source).convert("RGB")
except (UnidentifiedImageError, OSError) as exc:
raise ValueError("invalid_image") from exc
if normalized.width * normalized.height > Image.MAX_IMAGE_PIXELS:
raise ValueError("image_too_large")
out = io.BytesIO()
normalized.save(out, format=OUTPUT_FORMAT_BY_MIME[sniffed_mime], optimize=True)
return SanitizedImage(
mime=sniffed_mime,
width=normalized.width,
height=normalized.height,
sanitized_bytes=out.getvalue(),
)
Step 2: Integrate after provenance verification and before model-facing decode
# File: multimodal_guards/intake_pipeline.py
from multimodal_guards.image_sanitizer import sanitize_image_upload
def prepare_image_for_model(filename: str, blob: bytes, *, provenance_status: str | None) -> bytes:
if provenance_status in {"invalid", "stripped"}:
raise PermissionError("image_rejected_by_provenance_gate")
sanitized = sanitize_image_upload(filename, blob)
return sanitized.sanitized_bytes
Operational notes
- Do not sanitize first and verify provenance later, or you will destroy the metadata needed for
AID-H-002.007. - Reject animated images, decompression bombs, truncated files, and extension/MIME mismatches rather than trying to "repair" them.
- Keep the sanitizer output format narrow so downstream OCR and vision pipelines always receive one known-good baseline.
Action: Put one deterministic image sanitizer in front of every vision, OCR, or multimodal path. If the file fails raw-byte admission checks or safe decode, block it before any model-facing processing begins.
Before you begin
Apply AID-H-002.003-G001 only when an image, audio, video, document, archive, or other multimodal upload can reach a decoder, OCR/ASR component, embedding pipeline, model, or tool.
AID-H-002.003-G002
Concept:
Diffusion purification is a high-cost secondary sanitization stage for images that still look risky after file gates, metadata stripping, and re-encoding. Treat it as an escalated path for sensitive uploads, not the default for every image.
Step 1: Load a real image-to-image diffusion pipeline
Use a maintained Diffusers image-to-image pipeline pinned to a reviewed model revision. The module below reconstructs the uploaded image with low strength so the semantic content stays recognizable while fine-grained perturbations are dampened.
# File: hardening/diffusion_purifier.py
from __future__ import annotations
import math
import io
import os
from pathlib import Path
import torch
from diffusers import AutoPipelineForImage2Image
from PIL import Image
def _required_positive_int(name: str) -> int:
value = int(os.environ[name])
if value <= 0:
raise RuntimeError(f"{name} must be a positive integer")
return value
PURIFIER_IMAGE_WIDTH = _required_positive_int("VERIFIED_H002_PURIFIER_IMAGE_WIDTH")
PURIFIER_IMAGE_HEIGHT = _required_positive_int("VERIFIED_H002_PURIFIER_IMAGE_HEIGHT")
PURIFIER_INFERENCE_STEPS = _required_positive_int("VERIFIED_H002_PURIFIER_INFERENCE_STEPS")
def _required_positive_float(name: str) -> float:
value = float(os.environ[name])
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"{name} must be a finite positive number")
return value
PURIFIER_STRENGTH = _required_positive_float("VERIFIED_H002_PURIFIER_STRENGTH")
PURIFIER_GUIDANCE_SCALE = _required_positive_float("VERIFIED_H002_PURIFIER_GUIDANCE_SCALE")
MODEL_ID = "stable-diffusion-v1-5/stable-diffusion-v1-5"
MODEL_REVISION = os.environ["SD_MODEL_REVISION"]
if len(MODEL_REVISION) != 40 or any(ch not in "0123456789abcdef" for ch in MODEL_REVISION.lower()):
raise RuntimeError("SD_MODEL_REVISION must be a 40-character pinned commit SHA")
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
DTYPE = torch.float16 if DEVICE == "cuda" else torch.float32
pipe = AutoPipelineForImage2Image.from_pretrained(
MODEL_ID,
revision=MODEL_REVISION,
torch_dtype=DTYPE,
use_safetensors=True,
).to(DEVICE)
if PURIFIER_STRENGTH > 1:
raise RuntimeError("verified purifier strength exceeds the pipeline domain")
pipe.set_progress_bar_config(disable=True)
def purify_image_bytes(image_bytes: bytes) -> bytes:
source = Image.open(io.BytesIO(image_bytes)).convert("RGB").resize((PURIFIER_IMAGE_WIDTH, PURIFIER_IMAGE_HEIGHT))
result = pipe(
prompt="same image, preserve scene and objects, remove adversarial noise",
image=source,
strength=PURIFIER_STRENGTH,
guidance_scale=PURIFIER_GUIDANCE_SCALE,
num_inference_steps=PURIFIER_INFERENCE_STEPS,
).images[0]
out = io.BytesIO()
result.save(out, format="PNG")
return out.getvalue()
if __name__ == "__main__":
purified = purify_image_bytes(Path("samples/upload.png").read_bytes())
Path("samples/upload.purified.png").write_bytes(purified)
Step 2: Route only high-risk uploads through the purifier
Apply purification after basic sanitization and only for explicitly escalated tiers such as public uploads or regulated-identity workflows.
# File: hardening/multimodal_gateway.py
from __future__ import annotations
from hardening.diffusion_purifier import purify_image_bytes
from multimodal_guards.image_sanitizer import sanitize_image_upload
HIGH_SECURITY_RISK_TIERS = {"regulated_identity", "public_upload_untrusted"}
def prepare_image_for_model(filename: str, image_bytes: bytes, risk_tier: str) -> bytes:
sanitized = sanitize_image_upload(filename, image_bytes).sanitized_bytes
if risk_tier in HIGH_SECURITY_RISK_TIERS:
return purify_image_bytes(sanitized)
return sanitized
Step 3: Verify that utility remains acceptable
Keep a regression set of benign and suspicious images. Measure classifier-label or OCR stability before and after purification, and fail promotion if benign-image accuracy collapses.
Action: Deploy diffusion purification as an opt-in high-security sanitization module with pinned model artifacts, explicit routing criteria, and a regression test that checks both threat reduction and acceptable utility loss before promotion.
Before you begin
Apply AID-H-002.003-G002 only when an image, audio, video, document, archive, or other multimodal upload can reach a decoder, OCR/ASR component, embedding pipeline, model, or tool.
Keep the action safely bounded
release only the newly generated sanitized artifact and deny the original on ambiguity, decoder failure, resource exhaustion, or policy mismatch.
AID-H-002.003-G003
Production implementation: bounded audio decode and canonical re-encode
Run FFprobe and FFmpeg in a non-root, network-disabled worker with a read-only input mount, seccomp/AppArmor profile, CPU and memory limits, and an ephemeral-volume quota no larger than the signed output limit. Do not decode attacker-controlled bytes in the API process.
# File: multimodal_guards/audio_sanitizer.py
from __future__ import annotations
import os
import hashlib
import json
import math
import subprocess
import wave
from dataclasses import dataclass
from pathlib import Path
def _required_positive_seconds(name: str) -> float:
value = float(os.environ[name])
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"{name} must be a finite positive number of seconds")
return value
AUDIO_PROBE_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H002_AUDIO_PROBE_TIMEOUT_SECONDS")
@dataclass(frozen=True)
class AudioPolicy:
policy_version: str
allowed_demuxers: frozenset[str]
maximum_input_bytes: int
maximum_output_bytes: int
maximum_duration_seconds: float
sample_rate_hz: int
channels: int
transcode_minimum_timeout_seconds: int
transcode_duration_multiplier: float
def run(command: list[str], timeout: int) -> subprocess.CompletedProcess[str]:
return subprocess.run(
command, check=True, capture_output=True, text=True, timeout=timeout
)
def sanitize_audio(source: Path, destination: Path, policy: AudioPolicy) -> dict:
if source.is_symlink() or not source.is_file():
raise ValueError("audio input must be a regular non-symlink file")
if destination.exists():
raise FileExistsError("destination must not exist")
input_size = source.stat().st_size
if input_size < 1 or input_size > policy.maximum_input_bytes:
raise PermissionError("audio input size is outside signed policy")
probe = run([
"ffprobe", "-v", "error",
"-show_entries", "format=format_name,duration",
"-of", "json", str(source),
], timeout=AUDIO_PROBE_TIMEOUT_SECONDS)
metadata = json.loads(probe.stdout)
fmt = metadata.get("format", {})
demuxers = set(str(fmt.get("format_name", "")).split(","))
duration = float(fmt.get("duration", "nan"))
if (
not demuxers.intersection(policy.allowed_demuxers)
or not math.isfinite(duration)
or duration <= 0
or duration > policy.maximum_duration_seconds
):
raise PermissionError("audio format or duration is outside signed policy")
run([
"ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-xerror",
"-n", "-i", str(source), "-map", "0:a:0", "-vn", "-sn", "-dn",
"-map_metadata", "-1", "-map_chapters", "-1",
"-ac", str(policy.channels), "-ar", str(policy.sample_rate_hz),
"-sample_fmt", "s16", "-f", "wav", str(destination),
], timeout=max(policy.transcode_minimum_timeout_seconds, math.ceil(duration * policy.transcode_duration_multiplier)))
if not destination.is_file() or destination.stat().st_size > policy.maximum_output_bytes:
destination.unlink(missing_ok=True)
raise PermissionError("canonical audio output exceeded signed policy")
with wave.open(str(destination), "rb") as decoded:
if (
decoded.getnchannels() != policy.channels
or decoded.getframerate() != policy.sample_rate_hz
or decoded.getsampwidth() != 2
or decoded.getnframes() < 1
):
destination.unlink(missing_ok=True)
raise RuntimeError("independent canonical WAV readback failed")
return {
"schema_version": "aidefend.audio-sanitization.v1",
"policy_version": policy.policy_version,
"input_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
"output_sha256": hashlib.sha256(destination.read_bytes()).hexdigest(),
"input_bytes": input_size,
"output_bytes": destination.stat().st_size,
"duration_seconds": duration,
"sample_rate_hz": policy.sample_rate_hz,
"channels": policy.channels,
}
Action: Sign the returned receipt, persist it with the exact FFmpeg/FFprobe image digest, and expose only the new WAV object. Decoder timeout, malformed metadata, multiple/absent audio streams, quota exhaustion, readback failure, or digest drift denies the upload.
Before you begin
Apply AID-H-002.003-G003 only when an image, audio, video, document, archive, or other multimodal upload can reach a decoder, OCR/ASR component, embedding pipeline, model, or tool.
Keep the action safely bounded
release only the newly generated sanitized artifact and deny the original on ambiguity, decoder failure, resource exhaustion, or policy mismatch.
The transcode disables copying source metadata and chapters. It does not prove removal of spoken instructions in the waveform. Keep downstream processing limited to the canonical decoded PCM samples; if later code exposes output-container tags to a model, independently enumerate and allowlist those tags (the encoder may create new metadata).
AID-H-002.003-G004
Concept:
Put a deterministic file gate on the raw upload bytes before any decoder, OCR pipeline, or model runtime touches the content. A production gate verifies the sniffed media type, checks that the extension matches policy, rejects container formats the workflow does not need, and enforces image/audio limits before sanitization runs.
Step 1: Inspect the raw upload bytes with type-specific checks
# File: multimodal_guards/file_gates.py
from __future__ import annotations
import os
import io
from dataclasses import dataclass
from pathlib import Path
import magic
from PIL import Image, UnidentifiedImageError
from pydub import AudioSegment
def _required_positive_int(name: str) -> int:
value = int(os.environ[name])
if value <= 0:
raise RuntimeError(f"{name} must be a positive integer")
return value
MAX_BYTES = _required_positive_int("VERIFIED_H002_UPLOAD_GATE_MAX_BYTES")
MAX_IMAGE_PIXELS = _required_positive_int("VERIFIED_H002_UPLOAD_GATE_MAX_IMAGE_PIXELS")
MAX_AUDIO_MS = _required_positive_int("VERIFIED_H002_UPLOAD_GATE_MAX_AUDIO_MILLISECONDS")
ALLOWED_MIME = {
"image/jpeg": {".jpg", ".jpeg"},
"image/png": {".png"},
"audio/wav": {".wav"},
"audio/mpeg": {".mp3"},
}
DENY_MIME = {"application/x-dosexec", "application/zip", "application/x-tar"}
@dataclass(frozen=True)
class UploadFacts:
mime: str
size_bytes: int
width: int | None = None
height: int | None = None
duration_ms: int | None = None
def _sniff_mime(blob: bytes) -> str:
return magic.from_buffer(blob, mime=True)
def _extension_matches(filename: str, mime: str) -> bool:
return Path(filename).suffix.lower() in ALLOWED_MIME.get(mime, set())
def inspect_upload(filename: str, blob: bytes) -> UploadFacts:
if not blob:
raise ValueError("empty_upload")
if len(blob) > MAX_BYTES:
raise ValueError("file_too_large")
mime = _sniff_mime(blob)
if mime in DENY_MIME or mime not in ALLOWED_MIME:
raise ValueError("mime_not_allowed")
if not _extension_matches(filename, mime):
raise ValueError("extension_mime_mismatch")
if mime.startswith("image/"):
try:
image = Image.open(io.BytesIO(blob))
image.verify()
image = Image.open(io.BytesIO(blob))
except (UnidentifiedImageError, OSError) as exc:
raise ValueError("invalid_image") from exc
pixels = image.width * image.height
if pixels > MAX_IMAGE_PIXELS:
raise ValueError("image_dimensions_too_large")
return UploadFacts(
mime=mime,
size_bytes=len(blob),
width=image.width,
height=image.height,
)
audio = AudioSegment.from_file(io.BytesIO(blob), format="mp3" if mime == "audio/mpeg" else "wav")
if len(audio) > MAX_AUDIO_MS:
raise ValueError("audio_duration_too_long")
if audio.channels > 2:
raise ValueError("audio_channel_count_not_allowed")
return UploadFacts(mime=mime, size_bytes=len(blob), duration_ms=len(audio))Step 2: Put the gate in front of the sanitizers and model path
# File: multimodal_guards/upload_entrypoint.py
from pathlib import Path
from tempfile import TemporaryDirectory
from multimodal_guards.audio_sanitizer import AudioPolicy, sanitize_audio
from multimodal_guards.file_gates import inspect_upload
from multimodal_guards.image_sanitizer import sanitize_image_upload
def prepare_upload(
filename: str, blob: bytes, *, audio_policy: AudioPolicy
) -> tuple[bytes, dict]:
facts = inspect_upload(filename, blob)
if facts.mime.startswith("image/"):
sanitized = sanitize_image_upload(filename, blob).sanitized_bytes
receipt = None
else:
# The sanitizer accepts bounded file paths and a signed AudioPolicy; write
# the already-inspected bytes into an ephemeral directory and expose only
# the canonical WAV bytes returned by that sanitizer.
with TemporaryDirectory(prefix="upload-audio-") as directory:
root = Path(directory)
source_path = root / "input.bin"
destination_path = root / "sanitized.wav"
source_path.write_bytes(blob)
receipt = sanitize_audio(source_path, destination_path, audio_policy)
sanitized = destination_path.read_bytes()
return sanitized, {
"mime": facts.mime,
"size_bytes": facts.size_bytes,
"width": facts.width,
"height": facts.height,
"duration_ms": facts.duration_ms,
"sanitization_receipt": receipt,
}Operational notes
- Inspect bytes, not just on-disk paths, so the decision applies to the exact artifact the model will receive.
- Deny archives, office documents, and any other container formats your multimodal flow does not explicitly need.
- Fail closed on parser errors and emit a structured reject code such as
mime_not_allowedorextension_mime_mismatch.
Action: Run the file gate before any decode, OCR, transcription, or model call. If the gate cannot positively identify an allowed media type and enforce its size or dimension limits, reject the upload.
Before you begin
Apply AID-H-002.003-G004 only when an image, audio, video, document, archive, or other multimodal upload can reach a decoder, OCR/ASR component, embedding pipeline, model, or tool.
PDF/DOCX support is an explicit alternative path in AID-H-002.003-G005. Keep this media gate unchanged for image/audio routes; route required documents through the bounded document worker and its projection gate, rather than allowing raw containers to bypass sanitization.
AID-H-002.003-G005
Step 1: Enable a distinct document path
Use this method when the application ingests PDF or DOCX documents, including RAG over shared enterprise files. Keep the image/audio-only allowlist in AID-H-002.003-G004 for those routes. Add a separate document route with content sniffing, exact extension/type policy, compressed/uncompressed size limits, and a network-disabled parser sandbox; do not simply add Office MIME types to the image/audio decoder. Verify original provenance through AID-H-002.007 before altering metadata. Pin the parser image/dependencies and enforce CPU, memory, wall-clock and output-volume quotas outside the parser process.
Step 2: Define exactly what the model can receive
The example below implements a deliberately bounded text-only PDF/DOCX path using PyMuPDF text extraction and defusedxml. It preserves hidden text in the analysis/model projection so that a visual-only review cannot silently skip model-visible instructions. PDF Info/XMP and OOXML docProps are excluded by construction. The raw container, omitted metadata, images, and attachments must not subsequently reach the model or an embedding service through a second decoder. This path does not claim complete visual-document understanding; route scanned pages, forms, revisions, or required omitted content to a separately tested adapter or review. Do not silently drop required business content.
# document_projection.py -- run only inside the bounded, network-disabled worker
import hashlib, io, json, zipfile
from importlib.metadata import version
from defusedxml import ElementTree as ET
import pymupdf
W = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}"
OFF = {"0", "false", "off"}
def project_document(raw: bytes, media_type: str, policy: dict) -> dict:
if not raw or len(raw) > policy["max_input_bytes"]:
raise ValueError("input_size")
features, parts = [], []
if media_type == "application/pdf":
if not raw.startswith(b"%PDF-"):
raise ValueError("pdf_signature")
with pymupdf.open(stream=raw, filetype="pdf") as doc:
if doc.needs_pass or not 0 < len(doc) <= policy["max_pages"]:
raise ValueError("encrypted_or_page_limit")
if doc.get_ocgs():
features.append("optional_content_layers")
if doc.embfile_count():
features.append("embedded_files_excluded")
for page in doc:
text = page.get_text("text", sort=True)
if not text.strip():
raise ValueError("ocr_or_nontext_adapter_required")
parts.append(text)
for span in page.get_texttrace():
if span.get("type") == 3 or span.get("opacity", 1) == 0:
features.append("nonpainting_pdf_text")
if span.get("size", 0) < policy["small_font_points"]:
features.append("small_pdf_text")
if next(page.widgets() or iter(()), None) is not None:
raise ValueError("form_adapter_required")
# Info/XMP, attachments, annotation bodies, and object names are not
# concatenated into model text. The raw PDF is not released downstream.
parser = "PyMuPDF/" + version("PyMuPDF")
elif media_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
with zipfile.ZipFile(io.BytesIO(raw)) as archive:
infos = archive.infolist()
names = [entry.filename for entry in infos]
if len(names) != len(set(names)) or len(infos) > policy["max_zip_members"]:
raise ValueError("duplicate_or_excess_members")
if any(e.flag_bits & 1 or e.file_size > policy["max_part_bytes"] for e in infos):
raise ValueError("encrypted_or_oversize_part")
if sum(e.file_size for e in infos) > policy["max_unpacked_bytes"]:
raise ValueError("unpacked_size")
if "word/document.xml" not in names or "[Content_Types].xml" not in names:
raise ValueError("not_docx")
if any(n.endswith("vbaProject.bin") or n.startswith("word/embeddings/") for n in names):
raise ValueError("active_or_embedded_content")
main = ET.fromstring(archive.read("word/document.xml"))
# Narrow text projection: paragraph text, including tables and hidden
# runs. Never use python-docx Font.hidden == False as a coverage test:
# None means inheritance, not visible text.
unsupported = {W+"altChunk", W+"subDoc", W+"ins", W+"del", W+"fldSimple", W+"instrText"}
if any(node.tag in unsupported for node in main.iter()):
raise ValueError("alternate_content_or_revision_adapter_required")
for paragraph in main.iter(W+"p"):
# Text boxes nested inside a paragraph require another adapter;
# do not silently double-count or misorder their text.
if any(n is not paragraph and n.tag == W+"p" for n in paragraph.iter()):
raise ValueError("nested_paragraph_adapter_required")
text = []
for n in paragraph.iter():
if n.tag == W+"t": text.append(n.text or "")
elif n.tag == W+"tab": text.append("\t")
elif n.tag in {W+"br", W+"cr"}: text.append("\n")
parts.append("".join(text))
inspect = [main]
if "word/styles.xml" in names:
inspect.append(ET.fromstring(archive.read("word/styles.xml")))
for root in inspect:
for n in root.iter():
if n.tag in {W+"vanish", W+"webHidden", W+"specVanish"} and n.get(W+"val", "true") not in OFF:
features.append("hidden_run_or_style_declaration")
if n.tag == W+"color" and n.get(W+"val", "").upper() == "FFFFFF":
features.append("white_font_declaration")
if n.tag == W+"sz":
try:
if int(n.get(W+"val", ""))/2 < policy["small_font_points"]:
features.append("small_word_font")
except ValueError:
raise ValueError("invalid_font_size") from None
# Deliberately excluded: docProps, comments, headers, footers, notes,
# drawings/OCR, field instructions, relationships, and external targets.
# If a workflow needs these, use a tested adapter before enabling it.
features.append("body_text_projection_only")
parser = "OOXML-body-v1/defusedxml-" + version("defusedxml")
else:
raise ValueError("document_type_not_allowed")
text = "\n".join(parts)
encoded = text.encode("utf-8")
if not text.strip() or len(encoded) > policy["max_text_bytes"]:
raise ValueError("empty_or_oversize_projection")
return {"source_sha256": hashlib.sha256(raw).hexdigest(),
"projection_sha256": hashlib.sha256(encoded).hexdigest(),
"parser": parser, "policy_version": policy["version"],
"features": sorted(set(features)), "text": text}Step 3: Bind admission, detection, and indexing to one projection
Validate the signed policy's positive integer limits and permitted parser versions at worker startup. Run the projection through AID-H-002.002 and send carrier features to AID-D-007.001-G002. Keep metadata in a bounded forensic record if needed, never append it later to the model prompt. The input gate's receipt must cover the exact UTF-8 projection digest, policy and parser revision. Only that admitted text may be chunked, embedded, or inserted into context; retain the source/projection/chunk lineage with AID-H-020. Re-extraction by a different parser, enabling OCR, or switching to raw-file multimodal input invalidates the receipt and requires a new gate over every model-visible representation.
A font color of white alone does not establish hidden text: white-on-dark documents are legitimate. Style declarations can be unused, and OCR layers can be legitimate nonpainting text. The feature list is candidate evidence, not a maliciousness verdict. Use calibrated policy and semantic inspection of the actual extracted instructions; quarantine unresolved high-risk findings without claiming that all hidden text is malicious or that extraction proves safety. If humans approve a document, show model-visible hidden-text candidates and excluded-part notices alongside the rendered view.
Step 4: Verify the boundary with real fixtures
Include PDF nonpainting text, white-on-white and white-on-dark text, a legitimate OCR layer, metadata-only instructions, DOCX direct and inherited w:vanish, tiny text, document properties, tables, tracked revisions, malformed XML, oversized ZIP members, and duplicated ZIP entries. Assert that any released text is the exact scanned projection, excluded fields never reappear downstream, and unsupported inputs do not become a pass. Keep visible-plaintext injection cases too: hidden-text checks do not address ordinary malicious prose.
Open-source tools
- Pillow
- OpenCV
- pydub
- librosa
- Hugging Face Diffusers
- FFmpeg
- python-magic
- ExifTool
- Apache Tika (content detection and extraction; isolate parsers independently)
Commercial / hosted tools
- OPSWAT MetaDefender
- Glasswall CDR
- Menlo File Security
Threat mappings
MITRE ATLAS
- AML.T0043 Craft Adversarial Data
- AML.T0015 Evade AI Model
- AML.T0119 Exploit Automated Artifact Processing Pipeline (safe decoding and canonical re-encoding of uploaded media before automated processing)
- AML.T0123 Obfuscated Files or Information (canonical re-encoding strips payloads hidden inside media containers)
- AML.T0129 Triggers in Multimodal Inputs (metadata exclusion and a gated text projection close uninspected document and media channels)
- AML.T0068 LLM Prompt Obfuscation (hidden document text remains in the exact projection subjected to input admission)
- AML.T0051 LLM Prompt Injection (document admission binds the inspected projection to model-facing text)
- AML.T0051.001 LLM Prompt Injection: Indirect (external documents cannot bypass admission through omitted metadata or a second raw-file decoder)
MAESTRO
- Adversarial Examples (L1)
- Input Validation Attacks (L3)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection
- LLM06:2026 Unbounded Consumption
OWASP ML Top 10 2023
- ML01:2023 Input Manipulation Attack
OWASP Top 10 for Agentic Applications 2026
- ASI01:2026 Agent Goal Hijack (document projection prevents uninspected hidden content from reaching agent context)
NIST Adversarial Machine Learning 2025
- NISTAML.022 Evasion
- NISTAML.015 Indirect Prompt Injection (gates external document text and excludes uninspected metadata)
- NISTAML.018 Prompt Injection (binds inspected document text to the admitted model projection)
Cisco Integrated AI Security and Safety Framework
- AISubtech-1.3.2 Image Manipulation (Purifies bounded visual perturbations under measured utility constraints)
- AISubtech-13.1.2 Memory Flooding
- AISubtech-13.1.4 Application Denial of Service
- AITech-1.3 Multimodal Injection and Manipulation (Purifies bounded visual perturbations under measured utility constraints)
- AITech-13.1 Disruption of Availability
Google Secure AI Framework 2.0 - Risks
- MEV: Model Evasion
- PIJ: Prompt Injection (external document projection and metadata boundaries prevent uninspected instruction channels)
Databricks AI Security Framework 3.0
- Model Serving - Inference requests 9.9: Input Resource Control
- Model Serving - Inference requests 9.1: Prompt inject (document projection is gated before model context)