AID-H-019.001: URL Normalization & Allowlist Filtering

Version 1.20261015 · Actionable sub-technique

Create a safe HTTP wrapper that normalizes URLs, enforces scheme/domain allowlists, resolves DNS and blocks private/internal IP ranges, then verifies that the exact canonical URL and every redirect target are expected public resources before any background fetch, preview, or image load occurs.

Pillars: app

Lifecycle: building, operation

Scope boundary

Owns canonical URL normalization, scheme and port policy, DNS resolution, public-address validation, redirect revalidation, pinned connection, response-type and size bounds, and the decision to perform a remote fetch.

AID-H-002.002: AID-H-019.001 decides whether and how a remote destination may be fetched; AID-H-002.002 validates the URL field and other request inputs supplied to inference. Input schema validation does not prevent DNS rebinding, private-address resolution, or unsafe redirects.

AID-H-019.002: AID-H-019.001 controls the network destination and bounded response bytes; AID-H-019.002 removes active HTML behavior and demotes extracted content to untrusted text. A safely fetched page can still contain active or malicious content.

Implementation guidance

AID-H-019.001-G001

Fetch each URL and redirect exactly once over an HTTPS connection pinned to the freshly validated public IP, with strict port, type, encoding, and body limits.
Implement a single-pass pinned safe fetcher
# File: agent/safe_fetcher.py
from __future__ import annotations

import hashlib
import ipaddress
import json
import math
import os
import socket
import stat
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urljoin, urlsplit, urlunsplit

import urllib3


POLICY_PATH = Path(os.environ["SAFE_FETCH_POLICY_PATH"])
POLICY_SIGNATURE_PATH = Path(os.environ["SAFE_FETCH_POLICY_SIGNATURE_PATH"])
POLICY_VERIFY_KEY = os.environ["SAFE_FETCH_POLICY_VERIFY_KEY"]
POLICY_FIELDS = {
    "policy_version", "allowed_domains", "allowed_ports",
    "allowed_media_types", "max_redirects", "max_body_bytes",
    "connect_timeout_seconds", "read_timeout_seconds",
}
MAX_POLICY_BYTES = int(os.environ["VERIFIED_SAFE_FETCH_MAX_POLICY_BYTES"])
SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(os.environ["VERIFIED_SAFE_FETCH_SIGNATURE_TIMEOUT_SECONDS"])
if (
    MAX_POLICY_BYTES < 1
    or not math.isfinite(SIGNATURE_VERIFY_TIMEOUT_SECONDS)
    or SIGNATURE_VERIFY_TIMEOUT_SECONDS <= 0
):
    raise RuntimeError("verified safe-fetch bootstrap bounds are invalid")


