AID-DV-001: Honeypot AI Services & Decoy Models/APIs

Version 1.20261015 · Actionable technique

Deploy and isolate decoy AI systems such as fake LLM APIs, ML model endpoints serving synthetic or non-sensitive data, or imitation agent services that appear valuable, vulnerable, or legitimate to potential attackers; record every interaction without exposing production data or authority.

Intelligence Collection
These honeypots are instrumented for intensive monitoring to log all interactions, capture attacker TTPs (Tactics, Techniques, and Procedures), and gather threat intelligence without exposing real production systems or data. They can also be used to slow down attackers or waste their resources.

Operational Requirements
All honeypot services must be logically isolated from production networks, run with read-only/non-destructive behaviors, and be instrumented for forensic retention (full request capture, replayable context).

Pillars: infra, model, app

Lifecycle: operation

Scope boundary

Owns the isolated decoy AI service, its synthetic behavior, and the evidence generated by interactions with that service. It must not expose production data, authority, credentials, or a route to a real inference backend.

AID-DV-006: AID-DV-001 operates a complete decoy AI service; AID-DV-006 exposes narrow synthetic reconnaissance routes at a gateway or proxy. Use AID-DV-001 when the attacker interacts with an imitation model, agent, or API service. Use AID-DV-006 when only selected metadata or diagnostic-looking routes are populated and intercepted upstream.

Implementation guidance

AID-DV-001-G001

Deploy a manifest-defined decoy AI API whose advertised models and responses are static synthetic assets and whose runtime has no route, credential, import, or service binding to a real inference backend.

Runtime policy: Load the maximum request bytes from the signed, versioned synthetic manifest bound to this decoy route and release, bind its version/digest to admission and probe evidence, and return ERROR if the manifest or value is absent, invalid, or unverifiable.

When this method applies

Use this method only for an intentionally deployed decoy endpoint. The decoy may resemble the organization's public API shape, but every advertised model, owner, capability, and response must come from a signed synthetic manifest. It must never proxy an unknown route, load a production model, or inherit production workload credentials.

Signed manifest example: The request-size value below is illustrative signed-manifest data, not a framework default. Promotion renders it from the versioned deployment profile and the application requires and validates it without fallback.

Build and verify the decoy manifest

File: honeypot/decoy_manifest.json

{
  "schema_version": "aidefend.decoy-api.v1",
  "decoy_id": "dv001-public-api-usw2",
  "allowed_routes": ["/healthz", "/v1/models", "/v1/chat/completions"],
  "models": [
    {"id": "internal-reasoner-legacy", "object": "model", "owned_by": "research-platform"},
    {"id": "finance-assistant-preview", "object": "model", "owned_by": "automation-platform"}
  ],
  "chat_response": "Request accepted by the internal evaluation service. Provide the next operation.",
  "backend_mode": "static_only",
  "maximum_request_bytes": 1048576
}
#!/usr/bin/env bash
set -euo pipefail
umask 077

: "${DECOY_LISTEN_HOST:?required}"
: "${DECOY_LISTEN_PORT:?required}"
if [[ "$DECOY_LISTEN_HOST" =~ [[:cntrl:]] ]]; then
  echo "decoy listen host must not contain control characters" >&2
  exit 2
