Scope boundary

Owns preventive decode, type validation, bounds, canonical re-encoding, malware or content-disarm evidence, quarantine, and final release for generated media and binary artifacts.

AID-H-006.002: AID-H-006.004 owns non-text media and binary artifact release; AID-H-006.002 owns text, markup, URL, structured-payload, and token-stream release. A document may require bounded text extraction, but its container and embedded objects still belong to the binary gate.

AID-D-003.008: AID-H-006.004 transforms, quarantines, blocks, or releases the generated artifact; AID-D-003.008 emits policy, malware, type, and coverage findings over it. The detector supplies evidence, while this control owns the release decision.

AID-H-025: AID-H-006.004 rejects executable or unsafe binary output from release; AID-H-025 governs whether code or commands may be executed in a controlled runtime. Artifact release and execution admission remain separate even when the same bytes are involved.

Implementation guidance

• AID-H-006.004-G001: Decode generated media under policy-defined byte, dimension, frame, duration, and decompression limits; verify content-derived type and re-encode from decoded pixels or samples.
• AID-H-006.004-G002: Identify binary type from bytes, deny executables and archives by default, and require malware plus content-disarm receipts for approved document types.

AID-H-006.004-G001

Load and verify the versioned artifact runtime profile

Limits and allowed formats are deployment inputs, not AIDEFEND defaults. Mount the verifier key from an approved, read-only trust source that the codec worker cannot modify. The profile may choose a file path, but it may not supply or replace its own verifier key. Missing files, a bad signature, duplicate JSON keys, an unexpected schema, or an unversioned profile stops worker startup.

# File: output_guards/verified_artifact_profile.py
from __future__ import annotations

import base64
import hashlib
import hmac
import json
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


TRUSTED_VERIFIER_KEY = Path(
    "/etc/aidefend/trust/generated-artifact-policy-ed25519.pem"
)
PROFILE_PATH_ENV = "AIDEFEND_GENERATED_ARTIFACT_PROFILE"
SIGNATURE_PATH_ENV = "AIDEFEND_GENERATED_ARTIFACT_PROFILE_SIGNATURE"
PROFILE_DIGEST_ENV = "AIDEFEND_GENERATED_ARTIFACT_PROFILE_SHA256"
PROFILE_SCHEMA = "aidefend.generated-artifact-runtime-profile.v1"


class ProfileVerificationError(RuntimeError):
    pass


@dataclass(frozen=True)
class VerifiedArtifactProfile:
    version: str
    sha256: str
    image: dict[str, Any]
    document: dict[str, Any]


def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ProfileVerificationError(f"duplicate_profile_key:{key}")
        result[key] = value
    return result


def _reject_non_finite(value: str) -> None:
    raise ProfileVerificationError(f"non_finite_profile_number:{value}")


def load_verified_artifact_profile() -> VerifiedArtifactProfile:
    try:
        profile_path = Path(os.environ[PROFILE_PATH_ENV]).resolve(strict=True)
        signature_path = Path(os.environ[SIGNATURE_PATH_ENV]).resolve(strict=True)
        raw_profile = profile_path.read_bytes()
        expected_digest = os.environ[PROFILE_DIGEST_ENV].lower()
        actual_digest = hashlib.sha256(raw_profile).hexdigest()
        if (
            re.fullmatch(r"[0-9a-f]{64}", expected_digest) is None
            or not hmac.compare_digest(actual_digest, expected_digest)
        ):
            raise ProfileVerificationError("artifact_profile_digest_mismatch")
        signature = base64.b64decode(
            signature_path.read_text(encoding="ascii").strip(), validate=True
        )
        public_key = serialization.load_pem_public_key(
            TRUSTED_VERIFIER_KEY.read_bytes()
        )
        if not isinstance(public_key, Ed25519PublicKey):
            raise ProfileVerificationError("unexpected_profile_verifier_key_type")
        public_key.verify(signature, raw_profile)
        document = json.loads(
            raw_profile.decode("utf-8"),
            object_pairs_hook=_unique_object,
            parse_constant=_reject_non_finite,
        )
    except (
        KeyError,
        OSError,
        UnicodeError,
        ValueError,
        InvalidSignature,
    ) as exc:
        raise ProfileVerificationError("artifact_profile_verification_failed") from exc

    required = {"schema_version", "profile_version", "image", "document"}
    if not isinstance(document, dict) or set(document) != required:
        raise ProfileVerificationError("artifact_profile_schema_mismatch")
    if document["schema_version"] != PROFILE_SCHEMA:
        raise ProfileVerificationError("unsupported_artifact_profile_schema")
    version = document["profile_version"]
    if not isinstance(version, str) or not version.strip():
        raise ProfileVerificationError("artifact_profile_version_missing")
    if not isinstance(document["image"], dict) or not isinstance(document["document"], dict):
        raise ProfileVerificationError("artifact_profile_sections_invalid")
    return VerifiedArtifactProfile(
        version=version,
        sha256=actual_digest,
        image=dict(document["image"]),
        document=dict(document["document"]),
    )