def read_stable_regular_file(path: Path, maximum_bytes: int) -> bytes:
    flags = os.O_RDONLY | os.O_CLOEXEC
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(path, flags)
    try:
        before = os.fstat(descriptor)
        if (
            not stat.S_ISREG(before.st_mode)
            or before.st_size < 2
            or before.st_size > maximum_bytes
        ):
            raise RuntimeError("safe-fetch policy must be a bounded regular file")
        chunks: list[bytes] = []
        remaining = maximum_bytes + 1
        while remaining:
            chunk = os.read(descriptor, min(65_536, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        payload = b"".join(chunks)
        after = os.fstat(descriptor)
        if (
            len(payload) != before.st_size
            or len(payload) > maximum_bytes
            or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
            != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
        ):
            raise RuntimeError("safe-fetch policy changed while being read")
        return payload
    finally:
        os.close(descriptor)


def unique_json_object(pairs: list[tuple[str, object]]) -> dict:
    result: dict = {}
    for key, value in pairs:
        if key in result:
            raise RuntimeError("safe-fetch policy contains a duplicate JSON key")
        result[key] = value
    return result


# Read exact bytes once, verify those same bytes, and only then parse them.
POLICY_BYTES = read_stable_regular_file(POLICY_PATH, MAX_POLICY_BYTES)
with tempfile.NamedTemporaryFile("wb", delete=False) as verified_copy:
    verified_copy.write(POLICY_BYTES)
    verified_copy.flush()
    os.fsync(verified_copy.fileno())
    verified_policy_path = verified_copy.name
try:
    subprocess.run(
        [
            "cosign", "verify-blob", "--key", POLICY_VERIFY_KEY,
            "--bundle", str(POLICY_SIGNATURE_PATH), verified_policy_path,
        ],
        check=True, capture_output=True, text=True, timeout=SIGNATURE_VERIFY_TIMEOUT_SECONDS,
    )
finally:
    os.unlink(verified_policy_path)
try:
    POLICY = json.loads(
        POLICY_BYTES.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_json_object,
        parse_constant=lambda value: (_ for _ in ()).throw(
            RuntimeError(f"invalid JSON constant: {value}")
        ),
    )
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
    raise RuntimeError("safe-fetch policy is not strict UTF-8 JSON") from exc
if not isinstance(POLICY, dict) or set(POLICY) != POLICY_FIELDS:
    raise RuntimeError("safe-fetch policy schema differs")

POLICY_VERSION = POLICY["policy_version"]
if not isinstance(POLICY_VERSION, str) or not POLICY_VERSION.strip():
    raise RuntimeError("safe-fetch policy version is required")
if (
    not isinstance(POLICY["allowed_domains"], list)
    or not POLICY["allowed_domains"]
    or any(
        not isinstance(value, str)
        or not value
        or value != value.lower()
        or value.endswith(".")
        or "*" in value
        or value.encode("idna").decode("ascii") != value
        for value in POLICY["allowed_domains"]
    )
):
    raise RuntimeError("safe-fetch allowed domains must be canonical IDNA names")
if (
    not isinstance(POLICY["allowed_ports"], list)
    or not POLICY["allowed_ports"]
    or any(type(value) is not int or not 1 <= value <= 65_535 for value in POLICY["allowed_ports"])
):
    raise RuntimeError("safe-fetch allowed ports are invalid")
if (
    not isinstance(POLICY["allowed_media_types"], list)
    or not POLICY["allowed_media_types"]
    or any(
        not isinstance(value, str)
        or value != value.lower()
        or value.count("/") != 1
        or any(character.isspace() for character in value)
        for value in POLICY["allowed_media_types"]
    )
):
    raise RuntimeError("safe-fetch media types are invalid")
if type(POLICY["max_redirects"]) is not int or not 0 <= POLICY["max_redirects"] <= 10:
    raise RuntimeError("safe-fetch redirect bound is invalid")
if type(POLICY["max_body_bytes"]) is not int or not 1 <= POLICY["max_body_bytes"] <= 67_108_864:
    raise RuntimeError("safe-fetch body bound is invalid")
for field, maximum in (("connect_timeout_seconds", 30.0), ("read_timeout_seconds", 120.0)):
    value = POLICY[field]
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 < float(value) <= maximum:
        raise RuntimeError(f"safe-fetch {field} is invalid")

ALLOWED_DOMAINS = set(POLICY["allowed_domains"])
ALLOWED_PORTS = set(POLICY["allowed_ports"])
ALLOWED_MEDIA_TYPES = set(POLICY["allowed_media_types"])
MAX_REDIRECTS = POLICY["max_redirects"]
MAX_BODY_BYTES = POLICY["max_body_bytes"]
CONNECT_TIMEOUT_SECONDS = float(POLICY["connect_timeout_seconds"])
READ_TIMEOUT_SECONDS = float(POLICY["read_timeout_seconds"])
POLICY_SHA256 = hashlib.sha256(POLICY_BYTES).hexdigest()


class SafeFetchDenied(RuntimeError):
    pass


@dataclass(frozen=True)
class SafeFetchResult:
    final_url: str
    connected_ip: str
    media_type: str
    body: bytes
    body_sha256: str
    redirect_count: int
    policy_version: str


def host_allowed(hostname: str) -> bool:
    return any(
        hostname == domain or hostname.endswith("." + domain)
        for domain in ALLOWED_DOMAINS
    )


def canonicalize(raw_url: str) -> str:
    parts = urlsplit(raw_url)
    if parts.scheme.lower() != "https":
        raise SafeFetchDenied("https_required")
    if parts.username is not None or parts.password is not None:
        raise SafeFetchDenied("userinfo_forbidden")
    if not parts.hostname:
        raise SafeFetchDenied("hostname_required")

    try:
        hostname = parts.hostname.rstrip(".").encode("idna").decode("ascii").lower()
        port = parts.port or 443
    except (UnicodeError, ValueError) as exc:
        raise SafeFetchDenied("invalid_hostname_or_port") from exc

    if not host_allowed(hostname):
        raise SafeFetchDenied("hostname_not_allowlisted")
    if port not in ALLOWED_PORTS:
        raise SafeFetchDenied("port_not_allowlisted")
    if any(ord(char) < 0x20 for char in (parts.path + parts.query)):
        raise SafeFetchDenied("control_character_in_url")

    netloc = hostname if port == 443 else f"{hostname}:{port}"
    return urlunsplit(("https", netloc, parts.path or "/", parts.query, ""))


def resolve_and_validate(hostname: str, port: int) -> list[str]:
    addresses: set[str] = set()
    try:
        answers = socket.getaddrinfo(
            hostname,
            port,
            type=socket.SOCK_STREAM,
            proto=socket.IPPROTO_TCP,
        )
    except socket.gaierror as exc:
        raise SafeFetchDenied("dns_resolution_failed") from exc

    for answer in answers:
        candidate = answer[4][0]
        address = ipaddress.ip_address(candidate)
        if not address.is_global:
            raise SafeFetchDenied(f"non_public_dns_answer:{candidate}")
        addresses.add(str(address))
    if not addresses:
        raise SafeFetchDenied("dns_answer_empty")
    return sorted(addresses)


def request_pinned(canonical_url: str) -> tuple[urllib3.HTTPResponse, str]:
    parts = urlsplit(canonical_url)
    assert parts.hostname is not None
    port = parts.port or 443
    validated_ips = resolve_and_validate(parts.hostname, port)
    connected_ip = validated_ips[0]
    request_target = urlunsplit(("", "", parts.path or "/", parts.query, ""))
    host_header = parts.hostname if port == 443 else f"{parts.hostname}:{port}"

    pool = urllib3.HTTPSConnectionPool(
        connected_ip,
        port=port,
        server_hostname=parts.hostname,
        assert_hostname=parts.hostname,
        cert_reqs="CERT_REQUIRED",
        timeout=urllib3.Timeout(
            connect=CONNECT_TIMEOUT_SECONDS,
            read=READ_TIMEOUT_SECONDS,
        ),
        retries=False,
        maxsize=1,
        block=True,
    )
    response = pool.request(
        "GET",
        request_target,
        headers={
            "Host": host_header,
            "User-Agent": "AIDEFEND-SafeFetcher/1.0",
            "Accept": ", ".join(sorted(ALLOWED_MEDIA_TYPES)),
            "Accept-Encoding": "identity",
        },
        redirect=False,
        preload_content=False,
        decode_content=False,
    )
    return response, connected_ip


def read_bounded_body(response: urllib3.HTTPResponse) -> tuple[bytes, str]:
    encoding = (response.headers.get("Content-Encoding") or "identity").lower()
    if encoding not in {"identity", ""}:
        # Refuse compressed content at this boundary so decompression bombs cannot
        # expand after the wire-size check. A separate sandboxed transformer may
        # handle encoded content under its own compressed/decompressed limits.
        raise SafeFetchDenied("content_encoding_forbidden")

    media_type = (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower()
    if media_type not in ALLOWED_MEDIA_TYPES:
        raise SafeFetchDenied("content_type_forbidden")

    declared_length = response.headers.get("Content-Length")
    if declared_length is not None:
        try:
            if int(declared_length) > MAX_BODY_BYTES:
                raise SafeFetchDenied("declared_body_too_large")
        except ValueError as exc:
            raise SafeFetchDenied("invalid_content_length") from exc

    body = response.read(MAX_BODY_BYTES + 1, decode_content=False)
    if len(body) > MAX_BODY_BYTES:
        raise SafeFetchDenied("body_too_large")
    return body, media_type


def safe_fetch(raw_url: str) -> SafeFetchResult:
    current_url = canonicalize(raw_url)

    for redirect_count in range(MAX_REDIRECTS + 1):
        response = None
        try:
            # Validation and GET share this pinned connection; no second DNS
            # lookup or preview request occurs after the security decision.
            response, connected_ip = request_pinned(current_url)
            if response.status in {301, 302, 303, 307, 308}:
                location = response.headers.get("Location")
                if not location:
                    raise SafeFetchDenied("redirect_without_location")
                if redirect_count == MAX_REDIRECTS:
                    raise SafeFetchDenied("redirect_limit_exceeded")
                next_url = canonicalize(urljoin(current_url, location))
                response.close()
                response = None
                current_url = next_url
                continue

            if response.status < 200 or response.status >= 300:
                raise SafeFetchDenied(f"unexpected_http_status:{response.status}")

            body, media_type = read_bounded_body(response)
            return SafeFetchResult(
                final_url=current_url,
                connected_ip=connected_ip,
                media_type=media_type,
                body=body,
                body_sha256=hashlib.sha256(body).hexdigest(),
                redirect_count=redirect_count,
                policy_version=POLICY_VERSION,
            )
        finally:
            if response is not None:
                response.close()

    raise SafeFetchDenied("unreachable_redirect_state")
Mandatory service and evidence boundary

Deploy this fetcher as the only egress-capable retrieval service. Agent/application pods receive a default-deny egress policy that permits only the safe-fetch service (and the cluster DNS resolver when needed); direct Internet routes are denied. The safe-fetch policy file is signature-verified against a release-pinned digest before service start.

# File: k8s/agent-egress-only-safe-fetch.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: agent-egress-only-safe-fetch
  namespace: agent-runtime
spec:
  podSelector:
    matchLabels:
      app: ai-agent
  policyTypes: [Egress]
  egress:
    - to:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: safe-fetch
          podSelector:
            matchLabels:
              app: safe-fetch
      ports:
        - protocol: TCP
          port: 8443
    - to:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: kube-system
          podSelector:
            matchLabels:
              k8s-app: kube-dns
      ports:
        - protocol: UDP
          port: 53

Shared-service contract: Expose this exact pinned fetcher only through the fail-closed v2 service implementation below. The v2 boundary owns strict request/response schemas, authenticated caller identity, signed PASS/FAIL/ERROR decisions, exact returned-byte and final-URL bindings, and audit-writer failure behavior. A direct in-process caller must emit the same v2 decision fields to an independently authenticated audit writer and release no body unless the signed receipt is durably returned; do not deploy the obsolete /v1/fetch example or a separate verifier-then-refetch client.

Deployment and evidence requirements

Action: Remove verifier-then-refetch flows. Consume the body from the exact pinned request that passed validation and fail closed before any bytes reach the model or renderer.

AID-H-019.001-G002

Expose the pinned safe fetcher through a fail-closed v2 service contract that binds each permitted response to the final redirect target and exact returned bytes.

Use this service boundary when multiple RAG, preview, OCR, browser, or agent workloads consume the H-019 safe fetcher. The service is the only Internet-egress-capable workload; callers authenticate with mTLS and are denied direct external routes. It returns bytes only after a separate audit identity has durably signed the exact decision.

Issue a response bound to the final canonical URL
# File: safe_fetch_service/handler_v2.py
from __future__ import annotations

import hashlib
import ipaddress
import json
import math
import os
import re
import ssl

import httpx
import rfc8785
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response
from pydantic import BaseModel, ConfigDict, StrictStr, field_validator

from agent.safe_fetcher import (
    ALLOWED_MEDIA_TYPES,
    MAX_BODY_BYTES,
    MAX_REDIRECTS,
    POLICY_SHA256,
    POLICY_VERSION,
    SafeFetchDenied,
    canonicalize,
    safe_fetch,
)

app = FastAPI()
AUDIT_URL = os.environ["FETCH_AUDIT_WRITER_URL"]
SERVICE_IMAGE_DIGEST = os.environ["SAFE_FETCH_IMAGE_DIGEST"]
REQUEST_URL_MAX_BYTES = int(os.environ["VERIFIED_SAFE_FETCH_REQUEST_URL_MAX_BYTES"])
AUDIT_RESPONSE_MAX_BYTES = int(os.environ["VERIFIED_SAFE_FETCH_AUDIT_RESPONSE_MAX_BYTES"])
AUDIT_TOTAL_TIMEOUT_SECONDS = float(os.environ["VERIFIED_SAFE_FETCH_AUDIT_TOTAL_TIMEOUT_SECONDS"])
AUDIT_CONNECT_TIMEOUT_SECONDS = float(os.environ["VERIFIED_SAFE_FETCH_AUDIT_CONNECT_TIMEOUT_SECONDS"])
if (
    min(REQUEST_URL_MAX_BYTES, AUDIT_RESPONSE_MAX_BYTES) < 1
    or any(
        not math.isfinite(value) or value <= 0
        for value in (AUDIT_TOTAL_TIMEOUT_SECONDS, AUDIT_CONNECT_TIMEOUT_SECONDS)
    )
):
    raise RuntimeError("verified safe-fetch service bounds are invalid")
if not AUDIT_URL.startswith("https://"):
    raise RuntimeError("FETCH_AUDIT_WRITER_URL must use HTTPS")
if re.fullmatch(r"sha256:[0-9a-f]{64}", SERVICE_IMAGE_DIGEST) is None:
    raise RuntimeError("SAFE_FETCH_IMAGE_DIGEST must be an immutable sha256 digest")

SHA256_RE = re.compile(r"[0-9a-f]{64}")
RECEIPT_ID_RE = re.compile(r"[A-Za-z0-9._:-]{16,256}")
if SHA256_RE.fullmatch(POLICY_SHA256) is None:
    raise RuntimeError("safe-fetch policy digest must be lowercase sha256")
if not isinstance(POLICY_VERSION, str) or not POLICY_VERSION.strip():
    raise RuntimeError("safe-fetch policy version must be non-empty")
POLICY_DENIAL_CODES = {
    "https_required",
    "userinfo_forbidden",
    "hostname_required",
    "invalid_hostname_or_port",
    "hostname_not_allowlisted",
    "port_not_allowlisted",
    "control_character_in_url",
    "non_public_dns_answer",
    "redirect_limit_exceeded",
    "content_encoding_forbidden",
    "content_type_forbidden",
    "declared_body_too_large",
    "body_too_large",
}


class AuditUnavailable(RuntimeError):
    pass


class FetchRequest(BaseModel):
    # StrictStr preserves the exact submitted URL string for raw_url_sha256.
    # URL policy and canonicalization are performed only by canonicalize().
    model_config = ConfigDict(extra="forbid", strict=True)
    request_id: StrictStr
    url: StrictStr

    @field_validator("request_id")
    @classmethod
    def request_id_is_uuid(cls, value: str) -> str:
        import uuid

        try:
            parsed = uuid.UUID(value)
        except ValueError as exc:
            raise ValueError("request_id must be a canonical UUID") from exc
        if str(parsed) != value:
            raise ValueError("request_id must be a lowercase canonical UUID")
        return value

    @field_validator("url")
    @classmethod
    def url_is_utf8(cls, value: str) -> str:
        try:
            encoded = value.encode("utf-8")
        except UnicodeEncodeError as exc:
            raise ValueError("url must be valid UTF-8") from exc
        if not value or len(encoded) > REQUEST_URL_MAX_BYTES:
            raise ValueError("url is empty or exceeds the verified byte bound")
        return value


def strict_audit_json(payload: bytes) -> object:
    if len(payload) > AUDIT_RESPONSE_MAX_BYTES:
        raise AuditUnavailable("audit_writer_response_too_large")

    def unique_object(pairs: list[tuple[str, object]]) -> dict:
        result: dict = {}
        for key, value in pairs:
            if key in result:
                raise AuditUnavailable("audit_writer_duplicate_json_key")
            result[key] = value
        return result

    try:
        return json.loads(
            payload.decode("utf-8", errors="strict"),
            object_pairs_hook=unique_object,
            parse_constant=lambda value: (_ for _ in ()).throw(
                AuditUnavailable(f"audit_writer_invalid_constant:{value}")
            ),
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise AuditUnavailable("audit_writer_json_invalid") from exc


def retain_signed_decision(decision: dict) -> str:
    try:
        expected_decision_sha256 = hashlib.sha256(
            rfc8785.dumps(decision)
        ).hexdigest()
        audit_tls = ssl.create_default_context(cafile=os.environ["AUDIT_CA_BUNDLE"])
        audit_tls.load_cert_chain(
            os.environ["AUDIT_MTLS_CERT"], os.environ["AUDIT_MTLS_KEY"]
        )
    except Exception as exc:
        raise AuditUnavailable("audit_writer_setup_or_canonicalization_error") from exc
    try:
        with httpx.Client(
            verify=audit_tls,
            timeout=httpx.Timeout(AUDIT_TOTAL_TIMEOUT_SECONDS, connect=AUDIT_CONNECT_TIMEOUT_SECONDS),
            follow_redirects=False,
        ) as client:
            with client.stream("POST", AUDIT_URL, json=decision) as response:
                if response.status_code != 201:
                    raise AuditUnavailable("audit_writer_status_invalid")
                if response.headers.get("content-type", "").split(";", 1)[0].lower() != "application/json":
                    raise AuditUnavailable("audit_writer_content_type_invalid")
                raw = bytearray()
                for chunk in response.iter_bytes():
                    raw.extend(chunk)
                    if len(raw) > AUDIT_RESPONSE_MAX_BYTES:
                        raise AuditUnavailable("audit_writer_response_too_large")
        body = strict_audit_json(bytes(raw))
    except httpx.HTTPError as exc:
        raise AuditUnavailable("audit_writer_transport_or_json_error") from exc

    if not isinstance(body, dict) or set(body) != {
        "schema_version",
        "signed_receipt_id",
        "decision_sha256",
    }:
        raise AuditUnavailable("audit_writer_response_schema_invalid")
    receipt_id = body["signed_receipt_id"]
    if (
        body["schema_version"] != "aidefend.safe-fetch-audit-receipt.v2"
        or not isinstance(receipt_id, str)
        or RECEIPT_ID_RE.fullmatch(receipt_id) is None
        or body["decision_sha256"] != expected_decision_sha256
    ):
        raise AuditUnavailable("audit_writer_receipt_binding_invalid")
    return receipt_id


def base_decision(
    *,
    request: FetchRequest,
    caller_id: str,
    outcome: str,
    reason_code: str,
    canonical_initial_url_sha256: str | None,
) -> dict:
    return {
        "schema_version": "aidefend.safe-fetch-decision.v2",
        "request_id": request.request_id,
        "caller_id": caller_id,
        "outcome": outcome,
        "reason_code": reason_code,
        "raw_url_sha256": hashlib.sha256(request.url.encode("utf-8")).hexdigest(),
        "canonical_initial_url_sha256": canonical_initial_url_sha256,
        "final_url_sha256": None,
        "connected_ip": None,
        "redirect_count": 0,
        "media_type": None,
        "body_size": 0,
        "body_sha256": None,
        "fetch_policy_version": POLICY_VERSION,
        "fetch_policy_sha256": POLICY_SHA256,
        "service_image_digest": SERVICE_IMAGE_DIGEST,
    }


def audited_response(decision: dict, body: bytes | None = None) -> Response:
    outcome = decision.get("outcome")
    if outcome == "PASS" and (
        body is None
        or not isinstance(body, bytes)
        or hashlib.sha256(body).hexdigest() != decision.get("body_sha256")
        or len(body) != decision.get("body_size")
    ):
        decision = {
            **decision,
            "outcome": "ERROR",
            "reason_code": "response_body_binding_error",
            "final_url_sha256": None,
            "connected_ip": None,
            "redirect_count": 0,
            "media_type": None,
            "body_size": 0,
            "body_sha256": None,
        }
        body = None
        outcome = "ERROR"
    elif outcome in {"FAIL", "ERROR"}:
        body = None

    # No v2 PASS/FAIL/ERROR envelope is emitted until the independent audit
    # identity has durably returned a receipt bound to these exact fields.
    try:
        receipt_id = retain_signed_decision(decision)
    except Exception:
        # A receipt cannot be fabricated by the fetch service. An empty 503 lets
        # the caller record ERROR without accepting bytes; ingress/access logs
        # reconcile this attempted request into the complete population.
        return Response(status_code=503, content=b"")

    if outcome == "PASS":
        assert body is not None
        status_code = 200
        body_hex: str | None = body.hex()
    elif outcome == "FAIL":
        status_code = 403
        body_hex = None
    elif outcome == "ERROR":
        status_code = 502
        body_hex = None
    else:
        return Response(status_code=503, content=b"")

    envelope = {
        key: value
        for key, value in decision.items()
        if key not in {"schema_version", "caller_id"}
    }
    envelope["schema_version"] = "aidefend.safe-fetch-response.v2"
    envelope["body_hex"] = body_hex
    envelope["signed_receipt_id"] = receipt_id
    return JSONResponse(status_code=status_code, content=envelope)


@app.post("/v2/fetch")
def fetch(request: FetchRequest, transport: Request) -> Response:
    # Production middleware must set this from the verified client certificate.
    caller_id = transport.scope.get("verified_spiffe_id")
    if not isinstance(caller_id, str) or not caller_id:
        return Response(status_code=401, content=b"")

    canonical_initial_url_sha256 = None
    try:
        initial_url = canonicalize(request.url)
        canonical_initial_url_sha256 = hashlib.sha256(
            initial_url.encode("utf-8")
        ).hexdigest()
    except SafeFetchDenied as exc:
        reason_code = str(exc).split(":", 1)[0]
        decision = base_decision(
            request=request,
            caller_id=caller_id,
            outcome="FAIL",
            reason_code=reason_code,
            canonical_initial_url_sha256=None,
        )
        return audited_response(decision)

    try:
        result = safe_fetch(initial_url)
    except SafeFetchDenied as exc:
        reason_code = str(exc).split(":", 1)[0]
        outcome = "FAIL" if reason_code in POLICY_DENIAL_CODES else "ERROR"
        decision = base_decision(
            request=request,
            caller_id=caller_id,
            outcome=outcome,
            reason_code=reason_code,
            canonical_initial_url_sha256=canonical_initial_url_sha256,
        )
        return audited_response(decision)
    except Exception:
        decision = base_decision(
            request=request,
            caller_id=caller_id,
            outcome="ERROR",
            reason_code="fetch_transport_or_runtime_error",
            canonical_initial_url_sha256=canonical_initial_url_sha256,
        )
        return audited_response(decision)

    try:
        body = result.body
        final_url = result.final_url
        connected_ip = ipaddress.ip_address(result.connected_ip)
        if (
            not isinstance(body, bytes)
            or len(body) > MAX_BODY_BYTES
            or type(result.redirect_count) is not int
            or not 0 <= result.redirect_count <= MAX_REDIRECTS
            or result.media_type not in ALLOWED_MEDIA_TYPES
            or result.policy_version != POLICY_VERSION
            or SHA256_RE.fullmatch(result.body_sha256) is None
            or hashlib.sha256(body).hexdigest() != result.body_sha256
            or not connected_ip.is_global
            or canonicalize(final_url) != final_url
        ):
            raise ValueError("safe_fetch_result_contract_invalid")
    except Exception:
        decision = base_decision(
            request=request,
            caller_id=caller_id,
            outcome="ERROR",
            reason_code="safe_fetch_result_contract_invalid",
            canonical_initial_url_sha256=canonical_initial_url_sha256,
        )
        return audited_response(decision)

    decision = base_decision(
        request=request,
        caller_id=caller_id,
        outcome="PASS",
        reason_code="allowed",
        canonical_initial_url_sha256=canonical_initial_url_sha256,
    )
    decision.update(
        {
            "final_url_sha256": hashlib.sha256(
                final_url.encode("utf-8")
            ).hexdigest(),
            "connected_ip": str(connected_ip),
            "redirect_count": result.redirect_count,
            "media_type": result.media_type,
            "body_size": len(body),
            "body_sha256": result.body_sha256,
        }
    )
    return audited_response(decision, body)
Verify the audited response envelope

The audit writer accepts only the service's authenticated identity, rejects extra or missing decision fields, canonicalizes with RFC 8785, signs under a key unavailable to the fetch service, and stores append-only decision bytes. The service emits a structured 200/PASS, 403/FAIL, or 502/ERROR envelope only after receipt issuance; every FAIL/ERROR envelope has null body fields and body_size=0. An audit-writer or response-transport failure emits an empty 503, never releases bytes, and is recorded as ERROR by the caller because a signed v2 outcome cannot truthfully be claimed without the independent receipt.

raw_url_sha256 always hashes the exact submitted UTF-8 URL string. canonical_initial_url_sha256 hashes the normalized initial URL and is null when normalization fails. final_url_sha256 is present only for a successful final target. A verifier enumerates the complete request population from mTLS ingress, service, audit-writer, and consumer records; reads every receipt back through the authenticated audit API; verifies its signature and canonical decision digest; recomputes URL/body bindings; and reconciles status, redirect, policy, image, and outcome fields. Any missing receipt, unmatched request, invalid signature, digest drift, audit outage, or incomplete population blocks byte release and deployment acceptance. Exclude a workload only when a signed inventory rule proves that it performs no privileged URL retrieval.

Evidence: retain the canonical decision bytes/signature, receipt ID, caller SPIFFE ID, raw/canonical/final URL digests, connected IP, redirect count, media type, exact returned byte count/body digest, policy version/digest, service image digest, HTTP outcome, and independent complete-population reconciliation verdict. Never retain URL query secrets in clear text.

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-019.001: URL Normalization & Allowlist Filtering | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service