fi
if [[ ! "$DECOY_LISTEN_PORT" =~ ^[0-9]{1,5}$ ]] \
   || (( 10#$DECOY_LISTEN_PORT < 1 || 10#$DECOY_LISTEN_PORT > 65535 )); then
  echo "decoy listen port must contain only digits and be in 1..65535" >&2
  exit 2
fi

manifest=/run/policy/decoy_manifest.json
bundle=/run/policy/decoy_manifest.json.sigstore.json
verify_key=/run/trust/decoy-manifest.pub
verifier_deadline_ceiling_seconds=15
for input_path in "$manifest" "$bundle" "$verify_key"; do
  if [[ ! -f "$input_path" || -L "$input_path" ]]; then
    echo "manifest, bundle, and trust key must be regular non-symlink files" >&2
    exit 2
  fi
done
key_owner="$(stat -Lc '%u' -- "$verify_key")"
key_mode="$(stat -Lc '%a' -- "$verify_key")"
if (( 10#$key_owner != 0 || (8#$key_mode & 022) != 0 )); then
  echo "manifest trust key must be root-owned and not group/world writable" >&2
  exit 2
fi
command -v timeout >/dev/null || { echo "GNU timeout is required" >&2; exit 2; }

snapshot_root="$(mktemp -d)"
chmod 0700 "$snapshot_root"
trap 'rm -rf -- "$snapshot_root"' EXIT
manifest_snapshot="$snapshot_root/decoy_manifest.json"
bundle_snapshot="$snapshot_root/decoy_manifest.json.sigstore.json"
key_snapshot="$snapshot_root/decoy-manifest.pub"
install -m 0400 -- "$manifest" "$manifest_snapshot"
install -m 0400 -- "$bundle" "$bundle_snapshot"
install -m 0400 -- "$verify_key" "$key_snapshot"
manifest_sha256="$(sha256sum -- "$manifest_snapshot" | awk '{print $1}')"
bundle_sha256="$(sha256sum -- "$bundle_snapshot" | awk '{print $1}')"
key_sha256="$(sha256sum -- "$key_snapshot" | awk '{print $1}')"

timeout --signal=TERM --kill-after=2s "${verifier_deadline_ceiling_seconds}s" \
  cosign verify-blob --timeout "${verifier_deadline_ceiling_seconds}s" \
  --key "$key_snapshot" --bundle "$bundle_snapshot" "$manifest_snapshot" >/dev/null
if [[ "$(sha256sum -- "$manifest_snapshot" | awk '{print $1}')" != "$manifest_sha256" \
   || "$(sha256sum -- "$bundle_snapshot" | awk '{print $1}')" != "$bundle_sha256" \
   || "$(sha256sum -- "$key_snapshot" | awk '{print $1}')" != "$key_sha256" ]]; then
  echo "decoy manifest snapshot changed during verification" >&2
  exit 1
fi

export DECOY_MANIFEST_PATH="$manifest_snapshot"
export DECOY_MANIFEST_SHA256="$manifest_sha256"
python -m uvicorn honeypot.app:app \
  --host "$DECOY_LISTEN_HOST" \
  --port "$DECOY_LISTEN_PORT" \
  --no-access-log
Serve only the allowlisted static routes

File: honeypot/app.py

from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse


def reject_duplicate_keys(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate manifest JSON key: {key}")
        value[key] = item
    return value


def reject_nonfinite(value):
    raise ValueError(f"non-finite manifest JSON value: {value}")

MANIFEST_PATH = Path(os.environ["DECOY_MANIFEST_PATH"])
EXPECTED_SHA256 = os.environ["DECOY_MANIFEST_SHA256"].lower()
raw = MANIFEST_PATH.read_bytes()
if hashlib.sha256(raw).hexdigest() != EXPECTED_SHA256:
    raise RuntimeError("verified decoy manifest digest differs at runtime")
manifest = json.loads(
    raw.decode("utf-8", errors="strict"),
    object_pairs_hook=reject_duplicate_keys,
    parse_constant=reject_nonfinite,
)
expected_fields = {
    "schema_version", "decoy_id", "allowed_routes", "models",
    "chat_response", "backend_mode", "maximum_request_bytes",
}
if not isinstance(manifest, dict) or set(manifest) != expected_fields:
    raise RuntimeError("decoy manifest fields differ")
if manifest.get("schema_version") != "aidefend.decoy-api.v1":
    raise RuntimeError("unsupported decoy manifest")
if manifest.get("backend_mode") != "static_only":
    raise RuntimeError("decoy backend must be static_only")
if not isinstance(manifest["decoy_id"], str) or not manifest["decoy_id"].strip():
    raise RuntimeError("decoy_id must be a nonempty string")
if not isinstance(manifest["chat_response"], str) or not manifest["chat_response"].strip():
    raise RuntimeError("chat_response must be a nonempty static string")
models = manifest["models"]
if (
    not isinstance(models, list) or not models
    or any(not isinstance(item, dict) or set(item) != {"id", "object", "owned_by"} for item in models)
    or any(not all(isinstance(item[field], str) and item[field].strip() for field in item) for item in models)
    or len({item["id"] for item in models}) != len(models)
):
    raise RuntimeError("decoy model population is invalid")
maximum_request_bytes = manifest.get("maximum_request_bytes")
if (isinstance(maximum_request_bytes, bool)
        or not isinstance(maximum_request_bytes, int)
        or maximum_request_bytes < 1):
    raise RuntimeError("signed decoy request-size bound is invalid")
if sorted(manifest.get("allowed_routes", [])) != ["/healthz", "/v1/chat/completions", "/v1/models"]:
    raise RuntimeError("decoy route population differs")

app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None)

@app.middleware("http")
async def reject_unregistered_routes(request: Request, call_next):
    if request.url.path not in manifest["allowed_routes"]:
        return JSONResponse({"detail": "not found"}, status_code=404)
    return await call_next(request)

@app.get("/healthz")
def healthz():
    return {"status": "ready", "decoy_id": manifest["decoy_id"], "manifest_sha256": EXPECTED_SHA256}

@app.get("/v1/models")
def list_models():
    return {"object": "list", "data": manifest["models"]}

async def consume_bounded_body(request: Request) -> int:
    claimed = request.headers.get("content-length")
    if claimed is not None:
        try:
            claimed_bytes = int(claimed, 10)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail="invalid content-length") from exc
        if claimed_bytes < 0 or claimed_bytes > maximum_request_bytes:
            raise HTTPException(status_code=413, detail="request too large")
    observed = 0
    async for chunk in request.stream():
        observed += len(chunk)
        if observed > maximum_request_bytes:
            raise HTTPException(status_code=413, detail="request too large")
    return observed


@app.post("/v1/chat/completions")
async def chat_completion(request: Request):
    await consume_bounded_body(request)
    return {
        "object": "chat.completion",
        "choices": [{"index": 0, "message": {"role": "assistant", "content": manifest["chat_response"]}}],
    }
Independently verify the result

From an independent deployment identity, enumerate the running image digest, service account, effective routes, environment, mounts, network policy, and every returned model ID. Search the image and deployment for production SDK credentials, model mounts, production DNS names, and wildcard proxy routes. Retain the signed manifest and bundle, image digest, runtime manifest readback, route/model-set comparison, and negative connectivity-test evidence.

Action: Promote only the exact digest-bound decoy image and signed synthetic manifest that passed independent route, binding, and connectivity replay.

The handler rejects an oversized declared length before reading and counts each ASGI body chunk before accepting it, so a chunked attacker cannot force an unbounded request.body() allocation.

The verifier snapshots the fixed root-owned trust key with the manifest and bundle, and both Cosign v3.1.2 and the supervising process enforce a finite bootstrap deadline.

AID-DV-001-G002

Capture every decoy interaction through a bounded, privacy-aware evidence middleware that commits a digest-bound event to an append-only outbox before responding and verifies durable external readback.

Runtime policy: Load the capture-body ceiling, identity bounds, outbox, and SQLite timeout from the fixed root-controlled versioned ingest profile bound to this decoy deployment, bind its version/digest to each event and outbox readback, and return ERROR if the profile or either value is absent, invalid, or unverifiable.

Required inputs and boundaries

This method captures forensic evidence for the complete request population reaching the decoy ingress. It depends on a trusted gateway supplying an authenticated request identifier; forwarded headers are evidence only when the gateway strips caller-supplied copies and mTLS-authenticates the metadata channel. Do not place authorization headers, cookies, raw credentials, or unrestricted bodies in a general SIEM.

Capture bounded evidence before the response commits
# File: honeypot/evidence.py
from __future__ import annotations

import asyncio
import hashlib
import hmac
import json
import math
import os
import re
import sqlite3
import stat
import time
from pathlib import Path

from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.types import ASGIApp, Message, Receive, Scope, Send


PROFILE_PATH = Path("/run/policy/dv001-evidence-profile.json")
PROFILE_ARTIFACT_CEILING = 131_072  # Bootstrap ceiling; operational values come from the authenticated profile.
PROFILE_FIELDS = {
    "schema_version", "profile_version", "outbox_path",
    "maximum_body_bytes", "maximum_body_messages",
    "maximum_request_id_bytes", "maximum_header_value_bytes",
    "body_read_timeout_seconds", "sqlite_busy_timeout_seconds",
}
SAFE_HEADERS = {"content-type", "user-agent", "accept", "x-request-id"}


def unique_object(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate profile key: {key}")
        value[key] = item
    return value


def read_privileged(path: Path, ceiling: int) -> bytes:
    if not hasattr(os, "O_NOFOLLOW"):
        raise OSError("secure no-follow reads are unavailable")
    descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        before = os.fstat(descriptor)
        if (not stat.S_ISREG(before.st_mode) or before.st_uid != 0
                or before.st_mode & 0o022 or not 1 <= before.st_size <= ceiling):
            raise PermissionError("runtime profile must be bounded, root-owned, and non-writable")
        chunks, observed = [], 0
        while observed <= ceiling:
            chunk = os.read(descriptor, min(65_536, ceiling + 1 - observed))
            if not chunk:
                break
            chunks.append(chunk)
            observed += len(chunk)
        after = os.fstat(descriptor)
        fingerprint = lambda item: (
            item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns,
        )
        if observed != before.st_size or observed > ceiling or fingerprint(before) != fingerprint(after):
            raise OSError("runtime profile changed during read")
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def load_profile() -> tuple[dict, str]:
    raw = read_privileged(PROFILE_PATH, PROFILE_ARTIFACT_CEILING)
    profile = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError("non-finite profile value: " + value)),
    )
    if not isinstance(profile, dict) or set(profile) != PROFILE_FIELDS:
        raise RuntimeError("evidence profile schema differs")
    if profile["schema_version"] != "aidefend.dv001.evidence-profile.v1":
        raise RuntimeError("unsupported evidence profile")
    if not isinstance(profile["profile_version"], str) or not profile["profile_version"].strip():
        raise RuntimeError("profile version is missing")
    for field in (
        "maximum_body_bytes", "maximum_body_messages",
        "maximum_request_id_bytes", "maximum_header_value_bytes",
    ):
        if type(profile[field]) is not int or profile[field] < 1:
            raise RuntimeError(f"invalid authenticated bound: {field}")
    for field in ("body_read_timeout_seconds", "sqlite_busy_timeout_seconds"):
        value = profile[field]
        if (type(value) not in (int, float) or isinstance(value, bool)
                or not math.isfinite(float(value)) or value <= 0):
            raise RuntimeError(f"invalid authenticated timeout: {field}")
    outbox = Path(profile["outbox_path"])
    if not isinstance(profile["outbox_path"], str) or not outbox.is_absolute():
        raise RuntimeError("outbox_path must be an absolute authenticated path")
    return profile, hashlib.sha256(raw).hexdigest()


PROFILE, PROFILE_SHA256 = load_profile()
OUTBOX = Path(PROFILE["outbox_path"])


def initialize() -> None:
    OUTBOX.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(OUTBOX, timeout=float(PROFILE["sqlite_busy_timeout_seconds"])) as database:
        database.execute("PRAGMA journal_mode=WAL")
        database.execute("PRAGMA synchronous=FULL")
        database.execute(
            "CREATE TABLE IF NOT EXISTS events ("
            "event_id TEXT PRIMARY KEY, canonical_json BLOB NOT NULL, "
            "payload_sha256 TEXT NOT NULL, created_ns INTEGER NOT NULL, "
            "external_object_version TEXT, external_readback_sha256 TEXT)"
        )


def canonical_bytes(value: dict) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")


def commit_event(event: dict) -> None:
    payload = canonical_bytes(event)
    with sqlite3.connect(OUTBOX, timeout=float(PROFILE["sqlite_busy_timeout_seconds"])) as database:
        database.execute("PRAGMA synchronous=FULL")
        try:
            database.execute(
                "INSERT INTO events(event_id,canonical_json,payload_sha256,created_ns) VALUES(?,?,?,?)",
                (event["event_id"], payload, hashlib.sha256(payload).hexdigest(), time.time_ns()),
            )
        except sqlite3.IntegrityError:
            existing = database.execute(
                "SELECT canonical_json FROM events WHERE event_id = ?",
                (event["event_id"],),
            ).fetchone()
            if existing is None or not hmac.compare_digest(bytes(existing[0]), payload):
                raise RuntimeError("request ID was reused for different decoy evidence")


async def reject(scope: Scope, receive: Receive, send: Send, status_code: int, detail: str) -> None:
    await JSONResponse({"detail": detail}, status_code=status_code)(scope, receive, send)


class EvidenceMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        request = Request(scope)
        body_ceiling = PROFILE["maximum_body_bytes"]
        message_ceiling = PROFILE["maximum_body_messages"]
        content_lengths = [
            value for name, value in scope.get("headers", [])
            if name.lower() == b"content-length"
        ]
        if len(content_lengths) > 1:
            await reject(scope, receive, send, 400, "duplicate content-length")
            return
        claimed_bytes: int | None = None
        if content_lengths:
            try:
                claimed = content_lengths[0].decode("ascii", errors="strict")
                if not claimed.isdigit():
                    raise ValueError("content-length is not decimal")
                claimed_bytes = int(claimed, 10)
            except (UnicodeDecodeError, ValueError):
                await reject(scope, receive, send, 400, "invalid content-length")
                return
            if claimed_bytes > body_ceiling:
                await reject(scope, receive, send, 413, "request too large")
                return

        chunks: list[bytes] = []
        observed = 0
        messages = 0
        more_body = True
        stream_error: tuple[int, str] | None = None
        try:
            async with asyncio.timeout(float(PROFILE["body_read_timeout_seconds"])):
                while more_body:
                    message = await receive()
                    messages += 1
                    if messages > message_ceiling:
                        stream_error = (400, "too many body messages")
                        break
                    if message["type"] == "http.disconnect":
                        return
                    if message["type"] != "http.request":
                        stream_error = (400, "invalid ASGI request stream")
                        break
                    chunk = bytes(message.get("body", b""))
                    observed += len(chunk)
                    if observed > body_ceiling:
                        stream_error = (413, "request too large")
                        break
                    chunks.append(chunk)
                    more_body = bool(message.get("more_body", False))
        except TimeoutError:
            await reject(scope, receive, send, 408, "request body deadline exceeded")
            return
        if stream_error is not None:
            await reject(scope, receive, send, *stream_error)
            return
        if claimed_bytes is not None and observed != claimed_bytes:
            await reject(scope, receive, send, 400, "content-length mismatch")
            return
        body = b"".join(chunks)

        request_id_headers = [
            value for name, value in scope.get("headers", [])
            if name.lower() == b"x-request-id"
        ]
        if len(request_id_headers) != 1:
            await reject(scope, receive, send, 503, "exactly one trusted request id is required")
            return
        try:
            request_id = request_id_headers[0].decode("utf-8", errors="strict").strip()
        except UnicodeDecodeError:
            await reject(scope, receive, send, 503, "trusted request id is invalid")
            return
        if (not request_id or len(request_id.encode("utf-8")) > PROFILE["maximum_request_id_bytes"]):
            await reject(scope, receive, send, 503, "trusted request id missing or invalid")
            return
        safe_headers = {}
        for key, value in request.headers.items():
            lowered = key.lower()
            if lowered in SAFE_HEADERS:
                if len(value.encode("utf-8")) > PROFILE["maximum_header_value_bytes"]:
                    await reject(scope, receive, send, 431, "captured header exceeds authenticated bound")
                    return
                safe_headers[lowered] = value
        event = {
            "schema_version": "aidefend.decoy-interaction.v1",
            "runtime_profile_version": PROFILE["profile_version"],
            "runtime_profile_sha256": PROFILE_SHA256,
            "event_id": hashlib.sha256(canonical_bytes({
                "runtime_profile_sha256": PROFILE_SHA256,
                "request_id": request_id,
            })).hexdigest(),
            "request_id": request_id,
            "method": request.method,
            "path": request.url.path,
            "source_peer": request.client.host if request.client else None,
            "headers": safe_headers,
            "body_sha256": hashlib.sha256(body).hexdigest(),
            "body_length": len(body),
        }
        try:
            commit_event(event)
        except Exception:
            await reject(scope, receive, send, 503, "evidence unavailable")
            return

        delivered = False

        async def replay_receive() -> Message:
            nonlocal delivered
            if delivered:
                return {"type": "http.disconnect"}
            delivered = True
            return {"type": "http.request", "body": body, "more_body": False}

        await self.app(scope, replay_receive, send)


def install(app) -> None:
    initialize()
    app.add_middleware(EvidenceMiddleware)
External retention and independent readback

A separate least-privilege forwarder reads committed rows, writes each canonical event to retention-locked object storage using event_id as the idempotency key, records the returned object version/checksum, and reads the exact version back through an independent evidence-reader identity. The forwarder may place encrypted raw bodies in a separately authorized evidence vault only when the signed retention/privacy policy permits it; the normal event retains only the digest and size. Periodically reconcile gateway request IDs, outbox rows, and external objects as exact sets.

Action: Install the middleware before all decoy routes, fail the decoy request when evidence cannot commit, and retain the exact-population reconciliation receipt as the guidance evidence.

The edge proxy must strip any client-supplied x-request-id and inject its own authenticated value. The ASGI middleware captures at most the verified profile bound, commits evidence, and then replays those exact bytes to the application; oversized or incomplete streams never reach the route.

AID-DV-001-G003

Design honeypots to mimic production services but ensure strict network isolation.
When this method applies

Assume the decoy workload will be compromised. Place it in a separate cloud account/project or equivalent trust zone with no peering, transit, private endpoint, shared identity, or route to production. At the workload boundary, deny all ingress and egress except the inspected ingress tier, cluster DNS, and one monitored egress proxy. The proxy returns synthetic responses where realism is needed; the decoy never receives unrestricted Internet access.

Enforce a default-deny workload policy
# File: deploy/honeypot-network-policy.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: honeypot-default-deny
  namespace: ai-decoy
spec:
  podSelector: {}
  policyTypes: [Ingress, Egress]
---
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: honeypot-explicit-flows
  namespace: ai-decoy
spec:
  podSelector:
    matchLabels:
      app.kubernetes.io/name: ai-decoy
  policyTypes: [Ingress, Egress]
  ingress:
    - from:
        - namespaceSelector:
            matchLabels:
              aidefend.io/security-zone: inspected-ingress
          podSelector:
            matchLabels:
              app.kubernetes.io/name: decoy-gateway
      ports:
        - {protocol: TCP, port: 8443}
  egress:
    - to:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: kube-system
          podSelector:
            matchLabels:
              k8s-app: kube-dns
      ports:
        - {protocol: UDP, port: 53}
        - {protocol: TCP, port: 53}
    - to:
        - namespaceSelector:
            matchLabels:
              aidefend.io/security-zone: inspected-egress
          podSelector:
            matchLabels:
              app.kubernetes.io/name: decoy-egress-proxy
      ports:
        - {protocol: TCP, port: 8443}
Verify effective isolation, not only source YAML

Pin and sign the namespace, workload, gateway, proxy, and policy manifests. From a fresh running decoy pod, test every signed production CIDR/DNS target, cloud metadata endpoint, secret manager, registry, control-plane endpoint, public IP, and the permitted proxy endpoint. Require production, metadata, and direct-public probes to fail, and the mTLS proxy probe to succeed. Independently export effective cloud routes, peerings, firewall rules, service account bindings, Kubernetes policies, and flow-log coverage. Reconcile the complete running decoy-pod population to the signed expected inventory.

Action: Deploy the decoy only after the independent exact-population and negative-connectivity receipt passes; quarantine it immediately if a forbidden flow later appears.

AID-DV-001-G004

Apply a signed, decoy-only engagement profile that adds bounded deterministic jitter, rate limits, synthetic soft failures, and static low-value responses without consuming a real model or creating an attacker-controlled resource-exhaustion path.

Runtime policy: Load minimum/maximum delay and probability basis points from the signed, versioned friction policy bound to this decoy route, bind its version/digest to every decision and HMAC replay, and return ERROR if the policy or any value is absent, invalid, or unverifiable.

Required inputs and boundaries

This method depends on the signed decoy deployment identity and evidence middleware above. It must reject execution outside an attested decoy service. Derive response selection deterministically from a server-held HMAC key and request ID so an investigator can replay the profile without logging the secret or relying on non-reproducible randomness. Bounds come from a signed capacity policy, not hard-coded universal values.

Implement bounded and replayable friction
# File: honeypot/friction.py
from __future__ import annotations

import asyncio
import hashlib
import hmac
import json
import math
import os
import re
import stat
import time
from pathlib import Path

from redis.asyncio import Redis


PROFILE_PATH = Path("/run/policy/dv001-friction-profile.json")
SECRET_PATH = Path("/run/secrets/deception/friction-hmac.key")
RUNTIME_ID_PATH = Path("/run/identity/decoy-runtime-id")
BOOTSTRAP_ARTIFACT_CEILING = 131_072
PROFILE_FIELDS = {
    "schema_version", "profile_version", "decoy_runtime_id", "secret_key_id",
    "secret_key_sha256", "expires_at_epoch", "minimum_delay_ms",
    "maximum_delay_ms", "soft_failure_basis_points",
    "maximum_requests_per_window", "rate_limit_window_ms",
    "rate_store_timeout_ms", "request_id_maximum_bytes", "source_id_maximum_bytes",
}


RATE_LIMIT = """
local current = redis.call('INCR', KEYS[1])
if current == 1 then
  redis.call('PEXPIRE', KEYS[1], ARGV[1])
end
return {current, redis.call('PTTL', KEYS[1])}
"""


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate profile key: {key}")
        result[key] = value
    return result


def read_privileged(path: Path, ceiling: int) -> bytes:
    if not hasattr(os, "O_NOFOLLOW"):
        raise OSError("secure no-follow reads are unavailable")
    descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        before = os.fstat(descriptor)
        if (not stat.S_ISREG(before.st_mode) or before.st_uid != 0
                or before.st_mode & 0o022 or not 1 <= before.st_size <= ceiling):
            raise PermissionError(f"untrusted privileged artifact: {path}")
        chunks, observed = [], 0
        while observed <= ceiling:
            chunk = os.read(descriptor, min(65_536, ceiling + 1 - observed))
            if not chunk:
                break
            chunks.append(chunk)
            observed += len(chunk)
        after = os.fstat(descriptor)
        fingerprint = lambda item: (item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns)
        if observed != before.st_size or observed > ceiling or fingerprint(before) != fingerprint(after):
            raise OSError(f"privileged artifact changed during read: {path}")
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def load_contract() -> tuple[dict, str, bytes]:
    raw = read_privileged(PROFILE_PATH, BOOTSTRAP_ARTIFACT_CEILING)
    profile = json.loads(
        raw.decode("utf-8", errors="strict"), object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError("non-finite profile value: " + value)),
    )
    if not isinstance(profile, dict) or set(profile) != PROFILE_FIELDS:
        raise RuntimeError("friction profile schema differs")
    if profile["schema_version"] != "aidefend.dv001.friction-profile.v1":
        raise RuntimeError("unsupported friction profile")
    for field in ("profile_version", "decoy_runtime_id", "secret_key_id"):
        if not isinstance(profile[field], str) or not profile[field].strip():
            raise RuntimeError(f"authenticated profile field is invalid: {field}")
    integers = (
        "expires_at_epoch", "minimum_delay_ms", "maximum_delay_ms",
        "soft_failure_basis_points", "maximum_requests_per_window",
        "rate_limit_window_ms", "rate_store_timeout_ms",
        "request_id_maximum_bytes", "source_id_maximum_bytes",
    )
    if any(type(profile[field]) is not int for field in integers):
        raise RuntimeError("friction numeric fields must be integers")
    if not (0 <= profile["minimum_delay_ms"] <= profile["maximum_delay_ms"]):
        raise RuntimeError("authenticated delay bounds are invalid")
    if not 0 <= profile["soft_failure_basis_points"] <= 10_000:
        raise RuntimeError("authenticated soft-failure rate is invalid")
    if min(profile[field] for field in (
        "maximum_requests_per_window", "rate_limit_window_ms", "rate_store_timeout_ms",
        "request_id_maximum_bytes", "source_id_maximum_bytes",
    )) < 1:
        raise RuntimeError("authenticated friction bound is invalid")
    if profile["expires_at_epoch"] <= int(time.time()):
        raise RuntimeError("friction profile has expired")
    runtime_id = read_privileged(RUNTIME_ID_PATH, BOOTSTRAP_ARTIFACT_CEILING).decode("utf-8", errors="strict").strip()
    if not hmac.compare_digest(runtime_id, profile["decoy_runtime_id"]):
        raise RuntimeError("profile is not bound to this root-controlled runtime identity")
    secret = read_privileged(SECRET_PATH, BOOTSTRAP_ARTIFACT_CEILING)
    if len(secret) < 32 or not hmac.compare_digest(hashlib.sha256(secret).hexdigest(), profile["secret_key_sha256"]):
        raise RuntimeError("fixed friction secret does not match the authenticated profile")
    return profile, hashlib.sha256(raw).hexdigest(), secret


POLICY, PROFILE_SHA256, PROFILE_KEY = load_contract()


def require_current_policy() -> None:
    if int(time.time()) >= POLICY["expires_at_epoch"]:
        raise PermissionError("friction profile expired")


def decision(request_id: str) -> dict:
    require_current_policy()
    if (not isinstance(request_id, str) or not request_id
            or len(request_id.encode("utf-8")) > POLICY["request_id_maximum_bytes"]):
        raise ValueError("trusted request identity is invalid")
    digest = hmac.new(PROFILE_KEY, request_id.encode("utf-8"), hashlib.sha256).digest()
    span = POLICY["maximum_delay_ms"] - POLICY["minimum_delay_ms"] + 1
    return {
        "delay_ms": POLICY["minimum_delay_ms"] + (int.from_bytes(digest[:4], "big") % span),
        "soft_failure": int.from_bytes(digest[4:8], "big") % 10_000 < POLICY["soft_failure_basis_points"],
        "reply_index": int.from_bytes(digest[8:12], "big") % 3,
    }


async def apply_friction(request_id: str, source_id: str, rate_store: Redis) -> tuple[int, dict, dict]:
    selected = decision(request_id)
    if (not isinstance(source_id, str) or not source_id
            or len(source_id.encode("utf-8")) > POLICY["source_id_maximum_bytes"]):
        raise ValueError("trusted rate-limit source identity is invalid")
    source_digest = hmac.new(PROFILE_KEY, source_id.encode("utf-8"), hashlib.sha256).hexdigest()
    request_digest = hmac.new(PROFILE_KEY, request_id.encode("utf-8"), hashlib.sha256).hexdigest()
    key = f"dv001:friction:{PROFILE_SHA256}:{source_digest}"
    try:
        async with asyncio.timeout(POLICY["rate_store_timeout_ms"] / 1000):
            count, ttl = await rate_store.eval(RATE_LIMIT, 1, key, POLICY["rate_limit_window_ms"])
    except TimeoutError as exc:
        raise RuntimeError("friction rate store timed out") from exc
    if int(ttl) <= 0:
        raise RuntimeError("friction rate-limit TTL was not established")

    if int(count) > POLICY["maximum_requests_per_window"]:
        status, response = 429, {"detail": "request rate exceeded"}
    else:
        await asyncio.sleep(selected["delay_ms"] / 1000)
        if selected["soft_failure"]:
            status, response = 503, {"detail": "evaluation capacity temporarily unavailable"}
        else:
            replies = (
                "The evaluation request was accepted. Supply the next operation.",
                "The requested artifact is being staged. Confirm the requested scope.",
                "The internal workflow is ready. Provide the target identifier.",
            )
            status = 200
            response = {"object": "chat.completion", "content": replies[selected["reply_index"]]}
    evidence = {
        "schema_version": "aidefend.dv001.friction-decision.v1",
        "runtime_profile_version": POLICY["profile_version"],
        "runtime_profile_sha256": PROFILE_SHA256,
        "decoy_runtime_id": POLICY["decoy_runtime_id"],
        "secret_key_id": POLICY["secret_key_id"],
        "request_hmac_sha256": request_digest,
        "source_hmac_sha256": source_digest,
        "rate_count": int(count), "rate_ttl_ms": int(ttl),
        "delay_ms": selected["delay_ms"], "soft_failure": selected["soft_failure"],
        "status_code": status,
    }
    return status, response, evidence
Verify the result

Verify the signed policy, runtime identity, limiter configuration, and exact image digest before enabling the profile. Replay a fixed request-ID corpus and prove exact delay/failure/reply choices, upper bounds, per-source rate limiting, zero calls to model backends, and no impact on production routes. Retain policy and image digests, request/decision digests, limiter counters, response status/digest, and independent replay results.

Action: Enable friction only behind the attested decoy route and continuously reconcile its decisions and resource bounds against the signed profile.

Pass source_id only from a normalized network peer or authenticated principal that the caller cannot overwrite. The atomic Redis script both increments and starts the signed window; timeout, missing TTL, or rate-store failure fails closed instead of silently skipping the declared request ceiling.

AID-DV-001-G005

Integrate honeypot telemetry with central security monitoring (SIEM/SOC).
Required inputs and boundaries

Use the digest-bound interaction event produced by the preceding evidence middleware as the only source event. A separate least-privilege forwarder delivers each committed event to the SIEM with event_id as the idempotency key and marks delivery only after a verified broker or API acknowledgement. SIEM delivery does not replace the independently retained forensic object.

Verify the result

On every release and on the signed operational cadence, send a synthetic request with a unique drill ID through the real decoy ingress. Query the SIEM using an independent read identity and require exactly one event whose event ID, request ID, payload digest, decoy ID, parser version, and source outbox row match. Reconcile the full undelivered/delivered outbox population and alert on age, duplicate, dead-letter, schema-reject, or acknowledgement mismatch. Retain the signed drill policy, request receipt, outbox row digest, broker acknowledgement, SIEM event ID, query result, latency, and external forensic-object version.

Action: Operate the durable forwarder and exact-population reconciliation continuously; never delete or mark an event delivered from a transport success code alone.

AID-DV-001-G006

Detect high-confidence jailbreak tripwires with a signed, versioned rule pack and return a rule-bound static engagement response while retaining prompt content only in an authorized evidence vault.

Runtime policy: Load rule-pack and input bounds from the fixed root-controlled, versioned detector profile bound to this decoy route and rule-pack digest, bind its version/digest to findings and boundary-test evidence, and return ERROR if the profile or value is absent, invalid, or unverifiable.

Scope and safeguards

This is a honeypot tripwire, not a production prompt-injection detector. Use narrow, pre-tested rules whose purpose is to enrich decoy intelligence. Broad fragments such as act as or short names create unacceptable false positives. The decoy returns only signed-template text and never grants a capability or executes the requested instruction.

Load a signed rule pack and emit a bounded finding
# File: honeypot/jailbreak_tripwire.py
from __future__ import annotations

import hashlib
import hmac
import json
import math
import os
import stat
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path

import re2  # google-re2; pin the wheel and hashes in the deployment lockfile.


PROFILE_PATH = Path("/run/policy/dv001-jailbreak-profile.json")
RULE_VERIFY_KEY_PATH = Path("/run/trust/dv001-jailbreak-rules.pub")
PROFILE_ARTIFACT_CEILING = 131_072
PROFILE_FIELDS = {
    "schema_version", "profile_version", "rule_pack_path", "rule_pack_bundle_path",
    "rule_pack_sha256", "rule_pack_maximum_bytes", "bundle_maximum_bytes",
    "public_key_maximum_bytes", "rule_count_maximum", "pattern_maximum_chars",
    "field_maximum_chars", "prompt_maximum_chars", "prompt_maximum_bytes",
    "cosign_deadline_seconds",
}


@dataclass(frozen=True)
class Match:
    rule_id: str
    prompt_sha256: str
    response_template_id: str
    rule_pack_sha256: str
    runtime_profile_version: str
    runtime_profile_sha256: str


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def read_stable(path: Path, ceiling: int, *, privileged: bool = False) -> bytes:
    if not hasattr(os, "O_NOFOLLOW"):
        raise OSError("secure no-follow reads are unavailable")
    descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        before = os.fstat(descriptor)
        if not stat.S_ISREG(before.st_mode) or not 1 <= before.st_size <= ceiling:
            raise ValueError(f"artifact is empty, oversized, or not regular: {path}")
        if privileged and (before.st_uid != 0 or before.st_mode & 0o022):
            raise PermissionError(f"privileged artifact is not root-controlled: {path}")
        chunks, observed = [], 0
        while observed <= ceiling:
            chunk = os.read(descriptor, min(65_536, ceiling + 1 - observed))
            if not chunk:
                break
            chunks.append(chunk)
            observed += len(chunk)
        after = os.fstat(descriptor)
        fingerprint = lambda item: (item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns)
        if observed != before.st_size or observed > ceiling or fingerprint(before) != fingerprint(after):
            raise OSError(f"artifact changed during read: {path}")
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def load_profile() -> tuple[dict, str]:
    raw = read_stable(PROFILE_PATH, PROFILE_ARTIFACT_CEILING, privileged=True)
    profile = json.loads(
        raw.decode("utf-8", errors="strict"), object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError("non-finite profile value: " + value)),
    )
    if not isinstance(profile, dict) or set(profile) != PROFILE_FIELDS:
        raise RuntimeError("detector profile schema differs")
    if profile["schema_version"] != "aidefend.dv001.jailbreak-profile.v1":
        raise RuntimeError("unsupported detector profile")
    if not isinstance(profile["profile_version"], str) or not profile["profile_version"].strip():
        raise RuntimeError("detector profile version is missing")
    for field in (
        "rule_pack_maximum_bytes", "bundle_maximum_bytes", "public_key_maximum_bytes",
        "rule_count_maximum", "pattern_maximum_chars", "field_maximum_chars",
        "prompt_maximum_chars", "prompt_maximum_bytes",
    ):
        if type(profile[field]) is not int or profile[field] < 1:
            raise RuntimeError(f"invalid authenticated detector bound: {field}")
    deadline = profile["cosign_deadline_seconds"]
    if (type(deadline) not in (int, float) or isinstance(deadline, bool)
            or not math.isfinite(float(deadline)) or deadline <= 0):
        raise RuntimeError("invalid authenticated Cosign deadline")
    for field in ("rule_pack_path", "rule_pack_bundle_path"):
        if not isinstance(profile[field], str) or not Path(profile[field]).is_absolute():
            raise RuntimeError(f"authenticated artifact path is invalid: {field}")
    digest = profile["rule_pack_sha256"]
    if not isinstance(digest, str) or len(digest) != 64 or set(digest) - set("0123456789abcdef"):
        raise RuntimeError("authenticated rule-pack digest is invalid")
    return profile, hashlib.sha256(raw).hexdigest()