PROFILE = load_verified_artifact_profile()

Canonical image adapter

The image section of that verified profile supplies the input allowlist and every byte, pixel, width, and height bound. The adapter supports a single canonical PNG output contract; any other requested adapter contract fails startup. Do not trust a filename, declared MIME type, or Base64 wrapper.

# File: output_guards/canonical_image.py
from __future__ import annotations

import hashlib
import io
import warnings
from dataclasses import dataclass

from PIL import Image, UnidentifiedImageError

from output_guards.verified_artifact_profile import PROFILE, ProfileVerificationError


@dataclass(frozen=True)
class ImagePolicy:
    allowed_input_formats: frozenset[str]
    max_input_bytes: int
    max_output_bytes: int
    max_pixels: int
    max_width: int
    max_height: int
    output_format: str
    output_media_type: str


def _positive_int(value: object, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise ProfileVerificationError(f"invalid_image_policy_field:{field}")
    return value


def _load_image_policy() -> ImagePolicy:
    raw = PROFILE.image
    required = {
        "allowed_input_formats", "max_input_bytes", "max_output_bytes",
        "max_pixels", "max_width", "max_height", "output_format",
        "output_media_type",
    }
    if set(raw) != required:
        raise ProfileVerificationError("image_policy_schema_mismatch")
    formats = raw["allowed_input_formats"]
    if (
        not isinstance(formats, list)
        or not formats
        or any(not isinstance(item, str) or not item for item in formats)
    ):
        raise ProfileVerificationError("image_input_format_allowlist_invalid")
    if raw["output_format"] != "PNG" or raw["output_media_type"] != "image/png":
        raise ProfileVerificationError("unsupported_canonical_image_contract")
    return ImagePolicy(
        allowed_input_formats=frozenset(formats),
        max_input_bytes=_positive_int(raw["max_input_bytes"], "max_input_bytes"),
        max_output_bytes=_positive_int(raw["max_output_bytes"], "max_output_bytes"),
        max_pixels=_positive_int(raw["max_pixels"], "max_pixels"),
        max_width=_positive_int(raw["max_width"], "max_width"),
        max_height=_positive_int(raw["max_height"], "max_height"),
        output_format=raw["output_format"],
        output_media_type=raw["output_media_type"],
    )


POLICY = _load_image_policy()
Image.MAX_IMAGE_PIXELS = POLICY.max_pixels
warnings.simplefilter("error", Image.DecompressionBombWarning)


def _check_dimensions(width: int, height: int) -> None:
    if (
        width <= 0
        or height <= 0
        or width > POLICY.max_width
        or height > POLICY.max_height
        or width * height > POLICY.max_pixels
    ):
        raise PermissionError("image_dimensions_denied")


def canonicalize_image(raw: bytes) -> tuple[bytes, dict]:
    if not isinstance(raw, bytes) or not raw or len(raw) > POLICY.max_input_bytes:
        raise PermissionError("image_input_budget_denied")
    try:
        with Image.open(io.BytesIO(raw)) as probe:
            if probe.format not in POLICY.allowed_input_formats:
                raise PermissionError("image_format_denied")
            _check_dimensions(probe.width, probe.height)
            probe.verify()
        with Image.open(io.BytesIO(raw)) as decoded:
            decoded.load()
            if decoded.format not in POLICY.allowed_input_formats:
                raise PermissionError("decoded_image_format_denied")
            if getattr(decoded, "n_frames", 1) != 1:
                raise PermissionError("animated_image_denied")
            _check_dimensions(decoded.width, decoded.height)
            with decoded.convert("RGB") as pixels:
                with Image.frombytes("RGB", pixels.size, pixels.tobytes()) as clean:
                    target = io.BytesIO()
                    clean.save(target, format=POLICY.output_format, optimize=True)
    except (
        UnidentifiedImageError,
        OSError,
        Image.DecompressionBombError,
        Image.DecompressionBombWarning,
    ) as exc:
        raise PermissionError("image_decode_denied") from exc

    body = target.getvalue()
    if not body or len(body) > POLICY.max_output_bytes:
        raise PermissionError("canonical_image_output_budget_denied")
    with Image.open(io.BytesIO(body)) as readback:
        readback.load()
        if readback.format != POLICY.output_format:
            raise PermissionError("canonical_image_readback_type_mismatch")
        _check_dimensions(readback.width, readback.height)
        width, height = readback.size
    return body, {
        "status": "PASS",
        "policy_version": PROFILE.version,
        "policy_sha256": PROFILE.sha256,
        "input_sha256": hashlib.sha256(raw).hexdigest(),
        "output_sha256": hashlib.sha256(body).hexdigest(),
        "output_media_type": POLICY.output_media_type,
        "width": width,
        "height": height,
    }

Run codec workers without network, credentials, host mounts, or shared writable paths, and enforce CPU, memory, wall-clock, frame, and duration budgets from the same verified runtime profile outside the codec. Independently reopen the emitted artifact and reconcile its digest, content-derived type, dimensions, profile version, and profile digest with the stored release receipt. Signature verification, profile loading, codec failure, or receipt readback failure releases nothing.

AID-H-006.004-G002

Quarantine and document transformation

Use the signature-verified profile loaded by the preceding guidance. Its document section supplies the allowed content-derived media types, encoded/decoded/output byte bounds, scanner timeout, CDR timeout, and approved scanner and adapter identities plus executable digests. The caller supplies only the encoded artifact and expected release type; it cannot choose the policy version, limits, timeouts, scanner, or CDR adapter.

# File: output_guards/document_gate.py
from __future__ import annotations

import base64
import hashlib
import hmac
import io
import json
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import magic

from output_guards.verified_artifact_profile import PROFILE, ProfileVerificationError


@dataclass(frozen=True)
class ExecutableBinding:
    identity: str
    path: Path
    sha256: str


@dataclass(frozen=True)
class DocumentPolicy:
    allowed_media_types: frozenset[str]
    max_base64_encoded_bytes: int
    max_decoded_bytes: int
    max_output_bytes: int
    scanner_timeout_seconds: float
    cdr_timeout_seconds: float
    scanner: ExecutableBinding
    cdr_adapter: ExecutableBinding


def _positive_number(value: object, field: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)) or value <= 0:
        raise ProfileVerificationError(f"invalid_document_policy_field:{field}")
    return float(value)


def _positive_int(value: object, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise ProfileVerificationError(f"invalid_document_policy_field:{field}")
    return value


def _sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(io.DEFAULT_BUFFER_SIZE), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _load_executable(raw: object, field: str) -> ExecutableBinding:
    if not isinstance(raw, dict) or set(raw) != {"identity", "path", "sha256"}:
        raise ProfileVerificationError(f"invalid_executable_binding:{field}")
    identity, path_text, expected_digest = raw["identity"], raw["path"], raw["sha256"]
    if not all(isinstance(value, str) and value for value in (identity, path_text, expected_digest)):
        raise ProfileVerificationError(f"incomplete_executable_binding:{field}")
    path = Path(path_text).resolve(strict=True)
    actual_digest = _sha256_file(path)
    if not hmac.compare_digest(actual_digest, expected_digest.lower()):
        raise ProfileVerificationError(f"executable_digest_mismatch:{field}")
    return ExecutableBinding(identity=identity, path=path, sha256=actual_digest)


def _load_document_policy() -> DocumentPolicy:
    raw = PROFILE.document
    required = {
        "allowed_media_types", "max_base64_encoded_bytes", "max_decoded_bytes",
        "max_output_bytes", "scanner_timeout_seconds", "cdr_timeout_seconds",
        "scanner", "cdr_adapter",
    }
    if set(raw) != required:
        raise ProfileVerificationError("document_policy_schema_mismatch")
    allowed = raw["allowed_media_types"]
    if (
        not isinstance(allowed, list)
        or not allowed
        or any(not isinstance(item, str) or not item for item in allowed)
    ):
        raise ProfileVerificationError("document_media_type_allowlist_invalid")
    return DocumentPolicy(
        allowed_media_types=frozenset(allowed),
        max_base64_encoded_bytes=_positive_int(
            raw["max_base64_encoded_bytes"], "max_base64_encoded_bytes"
        ),
        max_decoded_bytes=_positive_int(raw["max_decoded_bytes"], "max_decoded_bytes"),
        max_output_bytes=_positive_int(raw["max_output_bytes"], "max_output_bytes"),
        scanner_timeout_seconds=_positive_number(
            raw["scanner_timeout_seconds"], "scanner_timeout_seconds"
        ),
        cdr_timeout_seconds=_positive_number(raw["cdr_timeout_seconds"], "cdr_timeout_seconds"),
        scanner=_load_executable(raw["scanner"], "scanner"),
        cdr_adapter=_load_executable(raw["cdr_adapter"], "cdr_adapter"),
    )


POLICY = _load_document_policy()


def _unique_receipt_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise PermissionError("cdr_receipt_duplicate_key")
        result[key] = value
    return result


def _reject_receipt_constant(value: str) -> None:
    raise PermissionError(f"cdr_receipt_non_finite_number:{value}")


def _load_json(path: Path) -> dict[str, Any]:
    value = json.loads(
        path.read_text(encoding="utf-8"),
        object_pairs_hook=_unique_receipt_object,
        parse_constant=_reject_receipt_constant,
    )
    if not isinstance(value, dict):
        raise PermissionError("cdr_receipt_not_an_object")
    return value


def _assert_executable_unchanged(binding: ExecutableBinding) -> None:
    try:
        current_digest = _sha256_file(binding.path)
    except OSError as exc:
        raise PermissionError("bound_executable_unavailable") from exc
    if not hmac.compare_digest(current_digest, binding.sha256):
        raise PermissionError("bound_executable_changed_after_profile_load")


def admit_document(encoded: str, expected_type: str) -> tuple[bytes, dict]:
    if not isinstance(encoded, str) or expected_type not in POLICY.allowed_media_types:
        raise PermissionError("document_policy_denied")
    try:
        encoded_bytes = encoded.encode("ascii", errors="strict")
    except UnicodeEncodeError as exc:
        raise PermissionError("document_base64_not_ascii") from exc
    if not encoded_bytes or len(encoded_bytes) > POLICY.max_base64_encoded_bytes:
        raise PermissionError("encoded_document_budget_denied")
    try:
        raw = base64.b64decode(encoded_bytes, validate=True)
    except (ValueError, base64.binascii.Error) as exc:
        raise PermissionError("invalid_document_base64") from exc
    if not raw or len(raw) > POLICY.max_decoded_bytes:
        raise PermissionError("decoded_document_budget_denied")
    detected_type = magic.from_buffer(raw, mime=True)
    if detected_type != expected_type:
        raise PermissionError("document_type_mismatch")

    input_digest = hashlib.sha256(raw).hexdigest()
    with tempfile.TemporaryDirectory(prefix="artifact-quarantine-") as directory:
        quarantine = Path(directory)
        candidate = quarantine / "candidate.bin"
        transformed_path = quarantine / "transformed.bin"
        cdr_receipt_path = quarantine / "cdr-receipt.json"
        candidate.write_bytes(raw)
        _assert_executable_unchanged(POLICY.scanner)
        try:
            scanner_version = subprocess.run(
                [str(POLICY.scanner.path), "--version"],
                capture_output=True,
                text=True,
                timeout=POLICY.scanner_timeout_seconds,
                check=True,
                close_fds=True,
            ).stdout.strip()
            scan = subprocess.run(
                [
                    str(POLICY.scanner.path), "--no-summary", "--infected",
                    "--stdout", str(candidate),
                ],
                capture_output=True,
                text=True,
                timeout=POLICY.scanner_timeout_seconds,
                check=False,
                close_fds=True,
            )
        except (OSError, subprocess.SubprocessError) as exc:
            raise PermissionError("malware_scanner_unavailable_or_timed_out") from exc
        if not scanner_version or scan.returncode != 0:
            raise PermissionError("malware_scan_not_clean_or_signatures_unavailable")

        _assert_executable_unchanged(POLICY.cdr_adapter)
        try:
            subprocess.run(
                [
                    str(POLICY.cdr_adapter.path),
                    "--input", str(candidate),
                    "--output", str(transformed_path),
                    "--receipt", str(cdr_receipt_path),
                    "--expected-media-type", expected_type,
                    "--policy-version", PROFILE.version,
                    "--policy-sha256", PROFILE.sha256,
                ],
                capture_output=True,
                text=True,
                timeout=POLICY.cdr_timeout_seconds,
                check=True,
                close_fds=True,
            )
            output_size = transformed_path.stat().st_size
            if output_size <= 0 or output_size > POLICY.max_output_bytes:
                raise PermissionError("transformed_document_budget_denied")
            transformed = transformed_path.read_bytes()
            cdr_receipt = _load_json(cdr_receipt_path)
        except (OSError, subprocess.SubprocessError, UnicodeError, ValueError) as exc:
            raise PermissionError("cdr_adapter_unavailable_or_failed") from exc

    if not transformed or len(transformed) > POLICY.max_output_bytes:
        raise PermissionError("transformed_document_budget_denied")
    output_type = magic.from_buffer(transformed, mime=True)
    output_digest = hashlib.sha256(transformed).hexdigest()
    required_receipt = {
        "status": "PASS",
        "policy_version": PROFILE.version,
        "policy_sha256": PROFILE.sha256,
        "adapter_identity": POLICY.cdr_adapter.identity,
        "adapter_sha256": POLICY.cdr_adapter.sha256,
        "input_sha256": input_digest,
        "input_media_type": detected_type,
        "output_sha256": output_digest,
        "output_media_type": output_type,
    }
    if output_type != expected_type or any(
        cdr_receipt.get(key) != value for key, value in required_receipt.items()
    ):
        raise PermissionError("cdr_receipt_binding_failed")
    return transformed, {
        **required_receipt,
        "scanner_identity": POLICY.scanner.identity,
        "scanner_sha256": POLICY.scanner.sha256,
        "scanner_version": scanner_version,
    }

Executables, scripts, disk images, archives, encrypted containers, unknown types, and polyglots remain denied unless a separately owned executable-artifact control admits them. Scanner timeout, missing signatures, non-clean or indeterminate return codes, CDR timeout, adapter digest drift, type or digest mismatch, malformed receipt, and evidence-write or readback failure release nothing. Test oversized encoded input before decode, oversized decoded input, malformed Base64, decompression bombs, appended executables, nested archives, encrypted documents, scanner outages, changed executables, and altered CDR receipts.

Open-source tools

• Pillow
• pyca/cryptography (runtime-profile signature verification)
• FFmpeg
• python-magic
• ClamAV
• Dangerzone
• Apache Tika (content detection and extraction; isolate parsers independently)

Commercial / hosted tools

• OPSWAT MetaDefender
• Glasswall CDR
• Menlo File Security
• VirusTotal

Threat mappings

MITRE ATLAS

• AML.T0050 Command and Scripting Interpreter
• AML.T0123 Obfuscated Files or Information (content-derived type identification and polyglot rejection defeat disguised binary artifacts)

MAESTRO

• Integration Risks (L7) (sanitized outputs protect downstream integrations)

OWASP LLM Top 10 2026

• LLM10:2026 Improper Output Handling

OWASP ML Top 10 2023

• ML09:2023 Output Integrity Attack

OWASP Top 10 for Agentic Applications 2026

• ASI05:2026 Unexpected Code Execution (RCE) (sanitization prevents code execution via output)

NIST Adversarial Machine Learning 2025

• NISTAML.039 Compromising connected resources (sanitized outputs prevent downstream exploitation)

Cisco Integrated AI Security and Safety Framework

• AISubtech-12.2.1 Code Detection / Malicious Code Output
• AISubtech-13.1.2 Memory Flooding
• AISubtech-13.1.4 Application Denial of Service
• AITech-12.2 Insecure Output Handling
• AITech-13.1 Disruption of Availability

Google Secure AI Framework 2.0 - Risks

• IMO: Insecure Model Output

Databricks AI Security Framework 3.0

• Agents - Core 13.11: Unexpected RCE and Code Attacks (output sanitization strips malicious code)