def private_snapshot(path: Path, payload: bytes) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o400)
    with os.fdopen(descriptor, "wb") as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())


def load_rules(profile: dict) -> tuple[list[tuple[dict, object]], str]:
    payload = read_stable(Path(profile["rule_pack_path"]), profile["rule_pack_maximum_bytes"])
    bundle = read_stable(Path(profile["rule_pack_bundle_path"]), profile["bundle_maximum_bytes"])
    key = read_stable(RULE_VERIFY_KEY_PATH, profile["public_key_maximum_bytes"], privileged=True)
    payload_sha256 = hashlib.sha256(payload).hexdigest()
    if not hmac.compare_digest(payload_sha256, profile["rule_pack_sha256"]):
        raise RuntimeError("rule pack differs from authenticated detector profile")
    with tempfile.TemporaryDirectory(prefix="aidefend-jailbreak-rules-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_path = root / "rules.json"
        bundle_path = root / "rules.sigstore.json"
        key_path = root / "verify.pub"
        for path, value in ((payload_path, payload), (bundle_path, bundle), (key_path, key)):
            private_snapshot(path, value)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(key_path), "--bundle", str(bundle_path), str(payload_path)],
            check=True, capture_output=True, text=True,
            timeout=float(profile["cosign_deadline_seconds"]),
        )
        if payload_path.read_bytes() != payload or bundle_path.read_bytes() != bundle or key_path.read_bytes() != key:
            raise OSError("verified rule-pack snapshot changed")
    document = json.loads(
        payload.decode("utf-8", errors="strict"), object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError("non-finite rule value: " + value)),
    )
    if not isinstance(document, dict) or set(document) != {"schema_version", "rules"}:
        raise ValueError("rule-pack root schema differs")
    if document["schema_version"] != "aidefend.decoy-jailbreak-rules.v1":
        raise ValueError("unsupported rule pack")
    rules = document["rules"]
    if not isinstance(rules, list) or not 1 <= len(rules) <= profile["rule_count_maximum"]:
        raise ValueError("rule-pack population is empty or exceeds the authenticated bound")
    options = re2.Options()
    options.case_sensitive = False
    options.log_errors = False
    compiled, seen = [], set()
    for rule in rules:
        if not isinstance(rule, dict) or set(rule) != {"rule_id", "pattern", "response_template_id"}:
            raise ValueError("rule schema differs")
        for field in ("rule_id", "pattern", "response_template_id"):
            value = rule[field]
            ceiling = profile["pattern_maximum_chars"] if field == "pattern" else profile["field_maximum_chars"]
            if not isinstance(value, str) or not value.strip() or len(value) > ceiling:
                raise ValueError(f"rule field is empty or oversized: {field}")
        if rule["rule_id"] in seen:
            raise ValueError(f"duplicate rule_id: {rule['rule_id']}")
        seen.add(rule["rule_id"])
        try:
            expression = re2.compile(rule["pattern"], options=options)
        except re2.error as exc:
            raise ValueError(f"rule uses invalid or unsupported RE2 syntax: {rule['rule_id']}") from exc
        compiled.append((rule, expression))
    return compiled, payload_sha256


PROFILE, PROFILE_SHA256 = load_profile()
RULES, RULE_PACK_SHA256 = load_rules(PROFILE)


def evaluate(prompt: str) -> Match | None:
    if not isinstance(prompt, str):
        raise TypeError("prompt must be text")
    encoded = prompt.encode("utf-8")
    if len(prompt) > PROFILE["prompt_maximum_chars"] or len(encoded) > PROFILE["prompt_maximum_bytes"]:
        raise ValueError("prompt exceeds authenticated detector bounds")
    prompt_sha256 = hashlib.sha256(encoded).hexdigest()
    for rule, expression in RULES:
        if expression.search(prompt):
            return Match(
                rule["rule_id"], prompt_sha256, rule["response_template_id"],
                RULE_PACK_SHA256, PROFILE["profile_version"], PROFILE_SHA256,
            )
    return None

The loader snapshots the rule pack, Sigstore bundle, and verification key, verifies those exact private bytes with cosign verify-blob, rejects any post-verification change, and parses only the verified payload. Store the matched rule ID, rule-pack digest, prompt digest, request/event ID, response-template ID, and decoy ID in normal evidence. Store raw prompt bytes only in the separately authorized encrypted/retention-locked forensic object created by the evidence pipeline, and reference its object version.

Verify the result

Replay a signed positive/negative/Unicode/encoding corpus through the deployed image. Prove exact rule/template selection, bounded execution, no external model or tool call, no raw prompt in SIEM, and a complete interaction-event link. Independently verify the rule-pack and template digests and read back the emitted finding.

Action: Promote the rule pack and static response templates only with the signed replay receipt, then monitor live findings through the decoy evidence pipeline.

The deployment must hash-lock google-re2. Its linear-time engine rejects unsupported constructs at startup, so an authenticated but mistaken pattern cannot trigger catastrophic backtracking on attacker-controlled prompt 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-DV-001: Honeypot AI Services & Decoy Models/APIs | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service