AID-DV-004: AI Output Telemetry Traps & Leak Beacons
Telemetry Traps
Design model or agent APIs to return unique, benign marker strings only for tightly controlled trap prompts, trap-only synthetic records, or synthetic probe workflows. A marker hit outside the trusted runtime is treated as evidence of scraping, model theft, unauthorized model replay, or enterprise data leakage.
Leak Beacons
Maintain a private registry of exact marker strings, trap identifiers, activation events, and investigation owners. Continuously scan relevant external surfaces and internal telemetry for exact marker reuse, then correlate hits with the original trap activation and access logs.
Design model or agent APIs to return unique, benign marker strings only for tightly controlled trap prompts, trap-only synthetic records, or synthetic probe workflows. A marker hit outside the trusted runtime is treated as evidence of scraping, model theft, unauthorized model replay, or enterprise data leakage.
Leak Beacons
Maintain a private registry of exact marker strings, trap identifiers, activation events, and investigation owners. Continuously scan relevant external surfaces and internal telemetry for exact marker reuse, then correlate hits with the original trap activation and access logs.
Pillars: data, model, app
Lifecycle: operation, validation
Scope boundary
Owns controlled trap prompts, exact benign marker strings, leak beacons, activation telemetry, external marker scanning, and marker-hit investigation. A marker establishes correlation with the registered trap workflow; it does not by itself prove watermark provenance, dataset access, or unauthorized training.
AID-H-013.001: AID-DV-004 registers and investigates exact deception markers; AID-H-013.001 embeds and verifies watermarks or signed content provenance. A leak investigation may consume AID-H-013.001 detector results, but this technique does not own watermark embedding, C2PA provenance, or modality-specific watermark registries.
AID-DV-002: AID-DV-004 activates and tracks output markers; AID-DV-002 owns pre-seeded honey data, decoy artifacts, fake credentials, and access-triggered canary tokens. Use AID-DV-004 for marker reuse observed after a controlled activation, and AID-DV-002 for interaction with a decoy item that already exists in an asset or repository.
AID-DV-007: AID-DV-004 investigates externally observed exact marker reuse; AID-DV-007 measures registered behavioral canaries in a controlled training or fine-tuning context. AID-DV-004 consumes a DV-007 canary only when that canary is deliberately registered for external leak investigation; it does not own the training-run measurement.
Implementation guidance
- AID-DV-004-G001: Evaluate exact trap prompts against a signed private registry, emit a deterministic benign marker without invoking the model, and durably bind every activation to the request and deployed trap version.
- AID-DV-004-G002: Scan the signed external-surface and finalized local-archive populations for exact telemetry markers through a bounded, SSRF-safe retrieval path.
- AID-DV-004-G003: Correlate signed trap activations and verified external marker hits into an immutable investigation record, attaching watermark or provenance results only by reference to AID-H-013.001.
AID-DV-004-G001
Evaluate exact trap prompts against a signed private registry, emit a deterministic benign marker without invoking the model, and durably bind every activation to the request and deployed trap version.
Trust boundary
Keep raw trap prompts and all three HMAC keys in a separately administered secret service. Before request handling, a root-controlled bootstrap copies each key into the fixed /run/secrets/deception hierarchy as a single-link 0400 or 0600 regular file; every ancestor is root-owned and non-writable by group or world. Load the keys through no-follow directory descriptors, then drop bootstrap privileges before accepting traffic. The signed private activation registry contains only trap IDs, keyed prompt digests, approved tenant/route/application bindings, logical marker IDs, validity windows, and owners. The trap evaluator returns None for ordinary traffic so the existing inference path remains separate; it never proxies or calls a model. Treat the returned marker as a candidate only: commit and independently read back the exact activation envelope before releasing it.
Evaluate with constant-time keyed digests
# File: deception/telemetry_trap.py
from __future__ import annotations
import hashlib
import hmac
import json
import os
import stat
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
PROMPT_KEY_PATH = "/run/secrets/deception/trap-prompt-hmac.key"
MARKER_KEY_PATH = "/run/secrets/deception/trap-marker-hmac.key"
ACTIVATION_KEY_PATH = "/run/secrets/deception/dv004-activation-hmac.key"
SECRET_PATHS = frozenset({PROMPT_KEY_PATH, MARKER_KEY_PATH, ACTIVATION_KEY_PATH})
SECRET_DIRECTORY_COMPONENTS = ("run", "secrets", "deception")
SECRET_FILES = {
PROMPT_KEY_PATH: "trap-prompt-hmac.key",
MARKER_KEY_PATH: "trap-marker-hmac.key",
ACTIVATION_KEY_PATH: "dv004-activation-hmac.key",
}
PROMPT_DIGEST_DOMAIN = b"aidefend.dv004.prompt-digest.v1\x00"
MARKER_DERIVATION_DOMAIN = b"aidefend.dv004.marker-derivation.v1\x00"
REQUEST_DIGEST_DOMAIN = b"aidefend.dv004.request-digest.v1\x00"
ACTIVATION_ENVELOPE_DOMAIN = b"aidefend.dv004.activation-envelope.v1\x00"
HEX = set("0123456789abcdef")
def secret(path: str) -> bytes:
if path not in SECRET_PATHS:
raise RuntimeError("secret path is not a compiled fixed path")
required_flags = ("O_NOFOLLOW", "O_DIRECTORY", "O_NONBLOCK")
if any(not hasattr(os, name) for name in required_flags) or os.open not in os.supports_dir_fd:
raise RuntimeError("this Linux deployment requires no-follow dirfd secret reads")
close_on_exec = getattr(os, "O_CLOEXEC", 0)
directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | close_on_exec
directory_fd = os.open("/", os.O_RDONLY | os.O_DIRECTORY | close_on_exec)
try:
for component in SECRET_DIRECTORY_COMPONENTS:
next_fd = os.open(component, directory_flags, dir_fd=directory_fd)
try:
directory_state = os.fstat(next_fd)
if (not stat.S_ISDIR(directory_state.st_mode)
or directory_state.st_uid != 0
or stat.S_IMODE(directory_state.st_mode) & 0o022):
raise PermissionError("secret ancestor is not root-controlled")
except Exception:
os.close(next_fd)
raise
os.close(directory_fd)
directory_fd = next_fd
file_flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | close_on_exec
descriptor = os.open(SECRET_FILES[path], file_flags, dir_fd=directory_fd)
try:
before = os.fstat(descriptor)
mode = stat.S_IMODE(before.st_mode)
if (not stat.S_ISREG(before.st_mode)
or before.st_uid != 0
or mode not in (0o400, 0o600)
or before.st_nlink != 1
or not 32 <= before.st_size <= 64):
raise PermissionError(
"trap secret must be a single-link root-owned 0400/0600 32..64-byte regular file"
)
chunks: list[bytes] = []
observed = 0
while observed <= 64:
chunk = os.read(descriptor, 65 - observed)
if not chunk:
break
chunks.append(chunk)
observed += len(chunk)
if observed > 64:
raise RuntimeError("trap secret exceeds the fixed byte bound")
after = os.fstat(descriptor)
identity = lambda item: (
item.st_dev, item.st_ino, item.st_mode, item.st_uid, item.st_gid,
item.st_nlink, item.st_size, item.st_mtime_ns, item.st_ctime_ns,
)
value = b"".join(chunks)
if len(value) != before.st_size or identity(before) != identity(after):
raise RuntimeError("trap secret changed while being read")
return value
finally:
os.close(descriptor)
finally:
os.close(directory_fd)
PROMPT_KEY = secret(PROMPT_KEY_PATH)
MARKER_KEY = secret(MARKER_KEY_PATH)
ACTIVATION_KEY = secret(ACTIVATION_KEY_PATH)
KEY_DIGESTS = tuple(
hashlib.sha256(key).digest()
for key in (PROMPT_KEY, MARKER_KEY, ACTIVATION_KEY)
)
if any(
hmac.compare_digest(KEY_DIGESTS[left], KEY_DIGESTS[right])
for left, right in ((0, 1), (0, 2), (1, 2))
):
raise RuntimeError("prompt, marker, and activation keys must be pairwise distinct")
@dataclass(frozen=True)
class TrapEntry:
tenant_id: str
application_id: str
marker_id: str
trap_id: str
prompt_hmac_sha256: str
route: str
registry_digest: str
valid_from_epoch: int
valid_until_epoch: int
def canonical(value: dict) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False,
).encode("utf-8")
def keyed_hex(key: bytes, value: bytes) -> str:
return hmac.new(key, value, hashlib.sha256).hexdigest()
def derive_marker(entry: TrapEntry) -> str:
deployment_identity = {
"tenant_id": entry.tenant_id,
"application_id": entry.application_id,
"marker_id": entry.marker_id,
"trap_id": entry.trap_id,
"route": entry.route,
"registry_digest": entry.registry_digest,
}
if (
any(not isinstance(value, str) or not value for value in deployment_identity.values())
or len(entry.registry_digest) != 64
or set(entry.registry_digest) - HEX
):
raise ValueError("registered deployment identity is invalid")
return "aidefend-trap-output-" + keyed_hex(
MARKER_KEY, MARKER_DERIVATION_DOMAIN + canonical(deployment_identity)
)[:32]
def evaluate_trap(*, prompt: str, request_id: str, tenant_id: str,
application_id: str, route: str,
entry: TrapEntry) -> tuple[str, bytes] | None:
if not isinstance(prompt, str):
raise ValueError("prompt must be a string")
request_identity = (request_id, tenant_id, application_id, route)
if any(not isinstance(value, str) or not value for value in request_identity):
raise ValueError("request identity fields must be non-empty strings")
if (
isinstance(entry.valid_from_epoch, bool)
or isinstance(entry.valid_until_epoch, bool)
or not isinstance(entry.valid_from_epoch, int)
or not isinstance(entry.valid_until_epoch, int)
or entry.valid_from_epoch >= entry.valid_until_epoch
):
raise ValueError("registered validity window is invalid")
if (
not isinstance(entry.prompt_hmac_sha256, str)
or len(entry.prompt_hmac_sha256) != 64
or set(entry.prompt_hmac_sha256) - HEX
):
raise ValueError("registered prompt digest is invalid")
marker = derive_marker(entry)
now_epoch = int(time.time())
if not entry.valid_from_epoch <= now_epoch < entry.valid_until_epoch:
return None
if tenant_id != entry.tenant_id or application_id != entry.application_id or route != entry.route:
return None
prompt_digest = keyed_hex(
PROMPT_KEY, PROMPT_DIGEST_DOMAIN + prompt.encode("utf-8")
)
if not hmac.compare_digest(prompt_digest, entry.prompt_hmac_sha256):
return None
payload = {
"schema": "aidefend.dv004.activation.v2",
"event_id": str(uuid.uuid4()),
"tenant_id": tenant_id,
"application_id": application_id,
"marker_id": entry.marker_id,
"trap_id": entry.trap_id,
"registry_digest": entry.registry_digest,
"request_digest": keyed_hex(
ACTIVATION_KEY,
REQUEST_DIGEST_DOMAIN + canonical({
"tenant_id": tenant_id,
"application_id": application_id,
"route": route,
"request_id": request_id,
}),
),
"marker_sha256": hashlib.sha256(marker.encode("ascii")).hexdigest(),
"activated_at": datetime.now(timezone.utc).isoformat(),
"model_invocations": 0,
}
envelope = {
"payload": payload,
"hmac_sha256": keyed_hex(
ACTIVATION_KEY, ACTIVATION_ENVELOPE_DOMAIN + canonical(payload)
),
}
return marker, canonical(envelope)Promotion and release sequence
Before promotion, verify the activation registry signature and exact schema, construct each TrapEntry, and call the same derive_marker function to precompute the actual marker. Insert that exact marker, scope, logical marker ID, trap ID, and activation-registry digest into the Cosign-signed aidefend.dv004-marker-registry.v3 surface registry consumed by scanning and correlation. The registry builder must compute each prompt_hmac_sha256 from PROMPT_DIGEST_DOMAIN plus the exact UTF-8 trap-prompt bytes and each marker through the same MARKER_DERIVATION_DOMAIN-prefixed derive_marker function used by the promotion probe. It must reject duplicate markers, logical IDs, or deployment identities. Run registry verification and TrapEntry construction only in the security-owned startup path, freeze the resulting mapping before request handling, and never construct or replace an entry from request data.
For a trap hit, a dedicated outbox-writer identity writes the returned canonical {"payload":...,"hmac_sha256":"..."} bytes with create-once semantics and has no read or overwrite permission. A distinct independent read-only identity has no write permission and must read back the exact object before marker release. Require the collector receipt and readback to bind the envelope digest, object version/checksum, writer and reader identity IDs, and their client-certificate or public-key digests; all must match the authenticated channel and exact bytes before releasing the candidate marker. If persistence, receipt validation, or readback fails, return an error and release neither the marker nor a normal model response. Never log raw prompt, marker, or HMAC keys to the general SIEM; ordinary inference remains in the production gateway and receives no trap key or activation registry.
Verify the result
Replay the complete signed trap population plus paraphrase, whitespace, Unicode, route, application, expiry, and ordinary-traffic negative controls. Verify that every registered deployment identity reproduces its pre-registered v3 surface marker across process restarts and distinct request IDs, while each activation retains a unique UUID event ID and a request-specific digest. Independently verify the domain-separated activation HMAC with the fixed activation key, recompute the context-bound request digest from each authorized replay record, require exact canonical-byte readback before release, confirm model_invocations=0 for every hit, and confirm negatives never produce an activation or marker. Retain activation and surface registries, signatures and digests, HMAC key IDs (not secret keys), writer/reader identity IDs and client-certificate or public-key digests, image digest, complete replay population, activation object versions/checksums, and promotion decision.
Action: Promote only precomputed markers present in the signed v3 surface registry, deploy the evaluator as a pre-inference exact-match branch, and release a marker only after the exact HMAC envelope is durably committed and independently read back.
AID-DV-004-G002
Scan the signed external-surface and finalized local-archive populations for exact telemetry markers through a bounded, SSRF-safe retrieval path.
Runtime policy: Load HTTP connect/read deadlines, maximum artifact bytes, and redirect/surface bounds from the signed, versioned scan profile bound to this marker registry and surface inventory, bind its version/digest to observations and replay, and return ERROR if the profile or any value is absent, invalid, or unverifiable.
Concept:
Telemetry markers are exact strings. Once a trap has emitted one, the detection workflow is disciplined inventory and exact-match scanning across the places where copied outputs, leaked prompts, or scraped artifacts actually appear. Do not hardcode two sample URLs and call that coverage. Drive scanning from an access-controlled marker registry and a versioned surface inventory.
Step 1: Maintain marker and surface inventories
File: deception/marker_registry.json
{
"schema_version": "aidefend.dv004-marker-registry.v3",
"scope": {"tenant_id": "tenant-example", "application_id": "support-assistant-prod"},
"markers": {
"aidefend-trap-output-6f83d5a1c90c4a2e9b011c13d72a4f80": {
"active": true,
"activation_registry_digest": "3333333333333333333333333333333333333333333333333333333333333333",
"marker_id": "telemetry_marker_v1",
"trap_id": "telemetry_trap_v1",
"owner": "security-deception-team"
},
"aidefend-trap-output-9b2d3172e6c1445786f5938da4c70e19": {
"active": true,
"activation_registry_digest": "4444444444444444444444444444444444444444444444444444444444444444",
"marker_id": "telemetry_marker_v2",
"trap_id": "telemetry_trap_v2",
"owner": "security-deception-team"
}
},
"scanner_limits": {
"request_timeout_seconds": 10,
"maximum_markers": 256,
"maximum_huggingface_file_bytes": 8388608,
"maximum_http_response_bytes": 8388608,
"http_response_chunk_bytes": 65536,
"maximum_github_pages": 10,
"maximum_local_file_bytes": 8388608,
"maximum_remote_entries": 4096,
"maximum_remote_aggregate_bytes": 268435456,
"maximum_watchlist_entries": 512,
"maximum_local_entries": 4096,
"maximum_local_aggregate_bytes": 536870912,
"maximum_redirects": 3
},
"egress_destinations": {
"github-api": {"hostname": "api.github.com", "port": 443, "path_prefixes": ["/search/code"]},
"huggingface-content": {"hostname": "huggingface.co", "port": 443, "path_prefixes": ["/external-org/", "/api/resolve-cache/"]},
"paste-watch": {"hostname": "paste.example.net", "port": 443, "path_prefixes": ["/monitored/"]},
"external-web": {"hostname": "security.example.org", "port": 443, "path_prefixes": ["/monitored/"]}
},
"surfaces": {
"github_code_search": {"enabled": true, "destination_id": "github-api", "expected_owner_allowlist": ["my-org", "trusted-partner-org"]},
"huggingface_hub": {"enabled": true, "artifacts": [
{"repo_id": "external-org/suspect-model", "repo_type": "model", "revision": "0123456789abcdef0123456789abcdef01234567", "path": "README.md", "destination_id": "huggingface-content", "url": "https://huggingface.co/external-org/suspect-model/resolve/0123456789abcdef0123456789abcdef01234567/README.md", "size": 8192, "sha256": "1111111111111111111111111111111111111111111111111111111111111111"}
]},
"paste_watchlist": {"enabled": true, "urls": [{"destination_id": "paste-watch", "url": "https://paste.example.net/monitored/dv004-feed"}]},
"html_watchlist": {"enabled": true, "urls": [{"destination_id": "external-web", "url": "https://security.example.org/monitored/external-artifacts"}]},
"local_archives": {"enabled": true, "path": "/srv/aidefend/dv004/archived-pages", "expected_population_sha256": "25bcd625fa108036df86cd33e8aa0bc4c1dca23ace7602e9da2c68a70cfc19c9", "expected_file_count": 1, "expected_aggregate_bytes": 8192}
},
"required_population": {
"algorithm": "aidefend.dv004.logical-work-items.v1",
"source_sha256": "ac2e2fb382ce6e1af432f5cbd728ea9d2ccf9f7ded0a83b84d0282cfb031a2ba",
"count": 6,
"sha256": "26281172db057223b5c3dbc4b6a676cab2430a03934db67a22c7393b12c5aa43"
}
}
The registry above is illustrative. Its six-item commitment is the SHA-256 of the canonical, lexicographically sorted work-item array produced by the code below: two active-marker GitHub queries, one immutable Hugging Face artifact, two watchlist URLs, and one finalized local archive population. The local example digest commits one 8,192-byte file named external-artifacts-2026-07-20.html whose SHA-256 is sixty-four 2 characters. Promotion must replace every scope, repository, URL, file measurement, and digest, then recompute both population commitments before signing. Sign the exact canonical file with Cosign v3.1.2, retain the generated bundle, and deploy only the registry, bundle, and public key. Pin the scanner image and Cosign v3.1.2 artifact by organizationally approved digest; never copy the private signing key into the scanner.
#!/usr/bin/env bash
set -euo pipefail
umask 077
: "${DV004_REGISTRY_SIGNING_KEY:?required}"
cosign sign-blob --yes --key "$DV004_REGISTRY_SIGNING_KEY" \
--bundle deception/marker_registry.json.sigstore.json \
deception/marker_registry.json
At deployment, copy projected policy material into fixed root-owned, non-symlink files before dropping privileges; the scanner intentionally rejects writable ancestors and symlink projections. The runtime public key at /run/trust/dv004-marker-registry.pub is an independently provisioned trust root, not a path selected by registry data.
Step 2: Scan public and controlled surfaces
# File: deception/telemetry_marker_scanner.py
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import hmac
import http.client
import ipaddress
import json
import math
import os
import re
import socket
import ssl
import stat
import subprocess
import tempfile
import time
from pathlib import Path
from urllib.parse import urlencode, urljoin, urlsplit
import yara
REGISTRY_PATH = Path("/run/policy/dv004-marker-registry.json")
REGISTRY_SIGNATURE = Path("/run/policy/dv004-marker-registry.json.sigstore.json")
REGISTRY_VERIFY_KEY = Path("/run/trust/dv004-marker-registry.pub")
COSIGN_BINARY = Path("/usr/local/bin/cosign")
PUBLIC_CA_BUNDLE = Path("/run/trust/public-ca-bundle.pem")
GITHUB_TOKEN_PATH = Path("/run/secrets/deception/github-read-token")
SCANNER_HMAC_KEY_PATH = Path("/run/secrets/deception/dv004-scanner-hmac.key")
# Compiled bootstrap ceiling used before any registry-carried limits are trusted.
REGISTRY_VERIFY_DEADLINE_CEILING_SECONDS = 10.0
REGISTRY_BOOTSTRAP_ARTIFACT_CEILING_BYTES = 2_000_000
TRUST_KEY_BOOTSTRAP_CEILING_BYTES = 65_536
DNS_ANSWER_BOOTSTRAP_CEILING = 64
# Implementation-capacity ceilings; signed policy may choose any smaller value.
REQUEST_DEADLINE_IMPLEMENTATION_CEILING_SECONDS = 60.0
OVERALL_SCAN_DEADLINE_IMPLEMENTATION_CEILING_SECONDS = 3_600.0
MARKER_COUNT_CEILING = 4_096
FILE_BYTES_CEILING = 67_108_864
HTTP_RESPONSE_BYTES_CEILING = 67_108_864
HTTP_CHUNK_BYTES_CEILING = 1_048_576
REMOTE_ENTRY_CEILING = 25_000
REMOTE_AGGREGATE_BYTES_CEILING = 536_870_912
WATCHLIST_ENTRY_CEILING = 4_096
LOCAL_ENTRY_CEILING = 10_000
LOCAL_AGGREGATE_BYTES_CEILING = 536_870_912
@dataclass(frozen=True)
class MarkerHit:
marker: str
trap_id: str
surface: str
location: str
artifact_digest: str
@dataclass(frozen=True)
class LocalSnapshot:
location: str
payload: bytes
sha256: str
def require_trusted_ancestors(path: Path) -> None:
absolute = Path(os.path.abspath(path))
current = Path(absolute.anchor)
for component in absolute.parent.parts[1:]:
current = current / component
metadata = current.lstat()
if (stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode)
or metadata.st_uid != 0 or metadata.st_mode & 0o022):
raise RuntimeError(f"untrusted policy ancestor: {current}")
def stable_bytes(path: Path, maximum: int, *, require_root_owned: bool = False,
require_private: bool = False) -> bytes:
if require_root_owned or require_private:
require_trusted_ancestors(path)
flags = (os.O_RDONLY | os.O_CLOEXEC
| getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_NONBLOCK", 0))
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode) or not 1 <= before.st_size <= maximum:
raise ValueError(f"artifact is not a bounded regular file: {path}")
if require_private:
if (before.st_uid != 0 or before.st_nlink != 1
or stat.S_IMODE(before.st_mode) not in {0o400, 0o600}):
raise RuntimeError(
f"secret must be a single-link root-owned 0400/0600 file: {path}"
)
elif require_root_owned and (before.st_uid != 0 or before.st_mode & 0o022):
raise RuntimeError(f"policy artifact is not root-owned read-only: {path}")
chunks, observed = [], 0
while observed <= maximum:
chunk = os.read(descriptor, min(65_536, maximum + 1 - observed))
if not chunk:
break
chunks.append(chunk)
observed += len(chunk)
raw = b"".join(chunks)
after = os.fstat(descriptor)
if len(raw) != before.st_size or len(raw) > maximum or (
before.st_dev, before.st_ino, before.st_mtime_ns, before.st_ctime_ns,
before.st_mode, before.st_uid, before.st_nlink,
) != (
after.st_dev, after.st_ino, after.st_mtime_ns, after.st_ctime_ns,
after.st_mode, after.st_uid, after.st_nlink,
):
raise RuntimeError(f"artifact changed while read: {path}")
return raw
finally:
os.close(descriptor)
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate JSON key: {key}")
value[key] = item
return value
def canonical(value) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False,
).encode("utf-8")
def digest(value) -> str:
return hashlib.sha256(canonical(value)).hexdigest()
def write_private_snapshot(path: Path, payload: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o400)
try:
offset = 0
while offset < len(payload):
written = os.write(descriptor, payload[offset:])
if written <= 0:
raise RuntimeError("short write while creating verification snapshot")
offset += written
os.fsync(descriptor)
finally:
os.close(descriptor)
def load_registry() -> dict:
cosign_metadata = COSIGN_BINARY.stat(follow_symlinks=False)
if (not stat.S_ISREG(cosign_metadata.st_mode) or cosign_metadata.st_uid != 0
or cosign_metadata.st_mode & 0o022 or not cosign_metadata.st_mode & 0o111):
raise RuntimeError("fixed Cosign verifier is not a root-owned read-only executable")
payload = stable_bytes(REGISTRY_PATH, REGISTRY_BOOTSTRAP_ARTIFACT_CEILING_BYTES, require_root_owned=True)
signature = stable_bytes(REGISTRY_SIGNATURE, REGISTRY_BOOTSTRAP_ARTIFACT_CEILING_BYTES, require_root_owned=True)
verification_key = stable_bytes(REGISTRY_VERIFY_KEY, TRUST_KEY_BOOTSTRAP_CEILING_BYTES, require_root_owned=True)
with tempfile.TemporaryDirectory(prefix="dv004-registry-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_path = root / "registry.json"
signature_path = root / "registry.sigstore.json"
key_path = root / "registry.pub"
write_private_snapshot(payload_path, payload)
write_private_snapshot(signature_path, signature)
write_private_snapshot(key_path, verification_key)
subprocess.run([
str(COSIGN_BINARY), "verify-blob", "--key", str(key_path),
"--bundle", str(signature_path), str(payload_path),
], check=True, capture_output=True, timeout=REGISTRY_VERIFY_DEADLINE_CEILING_SECONDS)
if (stable_bytes(payload_path, REGISTRY_BOOTSTRAP_ARTIFACT_CEILING_BYTES) != payload
or stable_bytes(signature_path, REGISTRY_BOOTSTRAP_ARTIFACT_CEILING_BYTES) != signature
or stable_bytes(key_path, TRUST_KEY_BOOTSTRAP_CEILING_BYTES) != verification_key):
raise RuntimeError("verified registry snapshot changed")
registry = json.loads(payload.decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError("non-finite JSON constant: " + value)))
expected = {"schema_version", "scope", "markers", "scanner_limits", "egress_destinations", "surfaces", "required_population"}
if (not isinstance(registry, dict) or set(registry) != expected
or registry["schema_version"] != "aidefend.dv004-marker-registry.v3"):
raise ValueError("marker registry schema differs")
registry["_verified_sha256"] = hashlib.sha256(payload).hexdigest()
registry["_verification_key_sha256"] = hashlib.sha256(verification_key).hexdigest()
return registry
REGISTRY = load_registry()
SCOPE = REGISTRY["scope"]
if (not isinstance(SCOPE, dict) or set(SCOPE) != {"tenant_id", "application_id"}
or any(not isinstance(SCOPE[field], str) or not SCOPE[field].strip()
or len(SCOPE[field]) > 128 for field in SCOPE)):
raise RuntimeError("signed scanner scope is invalid")
LIMITS = REGISTRY["scanner_limits"]
if not isinstance(LIMITS, dict) or set(LIMITS) != {
"request_timeout_seconds", "maximum_markers", "maximum_huggingface_file_bytes",
"maximum_http_response_bytes", "http_response_chunk_bytes",
"maximum_github_pages", "maximum_local_file_bytes",
"maximum_remote_entries", "maximum_remote_aggregate_bytes",
"maximum_watchlist_entries", "maximum_local_entries",
"maximum_local_aggregate_bytes", "maximum_redirects",
}:
raise RuntimeError("signed scanner limits differ")
REQUEST_TIMEOUT_SECONDS = float(LIMITS["request_timeout_seconds"])
MAXIMUM_MARKERS = int(LIMITS["maximum_markers"])
MAXIMUM_HUGGINGFACE_FILE_BYTES = int(LIMITS["maximum_huggingface_file_bytes"])
MAXIMUM_HTTP_RESPONSE_BYTES = int(LIMITS["maximum_http_response_bytes"])
HTTP_RESPONSE_CHUNK_BYTES = int(LIMITS["http_response_chunk_bytes"])
MAXIMUM_GITHUB_PAGES = int(LIMITS["maximum_github_pages"])
MAXIMUM_LOCAL_FILE_BYTES = int(LIMITS["maximum_local_file_bytes"])
MAXIMUM_REMOTE_ENTRIES = int(LIMITS["maximum_remote_entries"])
MAXIMUM_REMOTE_AGGREGATE_BYTES = int(LIMITS["maximum_remote_aggregate_bytes"])
MAXIMUM_WATCHLIST_ENTRIES = int(LIMITS["maximum_watchlist_entries"])
MAXIMUM_LOCAL_ENTRIES = int(LIMITS["maximum_local_entries"])
MAXIMUM_LOCAL_AGGREGATE_BYTES = int(LIMITS["maximum_local_aggregate_bytes"])
MAXIMUM_REDIRECTS = int(LIMITS["maximum_redirects"])
INTEGER_LIMIT_FIELDS = set(LIMITS) - {"request_timeout_seconds"}
if (type(LIMITS["request_timeout_seconds"]) not in (int, float)
or any(type(LIMITS[field]) is not int for field in INTEGER_LIMIT_FIELDS)
or isinstance(LIMITS["request_timeout_seconds"], bool)
or not math.isfinite(REQUEST_TIMEOUT_SECONDS)
or not 0 < REQUEST_TIMEOUT_SECONDS <= REQUEST_DEADLINE_IMPLEMENTATION_CEILING_SECONDS
or not 1 <= MAXIMUM_MARKERS <= MARKER_COUNT_CEILING
or not 1 <= MAXIMUM_HUGGINGFACE_FILE_BYTES <= FILE_BYTES_CEILING
or not 1 <= MAXIMUM_LOCAL_FILE_BYTES <= FILE_BYTES_CEILING
or not 1 <= MAXIMUM_HTTP_RESPONSE_BYTES <= HTTP_RESPONSE_BYTES_CEILING
or not 1 <= HTTP_RESPONSE_CHUNK_BYTES <= min(HTTP_CHUNK_BYTES_CEILING, MAXIMUM_HTTP_RESPONSE_BYTES)
or not 1 <= MAXIMUM_REMOTE_ENTRIES <= REMOTE_ENTRY_CEILING
or not 1 <= MAXIMUM_REMOTE_AGGREGATE_BYTES <= REMOTE_AGGREGATE_BYTES_CEILING
or not 1 <= MAXIMUM_WATCHLIST_ENTRIES <= WATCHLIST_ENTRY_CEILING
or not 1 <= MAXIMUM_LOCAL_ENTRIES <= LOCAL_ENTRY_CEILING
or not 1 <= MAXIMUM_LOCAL_AGGREGATE_BYTES <= LOCAL_AGGREGATE_BYTES_CEILING
or not 1 <= MAXIMUM_GITHUB_PAGES <= 10
or not 0 <= MAXIMUM_REDIRECTS <= 5):
raise RuntimeError("signed scanner limits are invalid")
SCAN_DEADLINE_MONOTONIC: float | None = None
def start_scan_deadline() -> None:
global SCAN_DEADLINE_MONOTONIC
if SCAN_DEADLINE_MONOTONIC is not None:
raise RuntimeError("scanner process permits one bounded scan")
SCAN_DEADLINE_MONOTONIC = (
time.monotonic() + OVERALL_SCAN_DEADLINE_IMPLEMENTATION_CEILING_SECONDS
)
def remaining_scan_seconds() -> float:
if SCAN_DEADLINE_MONOTONIC is None:
raise RuntimeError("overall scan deadline was not initialized")
remaining = SCAN_DEADLINE_MONOTONIC - time.monotonic()
if remaining <= 0:
raise RuntimeError("overall scan implementation deadline exhausted")
return remaining
def request_timeout_seconds() -> float:
return min(REQUEST_TIMEOUT_SECONDS, remaining_scan_seconds())
MARKERS = REGISTRY["markers"]
if (not isinstance(MARKERS, dict) or not 1 <= len(MARKERS) <= MAXIMUM_MARKERS):
raise RuntimeError("signed marker population is empty or exceeds its bound")
for marker, metadata in MARKERS.items():
if (not isinstance(marker, str)
or not re.fullmatch(r"aidefend-trap-output-[0-9a-f]{32}", marker)
or not isinstance(metadata, dict) or set(metadata) != {"active", "activation_registry_digest", "marker_id", "trap_id", "owner"}
or type(metadata["active"]) is not bool
or not isinstance(metadata["activation_registry_digest"], str)
or not re.fullmatch(r"[0-9a-f]{64}", metadata["activation_registry_digest"])
or any(not isinstance(metadata[field], str) or not metadata[field].strip()
for field in ("marker_id", "trap_id", "owner"))):
raise RuntimeError("signed marker entry is invalid")
if len({metadata["marker_id"] for metadata in MARKERS.values()}) != len(MARKERS):
raise RuntimeError("signed logical marker IDs are duplicated")
DESTINATIONS = REGISTRY["egress_destinations"]
if not isinstance(DESTINATIONS, dict) or not 1 <= len(DESTINATIONS) <= MAXIMUM_REMOTE_ENTRIES:
raise RuntimeError("signed egress destination population is empty")
for destination_id, destination in DESTINATIONS.items():
if (not isinstance(destination_id, str)
or not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", destination_id)
or not isinstance(destination, dict)
or set(destination) != {"hostname", "port", "path_prefixes"}):
raise RuntimeError("signed egress destination schema differs")
hostname = destination["hostname"]
prefixes = destination["path_prefixes"]
if (not isinstance(hostname, str) or hostname != hostname.lower().rstrip(".")
or hostname.encode("idna").decode("ascii") != hostname
or type(destination["port"]) is not int or destination["port"] != 443
or not isinstance(prefixes, list) or not prefixes or len(prefixes) != len(set(prefixes))
or any(not isinstance(prefix, str) or not prefix.startswith("/")
or "\\" in prefix or any(ord(char) < 0x20 for char in prefix)
for prefix in prefixes)):
raise RuntimeError("signed egress destination value is invalid")
try:
ipaddress.ip_address(hostname)
except ValueError:
pass
else:
raise RuntimeError("egress destinations must use verified DNS names, not IP literals")
SURFACES = REGISTRY["surfaces"]
EXPECTED_SURFACES = {"github_code_search", "huggingface_hub", "paste_watchlist", "html_watchlist", "local_archives"}
if not isinstance(SURFACES, dict) or set(SURFACES) != EXPECTED_SURFACES:
raise RuntimeError("signed surface inventory schema differs")
def population_source(registry: dict) -> dict:
return {
"schema_version": registry["schema_version"],
"scope": registry["scope"],
"markers": registry["markers"],
"egress_destinations": registry["egress_destinations"],
"surfaces": registry["surfaces"],
}
def derive_required_work_items(registry: dict) -> list[dict]:
surfaces = registry["surfaces"]
github = surfaces["github_code_search"]
huggingface = surfaces["huggingface_hub"]
html_watchlist = surfaces["html_watchlist"]
paste_watchlist = surfaces["paste_watchlist"]
local_archives = surfaces["local_archives"]
if (not isinstance(github, dict) or set(github) != {"enabled", "destination_id", "expected_owner_allowlist"}
or not isinstance(huggingface, dict) or set(huggingface) != {"enabled", "artifacts"}
or not isinstance(html_watchlist, dict) or set(html_watchlist) != {"enabled", "urls"}
or not isinstance(paste_watchlist, dict) or set(paste_watchlist) != {"enabled", "urls"}
or not isinstance(local_archives, dict) or set(local_archives) != {
"enabled", "path", "expected_population_sha256",
"expected_file_count", "expected_aggregate_bytes",
}
or any(type(surface["enabled"]) is not bool for surface in surfaces.values())):
raise RuntimeError("signed surface contract differs")
if not any(surface["enabled"] for surface in surfaces.values()):
raise RuntimeError("signed surface contract disables the complete scan population")
active_markers = sorted(marker for marker, meta in registry["markers"].items() if meta["active"])
if not active_markers:
raise RuntimeError("signed marker population has no active marker")
owners = github["expected_owner_allowlist"]
if (not isinstance(github["destination_id"], str)
or github["destination_id"] not in registry["egress_destinations"]
or not isinstance(owners, list) or owners != sorted(set(owners))
or any(not isinstance(owner, str) or not re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,38})", owner)
for owner in owners)
or github["enabled"] and not owners):
raise RuntimeError("signed GitHub surface is invalid or empty")
artifacts = huggingface["artifacts"]
if (not isinstance(artifacts, list) or len(artifacts) > MAXIMUM_REMOTE_ENTRIES
or huggingface["enabled"] and not artifacts):
raise RuntimeError("enabled Hugging Face surface is empty or exceeds its bound")
for watchlist in (html_watchlist, paste_watchlist):
if (not isinstance(watchlist["urls"], list)
or watchlist["enabled"] and not watchlist["urls"]):
raise RuntimeError("enabled URL watchlist is empty")
if len(html_watchlist["urls"]) + len(paste_watchlist["urls"]) > MAXIMUM_WATCHLIST_ENTRIES:
raise RuntimeError("signed watchlist population exceeds its bound")
items: list[dict] = []
if github["enabled"]:
for marker_id in active_markers:
items.append({
"kind": "github_code_search", "marker": marker_id,
"marker_id": registry["markers"][marker_id]["marker_id"],
"destination_id": github["destination_id"],
"expected_owner_allowlist": owners,
})
if huggingface["enabled"]:
items.extend({"kind": "huggingface_artifact", "artifact": artifact} for artifact in artifacts)
for kind, watchlist in (("html_watchlist", html_watchlist), ("paste_watchlist", paste_watchlist)):
if watchlist["enabled"]:
for entry in watchlist["urls"]:
if (not isinstance(entry, dict) or set(entry) != {"destination_id", "url"}
or not all(isinstance(entry[field], str) and entry[field] for field in entry)):
raise RuntimeError(f"enabled {kind} entry is invalid")
items.append({"kind": kind, "entry": entry})
if local_archives["enabled"]:
path = local_archives["path"]
expected_digest = local_archives["expected_population_sha256"]
expected_count = local_archives["expected_file_count"]
expected_bytes = local_archives["expected_aggregate_bytes"]
if (not isinstance(path, str) or not Path(path).is_absolute()
or not isinstance(expected_digest, str) or not re.fullmatch(r"[0-9a-f]{64}", expected_digest)
or type(expected_count) is not int or not 1 <= expected_count <= min(
MAXIMUM_LOCAL_ENTRIES, LOCAL_ENTRY_CEILING
)
or type(expected_bytes) is not int or not 1 <= expected_bytes <= min(
MAXIMUM_LOCAL_AGGREGATE_BYTES, LOCAL_AGGREGATE_BYTES_CEILING
)):
raise RuntimeError("enabled local archive lacks an exact non-empty population commitment")
items.append({
"kind": "local_archives", "path": path,
"expected_population_sha256": expected_digest,
"expected_file_count": expected_count,
"expected_aggregate_bytes": expected_bytes,
})
items.sort(key=canonical)
encoded = [canonical(item) for item in items]
if not items or len(encoded) != len(set(encoded)):
raise RuntimeError("required logical work-item population is empty or duplicated")
return items
WORK_ITEMS = derive_required_work_items(REGISTRY)
REQUIRED_POPULATION = REGISTRY["required_population"]
REQUIRED_SOURCE_SHA256 = digest(population_source(REGISTRY))
DERIVED_REQUIRED_SHA256 = digest(WORK_ITEMS)
if (not isinstance(REQUIRED_POPULATION, dict)
or set(REQUIRED_POPULATION) != {"algorithm", "source_sha256", "count", "sha256"}
or REQUIRED_POPULATION["algorithm"] != "aidefend.dv004.logical-work-items.v1"
or type(REQUIRED_POPULATION["count"]) is not int
or REQUIRED_POPULATION["count"] != len(WORK_ITEMS)
or not hmac.compare_digest(str(REQUIRED_POPULATION["source_sha256"]), REQUIRED_SOURCE_SHA256)
or not hmac.compare_digest(str(REQUIRED_POPULATION["sha256"]), DERIVED_REQUIRED_SHA256)):
raise RuntimeError("signed required-population commitment differs from independently derived work items")
SURFACE_INVENTORY_DIGEST = digest({
"scope": SCOPE, "egress_destinations": DESTINATIONS, "surfaces": SURFACES,
})
ACTIVE_MARKERS = {marker: meta for marker, meta in MARKERS.items() if meta["active"]}
@dataclass
class ScanBudget:
remote_entries: int = 0
remote_bytes: int = 0
def consume_remote(self, payload: bytes) -> None:
remaining_scan_seconds()
self.remote_entries += 1
self.remote_bytes += len(payload)
if (self.remote_entries > MAXIMUM_REMOTE_ENTRIES
or self.remote_bytes > MAXIMUM_REMOTE_AGGREGATE_BYTES):
raise RuntimeError("remote scan population exceeds signed bounds")
BUDGET = ScanBudget()
REMOTE_FETCH_RECEIPTS: list[dict] = []
SCANNER_HMAC_KEY = stable_bytes(
SCANNER_HMAC_KEY_PATH, 64, require_private=True
)
if not 32 <= len(SCANNER_HMAC_KEY) <= 64:
raise RuntimeError("scanner event HMAC key must contain 32..64 bytes")
CA_BUNDLE_BYTES = stable_bytes(PUBLIC_CA_BUNDLE, REGISTRY_BOOTSTRAP_ARTIFACT_CEILING_BYTES, require_root_owned=True)
TLS_CONTEXT = ssl.create_default_context(cafile=str(PUBLIC_CA_BUNDLE))
if stable_bytes(PUBLIC_CA_BUNDLE, REGISTRY_BOOTSTRAP_ARTIFACT_CEILING_BYTES, require_root_owned=True) != CA_BUNDLE_BYTES:
raise RuntimeError("public CA bundle changed while TLS context was created")
def with_event_id(payload: dict) -> dict:
if "event_id" in payload:
raise RuntimeError("event_id must be derived, not supplied")
return {**payload, "event_id": "dv004-" + hashlib.sha256(canonical(payload)).hexdigest()}
def signed_envelope(payload: dict) -> dict:
return {
"payload": payload,
"hmac_sha256": hmac.new(SCANNER_HMAC_KEY, canonical(payload), hashlib.sha256).hexdigest(),
}
def emit_external_marker_hit(hit: MarkerHit, observed_at: str) -> None:
payload = with_event_id({
"schema": "aidefend.dv004.external-hit.v1",
"tenant_id": SCOPE["tenant_id"],
"application_id": SCOPE["application_id"],
"marker_id": MARKERS[hit.marker]["marker_id"],
"marker_sha256": hashlib.sha256(hit.marker.encode("ascii")).hexdigest(),
"trap_id": hit.trap_id,
"registry_digest": MARKERS[hit.marker]["activation_registry_digest"],
"surface_registry_sha256": REGISTRY["_verified_sha256"],
"surface_inventory_digest": SURFACE_INVENTORY_DIGEST,
"surface": hit.surface,
"artifact_digest": hit.artifact_digest,
"location": hit.location,
"observed_at": observed_at,
})
print(canonical(signed_envelope(payload)).decode("utf-8"))
def scan_payload(markers: dict, surface: str, location: str, payload: bytes) -> list[MarkerHit]:
remaining_scan_seconds()
artifact_digest = hashlib.sha256(payload).hexdigest()
hits: list[MarkerHit] = []
for marker, meta in markers.items():
remaining_scan_seconds()
if marker.encode("ascii") in payload or marker.encode("utf-16le") in payload:
hits.append(MarkerHit(marker, meta["trap_id"], surface, location, artifact_digest))
return hits
def target_destination(url: str, required_destination_id: str | None = None):
if (not isinstance(url, str) or not url.isascii() or not 1 <= len(url) <= 4096
or "\\" in url or any(ord(char) <= 0x20 or ord(char) == 0x7f for char in url)):
raise ValueError("target URL is not bounded canonical ASCII")
parsed = urlsplit(url)
if (parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password
or parsed.fragment or parsed.port not in (None, 443)):
raise ValueError("target URL is not canonical HTTPS on port 443")
if "%" in parsed.path or any(segment in {".", ".."} for segment in parsed.path.split("/")):
raise ValueError("target path contains ambiguous encoding or dot segments")
hostname = parsed.hostname.encode("idna").decode("ascii").lower().rstrip(".")
if hostname != parsed.hostname or parsed.netloc not in {hostname, f"{hostname}:443"}:
raise ValueError("target hostname is not canonical")
candidate_ids = [required_destination_id] if required_destination_id is not None else list(DESTINATIONS)
matched = []
for destination_id in candidate_ids:
destination = DESTINATIONS.get(destination_id)
if (destination and destination["hostname"] == hostname
and destination["port"] == 443
and any(
(parsed.path or "/") == prefix.rstrip("/")
or (parsed.path or "/").startswith(prefix.rstrip("/") + "/")
for prefix in destination["path_prefixes"]
)):
matched.append(destination_id)
if len(matched) != 1:
raise RuntimeError("target does not match exactly one signed egress destination")
return matched[0], parsed
def public_endpoints(hostname: str, port: int):
remaining_scan_seconds()
answers = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP)
remaining_scan_seconds()
if not 1 <= len(answers) <= DNS_ANSWER_BOOTSTRAP_CEILING:
raise RuntimeError("destination DNS answer population is empty or excessive")
endpoints = {}
for family, socket_type, protocol, _canonical_name, socket_address in answers:
address = ipaddress.ip_address(socket_address[0].split("%", 1)[0])
effective = address.ipv4_mapped or address
if (not effective.is_global or effective.is_loopback or effective.is_private
or effective.is_link_local or effective.is_multicast
or effective.is_reserved or effective.is_unspecified):
raise RuntimeError(f"destination DNS included a non-public address: {effective}")
key = (family, socket_type, protocol, socket_address)
endpoints[key] = (family, socket_type, protocol, socket_address, str(effective))
return [endpoints[key] for key in sorted(endpoints, key=lambda item: repr(item))]
def connect_pinned_tls(hostname: str, port: int):
errors = []
for family, socket_type, protocol, socket_address, public_ip in public_endpoints(hostname, port):
remaining_scan_seconds()
raw_socket = socket.socket(family, socket_type, protocol)
raw_socket.settimeout(request_timeout_seconds())
try:
raw_socket.connect(socket_address)
raw_socket.settimeout(request_timeout_seconds())
tls_socket = TLS_CONTEXT.wrap_socket(raw_socket, server_hostname=hostname)
tls_socket.settimeout(request_timeout_seconds())
remaining_scan_seconds()
return tls_socket, public_ip
except (OSError, ssl.SSLError) as exc:
errors.append(f"{public_ip}: {type(exc).__name__}")
raw_socket.close()
raise RuntimeError("no signed public destination address connected: " + ", ".join(errors))
def one_pinned_https_get(url: str, destination_id: str, headers: dict[str, str]):
remaining_scan_seconds()
matched_id, parsed = target_destination(url, destination_id)
allowed_headers = {"accept", "authorization", "user-agent", "x-github-api-version"}
normalized_headers = {}
for name, value in headers.items():
lower_name = name.lower()
if (lower_name not in allowed_headers or not isinstance(value, str)
or not value or any(char in value for char in "\r\n")):
raise ValueError(f"unsafe outbound header: {name}")
normalized_headers[lower_name] = value
normalized_headers.setdefault("accept", "text/html, application/json;q=0.9, text/plain;q=0.8")
normalized_headers.setdefault("user-agent", "AIDEFEND-TelemetryScanner/2.0")
normalized_headers["accept-encoding"] = "identity"
tls_socket, public_ip = connect_pinned_tls(parsed.hostname, 443)
response = None
try:
request_target = parsed.path or "/"
if parsed.query:
request_target += "?" + parsed.query
request_lines = [f"GET {request_target} HTTP/1.1", f"Host: {parsed.hostname}", "Connection: close"]
request_lines.extend(f"{name}: {value}" for name, value in sorted(normalized_headers.items()))
tls_socket.settimeout(request_timeout_seconds())
tls_socket.sendall(("\r\n".join(request_lines) + "\r\n\r\n").encode("ascii"))
response = http.client.HTTPResponse(tls_socket, method="GET")
tls_socket.settimeout(request_timeout_seconds())
response.begin()
remaining_scan_seconds()
pairs = response.getheaders()
for singleton in ("content-length", "content-encoding", "location"):
if sum(1 for name, _value in pairs if name.lower() == singleton) > 1:
raise RuntimeError(f"duplicate response header: {singleton}")
content_encoding = response.getheader("Content-Encoding")
if content_encoding not in (None, "identity"):
raise RuntimeError("compressed external responses are forbidden")
content_length = response.getheader("Content-Length")
if content_length is not None:
try:
claimed = int(content_length, 10)
except ValueError as exc:
raise RuntimeError("invalid external content-length") from exc
if claimed < 0 or claimed > MAXIMUM_HTTP_RESPONSE_BYTES:
raise RuntimeError("external response exceeds signed per-response bound")
chunks: list[bytes] = []
observed = 0
while observed <= MAXIMUM_HTTP_RESPONSE_BYTES:
tls_socket.settimeout(request_timeout_seconds())
chunk = response.read(min(
HTTP_RESPONSE_CHUNK_BYTES,
MAXIMUM_HTTP_RESPONSE_BYTES + 1 - observed,
))
remaining_scan_seconds()
if not chunk:
break
chunks.append(chunk)
observed += len(chunk)
if observed > MAXIMUM_HTTP_RESPONSE_BYTES:
raise RuntimeError("external response exceeds signed per-response bound")
payload = b"".join(chunks)
return response.status, response.getheader("Location"), payload, matched_id, public_ip
finally:
if response is not None:
response.close()
tls_socket.close()
def pinned_https_get(url: str, destination_id: str, headers: dict[str, str] | None = None):
current_url = url
current_destination = destination_id
request_headers = dict(headers or {})
trace = []
for redirect_count in range(MAXIMUM_REDIRECTS + 1):
remaining_scan_seconds()
status, location, payload, matched_id, public_ip = one_pinned_https_get(
current_url, current_destination, request_headers
)
trace.append({
"destination_id": matched_id, "resolved_public_ip": public_ip, "status": status,
"url_sha256": hashlib.sha256(current_url.encode()).hexdigest(),
"body_sha256": hashlib.sha256(payload).hexdigest(),
})
BUDGET.consume_remote(payload)
if status in {301, 302, 303, 307, 308}:
if redirect_count == MAXIMUM_REDIRECTS or not location:
raise RuntimeError("external redirect exceeds signed bound or lacks Location")
if any(name.lower() == "authorization" for name in request_headers):
raise RuntimeError("authenticated external requests may not redirect")
current_url = urljoin(current_url, location)
current_destination, _parsed = target_destination(current_url)
continue
if status != 200:
raise RuntimeError(f"external destination returned HTTP {status}")
REMOTE_FETCH_RECEIPTS.extend(trace)
return payload, trace
raise RuntimeError("unreachable redirect state")
def scan_html_url(markers: dict, entry: dict, surface: str) -> list[MarkerHit]:
if not isinstance(entry, dict) or set(entry) != {"destination_id", "url"}:
raise ValueError("watchlist entry schema differs")
body, _trace = pinned_https_get(entry["url"], entry["destination_id"])
# Raw bounded bytes are authoritative: visible-text extraction alone misses
# markers in attributes, comments, script/style/code blocks, or raw source.
return scan_payload(markers, surface, entry["url"], body)
def scan_github_code(markers: dict, surface: dict) -> list[MarkerHit]:
if (not isinstance(surface, dict)
or set(surface) != {"enabled", "destination_id", "expected_owner_allowlist"}
or surface["enabled"] is not True
or not isinstance(surface["destination_id"], str)):
raise RuntimeError("GitHub surface schema differs")
expected_owners = surface["expected_owner_allowlist"]
if (not isinstance(expected_owners, list) or len(expected_owners) != len(set(expected_owners))
or any(not isinstance(owner, str) or not re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,38})", owner)
for owner in expected_owners)):
raise RuntimeError("GitHub expected-owner allowlist is invalid")
token = stable_bytes(GITHUB_TOKEN_PATH, 4096, require_private=True).decode(
"ascii", errors="strict"
).strip()
if not token or any(ord(char) <= 0x20 or ord(char) == 0x7f for char in token):
raise RuntimeError("GitHub read token is empty or contains control characters")
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
hits: list[MarkerHit] = []
for marker, meta in markers.items():
query = f"{marker} in:file"
covered, expected_total, result_identities = 0, None, set()
for page in range(1, MAXIMUM_GITHUB_PAGES + 1):
remaining_scan_seconds()
url = "https://api.github.com/search/code?" + urlencode({
"q": query, "per_page": 100, "page": page,
})
payload, _trace = pinned_https_get(url, surface["destination_id"], headers)
document = json.loads(payload.decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)))
if not isinstance(document, dict) or set(document) != {"total_count", "incomplete_results", "items"}:
raise RuntimeError("GitHub search schema differs")
if document["incomplete_results"] is not False or not isinstance(document["items"], list):
raise RuntimeError("GitHub search population is incomplete")
total = document["total_count"]
if isinstance(total, bool) or not isinstance(total, int) or total < 0 or total > 1000:
raise RuntimeError("GitHub search exceeds complete-query limits")
if expected_total is None:
expected_total = total
elif total != expected_total:
raise RuntimeError("GitHub search population changed during pagination")
for item in document["items"]:
if not isinstance(item, dict) or not isinstance(item.get("repository"), dict):
raise RuntimeError("GitHub result schema differs")
repository = item["repository"].get("full_name")
path, blob_sha = item.get("path"), item.get("sha")
if (not isinstance(repository, str) or not isinstance(path, str)
or not isinstance(blob_sha, str) or not re.fullmatch(r"[0-9a-f]{40}", blob_sha)):
raise RuntimeError("GitHub result identity differs")
identity = (repository, path, blob_sha)
if identity in result_identities:
raise RuntimeError("GitHub pagination returned a duplicate result identity")
result_identities.add(identity)
if repository.split("/", 1)[0] not in set(expected_owners):
artifact_digest = digest({
"repository": repository, "path": path, "git_blob_sha1": blob_sha,
})
hits.append(MarkerHit(
marker, meta["trap_id"], "github_code_search",
f"{repository}:{path}", artifact_digest,
))
covered = len(result_identities)
if covered >= expected_total:
break
if expected_total is None or covered < expected_total:
raise RuntimeError("signed GitHub page budget ended before complete coverage")
return hits
def scan_huggingface(markers: dict, artifacts: list[dict]) -> list[MarkerHit]:
if not isinstance(artifacts, list) or not artifacts or len(artifacts) > MAXIMUM_REMOTE_ENTRIES:
raise RuntimeError("Hugging Face artifact population is empty or exceeds its signed bound")
hits: list[MarkerHit] = []
identities = set()
for artifact in artifacts:
fields = {"repo_id", "repo_type", "revision", "path", "destination_id", "url", "size", "sha256"}
if not isinstance(artifact, dict) or set(artifact) != fields:
raise ValueError("Hugging Face artifact entry schema differs")
repo_id = artifact["repo_id"]
repo_type = artifact["repo_type"]
revision = artifact["revision"]
artifact_path = artifact["path"]
if repo_type not in {"model", "dataset"}:
raise ValueError(f"unsupported Hugging Face repo_type: {repo_type}")
if (not isinstance(repo_id, str) or not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repo_id)
or not isinstance(revision, str) or not re.fullmatch(r"[0-9a-f]{40}", revision)
or not isinstance(artifact_path, str) or not re.fullmatch(r"[A-Za-z0-9_.\/-]+", artifact_path)
or any(part in {"", ".", ".."} for part in artifact_path.split("/"))
or type(artifact["size"]) is not int or not 1 <= artifact["size"] <= MAXIMUM_HUGGINGFACE_FILE_BYTES
or not isinstance(artifact["sha256"], str)
or not re.fullmatch(r"[0-9a-f]{64}", artifact["sha256"])):
raise ValueError("Hugging Face artifact identity or bound is invalid")
identity = (repo_type, repo_id, revision, artifact_path)
if identity in identities:
raise ValueError("duplicate Hugging Face artifact identity")
identities.add(identity)
prefix = "" if repo_type == "model" else "/datasets"
expected_path = f"{prefix}/{repo_id}/resolve/{revision}/{artifact_path}"
if urlsplit(artifact["url"]).path != expected_path:
raise ValueError("Hugging Face artifact URL is not bound to its exact identity")
payload, _trace = pinned_https_get(artifact["url"], artifact["destination_id"])
if (len(payload) != artifact["size"]
or hashlib.sha256(payload).hexdigest() != artifact["sha256"]):
raise RuntimeError("Hugging Face artifact differs from the signed inventory")
hits.extend(scan_payload(
markers, f"huggingface_{repo_type}",
f"{repo_id}@{revision}:{artifact_path}", payload
))
return hits
def build_yara_rules(markers: dict):
remaining_scan_seconds()
strings = "\n".join(
f" $m{idx} = {json.dumps(marker)} ascii wide"
for idx, marker in enumerate(markers, start=1)
)
return yara.compile(source=f"rule aidefend_marker_hit {{ strings:\n{strings}\n condition: any of them }}")
def read_open_snapshot(descriptor: int, location: str) -> LocalSnapshot:
remaining_scan_seconds()
before = os.fstat(descriptor)
if (not stat.S_ISREG(before.st_mode) or before.st_uid != 0 or before.st_mode & 0o022
or before.st_nlink != 1 or before.st_size < 0
or before.st_size > MAXIMUM_LOCAL_FILE_BYTES):
raise RuntimeError(f"local archive file is not a bounded finalized regular file: {location}")
chunks, observed = [], 0
while observed <= MAXIMUM_LOCAL_FILE_BYTES:
remaining_scan_seconds()
chunk = os.read(descriptor, min(HTTP_RESPONSE_CHUNK_BYTES, MAXIMUM_LOCAL_FILE_BYTES + 1 - observed))
if not chunk:
break
chunks.append(chunk)
observed += len(chunk)
payload = b"".join(chunks)
after = os.fstat(descriptor)
identity_before = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns)
identity_after = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns)
if len(payload) != before.st_size or identity_before != identity_after:
raise RuntimeError(f"local archive file changed while snapshotted: {location}")
return LocalSnapshot(location, payload, hashlib.sha256(payload).hexdigest())
def walk_archive_directory(directory_fd: int, prefix: str, state: dict, snapshots: list[LocalSnapshot]) -> None:
remaining_scan_seconds()
before = os.fstat(directory_fd)
if (not stat.S_ISDIR(before.st_mode) or before.st_uid != 0 or before.st_mode & 0o022):
raise RuntimeError(f"local archive directory is not root-owned read-only: {prefix or '/'}")
names = sorted(os.listdir(directory_fd))
remaining_scan_seconds()
for name in names:
remaining_scan_seconds()
if (not isinstance(name, str) or name in {"", ".", ".."}
or "/" in name or "\\" in name
or any(ord(char) <= 0x1f or ord(char) == 0x7f for char in name)):
raise RuntimeError("local archive contains a non-canonical name")
state["entries"] += 1
if state["entries"] > MAXIMUM_LOCAL_ENTRIES:
raise RuntimeError("local archive population exceeds its signed entry bound")
location = f"{prefix}/{name}" if prefix else name
flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0)
child_fd = os.open(name, flags, dir_fd=directory_fd)
try:
metadata = os.fstat(child_fd)
inode = (metadata.st_dev, metadata.st_ino)
if inode in state["inodes"]:
raise RuntimeError("local archive contains duplicate inode aliases")
state["inodes"].add(inode)
if stat.S_ISDIR(metadata.st_mode):
walk_archive_directory(child_fd, location, state, snapshots)
elif stat.S_ISREG(metadata.st_mode):
snapshot = read_open_snapshot(child_fd, location)
state["bytes"] += len(snapshot.payload)
if state["bytes"] > MAXIMUM_LOCAL_AGGREGATE_BYTES:
raise RuntimeError("local archive population exceeds its signed aggregate bound")
snapshots.append(snapshot)
else:
raise RuntimeError(f"local archive contains a special file: {location}")
finally:
os.close(child_fd)
after = os.fstat(directory_fd)
if (before.st_dev, before.st_ino, before.st_mtime_ns, before.st_ctime_ns) != (
after.st_dev, after.st_ino, after.st_mtime_ns, after.st_ctime_ns):
raise RuntimeError(f"local archive directory changed during enumeration: {prefix or '/'}")
def snapshot_local_archives(archive_dir: Path) -> list[LocalSnapshot]:
if not archive_dir.is_absolute():
raise RuntimeError("signed local archive path must be absolute")
require_trusted_ancestors(archive_dir / ".snapshot-boundary")
flags = os.O_RDONLY | os.O_CLOEXEC | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
root_fd = os.open(archive_dir, flags)
snapshots: list[LocalSnapshot] = []
try:
walk_archive_directory(root_fd, "", {"entries": 0, "bytes": 0, "inodes": set()}, snapshots)
finally:
os.close(root_fd)
return snapshots
def scan_local_archives(markers: dict, archive_dir: Path):
rules = build_yara_rules(markers)
snapshots = snapshot_local_archives(archive_dir)
hits: list[MarkerHit] = []
for snapshot in snapshots:
remaining_scan_seconds()
yara_matches = rules.match(data=snapshot.payload)
remaining_scan_seconds()
exact_hits = scan_payload(markers, "local_archives", snapshot.location, snapshot.payload)
if bool(yara_matches) != bool(exact_hits):
raise RuntimeError(f"YARA and exact-byte scans disagree for {snapshot.location}")
hits.extend(exact_hits)
population = [{"location": item.location, "sha256": item.sha256, "bytes": len(item.payload)} for item in snapshots]
population_digest = digest(population)
return hits, population_digest, len(snapshots), sum(len(item.payload) for item in snapshots)
def run_scan() -> None:
start_scan_deadline()
remaining_scan_seconds()
hits: list[MarkerHit] = []
completed_items: list[dict] = []
local_population_digest = digest([])
local_entry_count = local_aggregate_bytes = 0
for item in WORK_ITEMS:
remaining_scan_seconds()
kind = item["kind"]
if kind == "github_code_search":
marker = item["marker"]
hits.extend(scan_github_code(
{marker: ACTIVE_MARKERS[marker]},
{"enabled": True, "destination_id": item["destination_id"],
"expected_owner_allowlist": item["expected_owner_allowlist"]},
))
elif kind == "huggingface_artifact":
hits.extend(scan_huggingface(ACTIVE_MARKERS, [item["artifact"]]))
elif kind in {"html_watchlist", "paste_watchlist"}:
hits.extend(scan_html_url(ACTIVE_MARKERS, item["entry"], kind))
elif kind == "local_archives":
local_hits, observed_digest, observed_count, observed_bytes = scan_local_archives(
ACTIVE_MARKERS, Path(item["path"])
)
if (not hmac.compare_digest(observed_digest, item["expected_population_sha256"])
or observed_count != item["expected_file_count"]
or observed_bytes != item["expected_aggregate_bytes"]):
raise RuntimeError("local archive readback differs from its signed population commitment")
local_population_digest = observed_digest
local_entry_count, local_aggregate_bytes = observed_count, observed_bytes
hits.extend(local_hits)
else:
raise RuntimeError(f"unhandled signed work-item kind: {kind}")
# A logical item becomes complete only after its entire scanner path returns.
remaining_scan_seconds()
completed_items.append(item)
remaining_scan_seconds()
completed_population_digest = digest(completed_items)
if (len(completed_items) != REQUIRED_POPULATION["count"]
or not hmac.compare_digest(completed_population_digest, REQUIRED_POPULATION["sha256"])):
raise RuntimeError("completed work-item population differs from the signed requirement")
remaining_scan_seconds()
scanned_at = datetime.now(timezone.utc).isoformat()
for hit in hits:
remaining_scan_seconds()
emit_external_marker_hit(hit, scanned_at)
remote_receipts_digest = digest(REMOTE_FETCH_RECEIPTS)
for marker_id, metadata in sorted(ACTIVE_MARKERS.items()):
remaining_scan_seconds()
marker_hits = [hit for hit in hits if hit.marker == marker_id]
payload = with_event_id({
"schema": "aidefend.dv004.external-scan.v1",
"outcome": "FAIL" if marker_hits else "PASS",
"tenant_id": SCOPE["tenant_id"],
"application_id": SCOPE["application_id"],
"marker_id": metadata["marker_id"],
"marker_sha256": hashlib.sha256(marker_id.encode("ascii")).hexdigest(),
"trap_id": metadata["trap_id"],
"registry_digest": metadata["activation_registry_digest"],
"surface_registry_sha256": REGISTRY["_verified_sha256"],
"surface_registry_key_sha256": REGISTRY["_verification_key_sha256"],
"surface_inventory_digest": SURFACE_INVENTORY_DIGEST,
"required_population_source_sha256": REQUIRED_SOURCE_SHA256,
"required_population_digest": REQUIRED_POPULATION["sha256"],
"required_population_count": REQUIRED_POPULATION["count"],
"completed_population_digest": completed_population_digest,
"completed_population_count": len(completed_items),
"population_complete": True,
"marker_count": len(ACTIVE_MARKERS),
"verified_hit_count": len(marker_hits),
"remote_entries": BUDGET.remote_entries,
"remote_bytes": BUDGET.remote_bytes,
"remote_fetch_receipts_digest": remote_receipts_digest,
"local_entries": local_entry_count,
"local_bytes": local_aggregate_bytes,
"local_population_digest": local_population_digest,
"scanned_at": scanned_at,
})
print(canonical(signed_envelope(payload)).decode("utf-8"))
if __name__ == "__main__":
run_scan()Step 3: Treat hits as incident signals and prove scan completeness
External hits should open an investigation that correlates marker activation time, caller identity, access logs, downstream sharing path, and any matching watermark/provenance detector output owned by AID-H-013.001. Persist the root-key digest, signed registry and surface-inventory digests, scanner release, provider pagination/fetch receipts, local population digest, every hit location, and completed population counts. A provider call that returns successfully does not establish coverage when pagination, an immutable artifact, a watchlist entry, or a finalized local snapshot is missing.
Action: Route every remote request through the signed-destination socket-pinning function, and apply a deny-by-default workload egress policy as defense in depth. The function rejects credentials, non-HTTPS ports, unsigned host/path pairs, any DNS answer that is not globally routable, authenticated redirects, unsigned redirect destinations, compressed bodies, and per-response or aggregate overruns; TLS certificate and hostname validation still use the fixed root-owned CA bundle.
The compiled one-hour overall deadline, 512 MiB retained-local/remote ceilings, and entry ceilings are implementation capacity guards, not organizational defaults or recommended scan sizes. Signed policy must select values within them; exceeding either signed or compiled capacity raises ERROR and cannot publish a complete or PASS record.
Finalize local archives into a root-owned, non-writable tree before scanning. Directory-file-descriptor traversal rejects symlinks, special files, hard-link aliases, writable files/directories, mutation, and count/byte overruns. YARA and exact marker matching consume the same immutable in-memory bytes and must agree before the run can publish a complete result. Wrap the completed v1 scan record and per-hit records in the separately controlled evidence signer/outbox expected by correlation; a thrown exception is ERROR, never PASS or complete coverage.
AID-DV-004-G003
Correlate signed trap activations and verified external marker hits into an immutable investigation record, attaching watermark or provenance results only by reference to AID-H-013.001.
Runtime policy: Load freshness, evidence-size, transport, and verifier bounds from the fixed-path Cosign-verified correlation policy. Independently Cosign-verify the fixed v3 surface registry and recompute its source, surface-inventory, and logical-work-item commitments before trusting scanner evidence. Signed values are additionally constrained by compiled implementation-capacity ceilings; an authentic but operationally unsafe value is ERROR.
Verify producer evidence independently
# File: deception/leak_investigation.py
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import math
import os
import re
import ssl
import stat
import subprocess
import tempfile
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote, urlsplit
from urllib.request import HTTPSHandler, HTTPRedirectHandler, ProxyHandler, Request, build_opener
class EvidenceFault(Exception):
pass
COSIGN = Path("/usr/local/bin/cosign") # image pins Cosign v3.1.2
REGISTRY_PATH = Path("/run/policy/dv004-marker-registry.json")
REGISTRY_BUNDLE = Path("/run/policy/dv004-marker-registry.json.sigstore.json")
REGISTRY_KEY = Path("/run/trust/dv004-marker-registry.pub")
POLICY_PATH = Path("/run/policy/dv004-correlation-policy.json")
POLICY_BUNDLE = Path("/run/policy/dv004-correlation-policy.json.sigstore.json")
POLICY_KEY = Path("/run/trust/dv004-correlation-policy.pub")
KEY_PATHS = {
"scanner": Path("/run/secrets/deception/dv004-scanner-hmac.key"),
"activation": Path("/run/secrets/deception/dv004-activation-hmac.key"),
"h013": Path("/run/secrets/deception/h013-receipt-hmac.key"),
"result": Path("/run/secrets/deception/dv004-correlation-signing.key"),
}
EVIDENCE_CA_PATH = Path("/run/trust/evidence-ca.pem")
EVIDENCE_WRITER_CERT_PATH = Path("/run/secrets/deception/evidence-writer-client.crt")
EVIDENCE_WRITER_KEY_PATH = Path("/run/secrets/deception/evidence-writer-client.key")
EVIDENCE_READER_CERT_PATH = Path("/run/secrets/deception/evidence-reader-client.crt")
EVIDENCE_READER_KEY_PATH = Path("/run/secrets/deception/evidence-reader-client.key")
BOOTSTRAP_BYTES = 2_000_000
KEY_BYTES = 65_536
COSIGN_DEADLINE = 10.0
INPUT_BYTES_CAP = 2_000_000
RECEIPT_BYTES_CAP = 1_000_000
RESULT_BODY_BYTES_CAP = 1_000_000
TLS_MATERIAL_BYTES_CAP = 1_000_000
TRANSPORT_SECONDS_CAP = 60.0
FRESHNESS_SECONDS_CAP = 31_536_000
REQUEST_DEADLINE_IMPLEMENTATION_CEILING_SECONDS = 60.0
MARKER_COUNT_CEILING = 4_096
FILE_BYTES_CEILING = 67_108_864
HTTP_RESPONSE_BYTES_CEILING = 67_108_864
HTTP_CHUNK_BYTES_CEILING = 1_048_576
REMOTE_ENTRY_CEILING = 25_000
REMOTE_AGGREGATE_BYTES_CEILING = 536_870_912
WATCHLIST_ENTRY_CEILING = 4_096
LOCAL_ENTRY_CEILING = 10_000
LOCAL_AGGREGATE_BYTES_CEILING = 536_870_912
HEX64 = re.compile(r"[0-9a-f]{64}")
MARKER = re.compile(r"aidefend-trap-output-[0-9a-f]{32}")
ACTIVATION_ENVELOPE_DOMAIN = b"aidefend.dv004.activation-envelope.v1\x00"
SCAN_FIELDS = set("""schema event_id outcome tenant_id application_id marker_id
marker_sha256 trap_id registry_digest surface_registry_sha256
surface_registry_key_sha256 surface_inventory_digest
required_population_source_sha256 required_population_digest
required_population_count completed_population_digest completed_population_count
population_complete marker_count verified_hit_count remote_entries remote_bytes
remote_fetch_receipts_digest local_entries local_bytes local_population_digest
scanned_at""".split())
HIT_FIELDS = set("""schema event_id tenant_id application_id marker_id marker_sha256
trap_id registry_digest surface_registry_sha256 surface_inventory_digest surface
artifact_digest location observed_at""".split())
ACTIVATION_FIELDS = set("""schema event_id tenant_id application_id marker_id trap_id
registry_digest request_digest marker_sha256 activated_at model_invocations""".split())
WATERMARK_FIELDS = set("""schema check_id tenant_id artifact_digest detector_release
policy_digest""".split())
HIT_SURFACES = {
"github_code_search", "huggingface_model", "huggingface_dataset",
"html_watchlist", "paste_watchlist", "local_archives",
}
class NoRedirect(HTTPRedirectHandler):
def redirect_request(self, request, file_pointer, code, message, headers, new_url):
return None
def unique(pairs):
value = {}
for name, item in pairs:
if name in value:
raise EvidenceFault(f"duplicate JSON key: {name}")
value[name] = item
return value
def nonfinite(value):
raise EvidenceFault(f"non-finite JSON constant: {value}")
def canonical(value: Any) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False,
).encode("utf-8")
def sha(value: Any) -> str:
return hashlib.sha256(canonical(value)).hexdigest()
def ancestors(path: Path) -> None:
absolute = Path(os.path.abspath(path))
current = Path(absolute.anchor)
for part in absolute.parent.parts[1:]:
current /= part
meta = current.lstat()
if (stat.S_ISLNK(meta.st_mode) or not stat.S_ISDIR(meta.st_mode)
or meta.st_uid != 0 or meta.st_mode & 0o022):
raise EvidenceFault(f"untrusted fixed-path ancestor: {current}")
def read_stable(path: Path, maximum: int, privileged: bool = False,
private: bool = False) -> bytes:
if privileged or private:
ancestors(path)
try:
fd = os.open(
path, os.O_RDONLY | os.O_CLOEXEC
| getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_NONBLOCK", 0),
)
except OSError as exc:
raise EvidenceFault(f"cannot open {path}: {exc}") from exc
try:
before = os.fstat(fd)
if not stat.S_ISREG(before.st_mode) or not 1 <= before.st_size <= maximum:
raise EvidenceFault(f"unbounded or non-regular artifact: {path}")
if private:
if (before.st_uid != 0 or before.st_nlink != 1
or stat.S_IMODE(before.st_mode) not in {0o400, 0o600}):
raise EvidenceFault(
f"secret must be a single-link root-owned 0400/0600 file: {path}"
)
elif privileged and (before.st_uid != 0 or before.st_mode & 0o022):
raise EvidenceFault(f"fixed artifact is not root-owned read-only: {path}")
chunks, total = [], 0
while total <= maximum:
chunk = os.read(fd, min(65_536, maximum + 1 - total))
if not chunk:
break
chunks.append(chunk)
total += len(chunk)
data, after = b"".join(chunks), os.fstat(fd)
identity = lambda item: (
item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns,
item.st_mode, item.st_uid, item.st_nlink,
)
if len(data) != before.st_size or len(data) > maximum or identity(before) != identity(after):
raise EvidenceFault(f"artifact changed during read: {path}")
return data
finally:
os.close(fd)
def decode(data: bytes, label: str) -> Any:
try:
return json.loads(
data.decode("utf-8", errors="strict"),
object_pairs_hook=unique, parse_constant=nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise EvidenceFault(f"{label} is not strict JSON: {exc}") from exc
def cosign_json(path: Path, bundle: Path, key_path: Path, label: str) -> tuple[dict, str, str]:
ancestors(COSIGN)
binary = COSIGN.stat(follow_symlinks=False)
if (not stat.S_ISREG(binary.st_mode) or binary.st_uid != 0
or binary.st_mode & 0o022 or not binary.st_mode & 0o111):
raise EvidenceFault("fixed Cosign v3.1.2 binary is not trusted")
payload = read_stable(path, BOOTSTRAP_BYTES, True)
signature = read_stable(bundle, BOOTSTRAP_BYTES, True)
public_key = read_stable(key_path, KEY_BYTES, True)
with tempfile.TemporaryDirectory(prefix="dv004-verify-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
copies = (root / "payload.json", root / "bundle.json", root / "key.pub")
for target, data in zip(copies, (payload, signature, public_key), strict=True):
target.write_bytes(data)
os.chmod(target, 0o400)
try:
subprocess.run(
[str(COSIGN), "verify-blob", "--key", str(copies[2]),
"--bundle", str(copies[1]), str(copies[0])],
check=True, capture_output=True, timeout=COSIGN_DEADLINE,
)
except (OSError, subprocess.SubprocessError) as exc:
raise EvidenceFault(f"{label} Cosign verification failed: {exc}") from exc
if tuple(item.read_bytes() for item in copies) != (payload, signature, public_key):
raise EvidenceFault(f"verified {label} snapshot changed")
document = decode(payload, label)
if not isinstance(document, dict):
raise EvidenceFault(f"{label} root is not an object")
return document, hashlib.sha256(payload).hexdigest(), hashlib.sha256(public_key).hexdigest()
def load_policy() -> tuple[dict, str]:
policy, policy_sha, _ = cosign_json(POLICY_PATH, POLICY_BUNDLE, POLICY_KEY, "policy")
fields = {
"schema", "policy_version", "max_scan_age_seconds", "max_hit_age_seconds",
"max_correlation_age_seconds", "max_future_skew_seconds",
"evidence_timeout_seconds", "input_artifact_max_bytes",
"result_body_max_bytes", "receipt_max_bytes", "evidence_endpoint",
}
if set(policy) != fields or policy["schema"] != "aidefend.dv004.correlation-policy.v1":
raise EvidenceFault("correlation policy schema differs")
for name in ("max_scan_age_seconds", "max_hit_age_seconds", "max_correlation_age_seconds"):
if type(policy[name]) is not int or not 1 <= policy[name] <= FRESHNESS_SECONDS_CAP:
raise EvidenceFault(f"signed freshness exceeds implementation capacity: {name}")
if (type(policy["max_future_skew_seconds"]) is not int
or not 0 <= policy["max_future_skew_seconds"] <= 3600):
raise EvidenceFault("signed future skew exceeds implementation capacity")
if (type(policy["input_artifact_max_bytes"]) is not int
or not 1 <= policy["input_artifact_max_bytes"] <= INPUT_BYTES_CAP
or type(policy["result_body_max_bytes"]) is not int
or not 1 <= policy["result_body_max_bytes"] <= RESULT_BODY_BYTES_CAP
or type(policy["receipt_max_bytes"]) is not int
or not policy["result_body_max_bytes"] <= policy["receipt_max_bytes"] <= RECEIPT_BYTES_CAP):
raise EvidenceFault("signed evidence bytes exceed implementation capacity")
timeout = policy["evidence_timeout_seconds"]
if (type(timeout) not in (int, float) or isinstance(timeout, bool)
or not math.isfinite(float(timeout))
or not 0 < timeout <= TRANSPORT_SECONDS_CAP):
raise EvidenceFault("signed transport timeout exceeds implementation capacity")
endpoint_text = bounded_text(
policy["evidence_endpoint"], "evidence endpoint", 2048
)
try:
endpoint = urlsplit(endpoint_text)
hostname, port = endpoint.hostname, endpoint.port
except ValueError as exc:
raise EvidenceFault("signed evidence endpoint is malformed") from exc
if (endpoint.scheme != "https" or not hostname or endpoint.username
or endpoint.password or endpoint.query or endpoint.fragment
or endpoint.path or port is not None
or hostname != hostname.lower().rstrip(".")
or hostname.encode("idna").decode("ascii") != hostname
or endpoint_text != f"https://{hostname}"):
raise EvidenceFault("signed evidence endpoint is not a canonical HTTPS origin")
return policy, policy_sha
def bounded_text(value: Any, label: str, maximum: int = 256) -> str:
if (not isinstance(value, str) or not value.strip() or len(value) > maximum
or any(ord(char) <= 0x1f or ord(char) == 0x7f for char in value)):
raise EvidenceFault(f"{label} is not bounded canonical text")
return value
def validate_destination_url(url: Any, destination_id: Any,
destinations: dict) -> None:
if (not isinstance(url, str) or not url.isascii() or not 1 <= len(url) <= 4096
or "\\" in url or any(ord(char) <= 0x20 or ord(char) == 0x7f for char in url)):
raise EvidenceFault("surface URL is not bounded canonical ASCII")
parsed = urlsplit(url)
if (parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password
or parsed.fragment or parsed.port not in (None, 443)):
raise EvidenceFault("surface URL is not canonical HTTPS on port 443")
if "%" in parsed.path or any(segment in {".", ".."} for segment in parsed.path.split("/")):
raise EvidenceFault("surface URL path contains ambiguous encoding or dot segments")
hostname = parsed.hostname.encode("idna").decode("ascii").lower().rstrip(".")
if hostname != parsed.hostname or parsed.netloc not in {hostname, f"{hostname}:443"}:
raise EvidenceFault("surface URL hostname is not canonical")
destination = destinations.get(destination_id)
if (not destination or destination["hostname"] != hostname
or not any(
(parsed.path or "/") == prefix.rstrip("/")
or (parsed.path or "/").startswith(prefix.rstrip("/") + "/")
for prefix in destination["path_prefixes"]
)):
raise EvidenceFault("surface URL does not match its signed destination")
def derive_items(registry: dict, limits: dict) -> list[dict]:
surfaces = registry["surfaces"]
expected = {"github_code_search", "huggingface_hub", "html_watchlist",
"paste_watchlist", "local_archives"}
if not isinstance(surfaces, dict) or set(surfaces) != expected:
raise EvidenceFault("surface inventory schema differs")
github, hf = surfaces["github_code_search"], surfaces["huggingface_hub"]
html, paste = surfaces["html_watchlist"], surfaces["paste_watchlist"]
local = surfaces["local_archives"]
if (not isinstance(github, dict)
or set(github) != {"enabled", "destination_id", "expected_owner_allowlist"}
or not isinstance(hf, dict) or set(hf) != {"enabled", "artifacts"}
or not isinstance(html, dict) or set(html) != {"enabled", "urls"}
or not isinstance(paste, dict) or set(paste) != {"enabled", "urls"}
or not isinstance(local, dict) or set(local) != {
"enabled", "path", "expected_population_sha256",
"expected_file_count", "expected_aggregate_bytes",
}
or any(type(surface["enabled"]) is not bool for surface in surfaces.values())):
raise EvidenceFault("surface contract differs")
if not any(surface["enabled"] for surface in surfaces.values()):
raise EvidenceFault("all scan surfaces are disabled")
active = sorted(
(marker, meta) for marker, meta in registry["markers"].items() if meta["active"]
)
if not active:
raise EvidenceFault("active marker population is empty")
destinations = registry["egress_destinations"]
owners = github["expected_owner_allowlist"]
if (not isinstance(github["destination_id"], str)
or github["destination_id"] not in destinations
or not isinstance(owners, list) or owners != sorted(set(owners))
or any(not isinstance(owner, str)
or re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,38})", owner) is None
for owner in owners)
or github["enabled"] and not owners):
raise EvidenceFault("GitHub surface is invalid or empty")
artifacts = hf["artifacts"]
if (not isinstance(artifacts, list)
or len(artifacts) > limits["maximum_remote_entries"]
or hf["enabled"] and not artifacts):
raise EvidenceFault("Hugging Face population is invalid or empty")
hf_identities = set()
for artifact in artifacts:
fields = {"repo_id", "repo_type", "revision", "path", "destination_id",
"url", "size", "sha256"}
if not isinstance(artifact, dict) or set(artifact) != fields:
raise EvidenceFault("Hugging Face artifact schema differs")
repo_id, repo_type = artifact["repo_id"], artifact["repo_type"]
revision, artifact_path = artifact["revision"], artifact["path"]
if (repo_type not in {"model", "dataset"}
or not isinstance(repo_id, str)
or re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repo_id) is None
or not isinstance(revision, str) or re.fullmatch(r"[0-9a-f]{40}", revision) is None
or not isinstance(artifact_path, str)
or re.fullmatch(r"[A-Za-z0-9_./-]+", artifact_path) is None
or any(part in {"", ".", ".."} for part in artifact_path.split("/"))
or type(artifact["size"]) is not int
or not 1 <= artifact["size"] <= limits["maximum_huggingface_file_bytes"]
or not isinstance(artifact["sha256"], str)
or HEX64.fullmatch(artifact["sha256"]) is None):
raise EvidenceFault("Hugging Face artifact identity or bound is invalid")
identity = (repo_type, repo_id, revision, artifact_path)
if identity in hf_identities:
raise EvidenceFault("duplicate Hugging Face artifact identity")
hf_identities.add(identity)
prefix = "" if repo_type == "model" else "/datasets"
expected_path = f"{prefix}/{repo_id}/resolve/{revision}/{artifact_path}"
validate_destination_url(artifact["url"], artifact["destination_id"], destinations)
if urlsplit(artifact["url"]).path != expected_path:
raise EvidenceFault("Hugging Face URL is not bound to its immutable identity")
watchlist_entries = 0
for name, watchlist in (("html_watchlist", html), ("paste_watchlist", paste)):
entries = watchlist["urls"]
if not isinstance(entries, list) or watchlist["enabled"] and not entries:
raise EvidenceFault(f"enabled {name} population is empty")
watchlist_entries += len(entries)
encoded_entries = set()
for entry in entries:
if (not isinstance(entry, dict) or set(entry) != {"destination_id", "url"}
or not all(isinstance(entry[field], str) and entry[field] for field in entry)):
raise EvidenceFault(f"{name} entry schema differs")
validate_destination_url(entry["url"], entry["destination_id"], destinations)
encoded = canonical(entry)
if encoded in encoded_entries:
raise EvidenceFault(f"{name} contains a duplicate entry")
encoded_entries.add(encoded)
if watchlist_entries > limits["maximum_watchlist_entries"]:
raise EvidenceFault("watchlist population exceeds its signed bound")
if local["enabled"]:
if (not isinstance(local["path"], str) or not Path(local["path"]).is_absolute()
or not isinstance(local["expected_population_sha256"], str)
or HEX64.fullmatch(local["expected_population_sha256"]) is None
or type(local["expected_file_count"]) is not int
or not 1 <= local["expected_file_count"] <= min(
limits["maximum_local_entries"], LOCAL_ENTRY_CEILING)
or type(local["expected_aggregate_bytes"]) is not int
or not 1 <= local["expected_aggregate_bytes"] <= min(
limits["maximum_local_aggregate_bytes"], LOCAL_AGGREGATE_BYTES_CEILING)):
raise EvidenceFault("enabled local archive commitment is invalid")
items = []
if github["enabled"]:
items.extend({
"kind": "github_code_search", "marker": marker,
"marker_id": meta["marker_id"], "destination_id": github["destination_id"],
"expected_owner_allowlist": owners,
} for marker, meta in active)
if hf["enabled"]:
items.extend({"kind": "huggingface_artifact", "artifact": item} for item in artifacts)
for name, watchlist in (("html_watchlist", html), ("paste_watchlist", paste)):
if watchlist["enabled"]:
items.extend({"kind": name, "entry": item} for item in watchlist["urls"])
if local["enabled"]:
items.append({
"kind": "local_archives", "path": local["path"],
"expected_population_sha256": local["expected_population_sha256"],
"expected_file_count": local["expected_file_count"],
"expected_aggregate_bytes": local["expected_aggregate_bytes"],
})
items.sort(key=canonical)
encoded = [canonical(item) for item in items]
if not items or len(encoded) != len(set(encoded)):
raise EvidenceFault("derived work-item population is empty or duplicated")
return items
def load_registry(marker_id: str | None = None) -> tuple[dict, dict]:
registry, registry_sha, key_sha = cosign_json(
REGISTRY_PATH, REGISTRY_BUNDLE, REGISTRY_KEY, "surface registry"
)
fields = {"schema_version", "scope", "markers", "scanner_limits",
"egress_destinations", "surfaces", "required_population"}
if set(registry) != fields or registry["schema_version"] != "aidefend.dv004-marker-registry.v3":
raise EvidenceFault("surface registry v3 schema differs")
scope, markers = registry["scope"], registry["markers"]
if (not isinstance(scope, dict) or set(scope) != {"tenant_id", "application_id"}
or any(not isinstance(scope[name], str) or not scope[name].strip()
or len(scope[name]) > 128 for name in scope)
or not isinstance(markers, dict) or not markers):
raise EvidenceFault("registry scope or marker population is invalid")
limits = registry["scanner_limits"]
limit_fields = {
"request_timeout_seconds", "maximum_markers", "maximum_huggingface_file_bytes",
"maximum_http_response_bytes", "http_response_chunk_bytes",
"maximum_github_pages", "maximum_local_file_bytes", "maximum_remote_entries",
"maximum_remote_aggregate_bytes", "maximum_watchlist_entries",
"maximum_local_entries", "maximum_local_aggregate_bytes", "maximum_redirects",
}
if not isinstance(limits, dict) or set(limits) != limit_fields:
raise EvidenceFault("signed scanner-limit schema differs")
integer_fields = limit_fields - {"request_timeout_seconds"}
timeout = limits["request_timeout_seconds"]
if (type(timeout) not in (int, float) or isinstance(timeout, bool)
or not math.isfinite(float(timeout))
or not 0 < timeout <= REQUEST_DEADLINE_IMPLEMENTATION_CEILING_SECONDS
or any(type(limits[name]) is not int for name in integer_fields)
or not 1 <= limits["maximum_markers"] <= MARKER_COUNT_CEILING
or not 1 <= limits["maximum_huggingface_file_bytes"] <= FILE_BYTES_CEILING
or not 1 <= limits["maximum_local_file_bytes"] <= FILE_BYTES_CEILING
or not 1 <= limits["maximum_http_response_bytes"] <= HTTP_RESPONSE_BYTES_CEILING
or not 1 <= limits["http_response_chunk_bytes"] <= min(
HTTP_CHUNK_BYTES_CEILING, limits["maximum_http_response_bytes"])
or not 1 <= limits["maximum_remote_entries"] <= REMOTE_ENTRY_CEILING
or not 1 <= limits["maximum_remote_aggregate_bytes"] <= REMOTE_AGGREGATE_BYTES_CEILING
or not 1 <= limits["maximum_watchlist_entries"] <= WATCHLIST_ENTRY_CEILING
or not 1 <= limits["maximum_local_entries"] <= LOCAL_ENTRY_CEILING
or not 1 <= limits["maximum_local_aggregate_bytes"] <= LOCAL_AGGREGATE_BYTES_CEILING
or not 1 <= limits["maximum_github_pages"] <= 10
or not 0 <= limits["maximum_redirects"] <= 5):
raise EvidenceFault("signed scanner limits exceed implementation capacity")
if not 1 <= len(markers) <= limits["maximum_markers"]:
raise EvidenceFault("marker population exceeds its signed bound")
logical_ids = set()
for marker, meta in markers.items():
if (not isinstance(marker, str) or MARKER.fullmatch(marker) is None
or not isinstance(meta, dict)
or set(meta) != {"active", "activation_registry_digest", "marker_id", "trap_id", "owner"}
or type(meta["active"]) is not bool
or not isinstance(meta["activation_registry_digest"], str)
or HEX64.fullmatch(meta["activation_registry_digest"]) is None
or any(not isinstance(meta[name], str) or not meta[name].strip()
for name in ("marker_id", "trap_id", "owner"))
or meta["marker_id"] in logical_ids):
raise EvidenceFault("registry marker entry is invalid or duplicated")
logical_ids.add(meta["marker_id"])
selected = [meta for meta in markers.values() if meta["marker_id"] == marker_id]
all_inactive = not any(meta["active"] for meta in markers.values())
if all_inactive and len(selected) == 1:
source = {name: registry[name] for name in
("schema_version", "scope", "markers", "egress_destinations", "surfaces")}
required = registry["required_population"]
if (not isinstance(registry["egress_destinations"], dict)
or not isinstance(registry["surfaces"], dict)
or not isinstance(required, dict)
or set(required) != {"algorithm", "source_sha256", "count", "sha256"}
or required["algorithm"] != "aidefend.dv004.logical-work-items.v1"
or required["count"] != 0
or not isinstance(required["source_sha256"], str)
or not hmac.compare_digest(required["source_sha256"], sha(source))
or not isinstance(required["sha256"], str)
or not hmac.compare_digest(required["sha256"], sha([]))):
raise EvidenceFault("inactive registry empty-population commitment differs")
return registry, {
"surface_registry_sha256": registry_sha,
"surface_registry_key_sha256": key_sha,
"surface_inventory_digest": sha({
"scope": scope,
"egress_destinations": registry["egress_destinations"],
"surfaces": registry["surfaces"],
}),
"required_population_source_sha256": sha(source),
"required_population_digest": sha([]),
"required_population_count": 0,
}
destinations = registry["egress_destinations"]
if (not isinstance(destinations, dict)
or not 1 <= len(destinations) <= limits["maximum_remote_entries"]):
raise EvidenceFault("egress destination population is invalid")
for destination_id, destination in destinations.items():
if (not isinstance(destination_id, str)
or re.fullmatch(r"[a-z0-9][a-z0-9-]{0,63}", destination_id) is None
or not isinstance(destination, dict)
or set(destination) != {"hostname", "port", "path_prefixes"}):
raise EvidenceFault("egress destination schema differs")
hostname, prefixes = destination["hostname"], destination["path_prefixes"]
if (not isinstance(hostname, str) or hostname != hostname.lower().rstrip(".")
or hostname.encode("idna").decode("ascii") != hostname
or type(destination["port"]) is not int or destination["port"] != 443
or not isinstance(prefixes, list) or not prefixes
or len(prefixes) != len(set(prefixes))
or any(not isinstance(prefix, str) or not prefix.startswith("/")
or "\\" in prefix or any(ord(char) < 0x20 for char in prefix)
for prefix in prefixes)):
raise EvidenceFault("egress destination value is invalid")
items = derive_items(registry, limits)
source = {name: registry[name] for name in
("schema_version", "scope", "markers", "egress_destinations", "surfaces")}
required = registry["required_population"]
if (not isinstance(required, dict)
or set(required) != {"algorithm", "source_sha256", "count", "sha256"}
or required["algorithm"] != "aidefend.dv004.logical-work-items.v1"
or not isinstance(required["source_sha256"], str)
or HEX64.fullmatch(required["source_sha256"]) is None
or not isinstance(required["sha256"], str)
or HEX64.fullmatch(required["sha256"]) is None
or type(required["count"]) is not int or required["count"] != len(items)
or not hmac.compare_digest(required["source_sha256"], sha(source))
or not hmac.compare_digest(required["sha256"], sha(items))):
raise EvidenceFault("signed population commitment differs from independent derivation")
inventory = sha({"scope": scope, "egress_destinations": destinations,
"surfaces": registry["surfaces"]})
return registry, {
"surface_registry_sha256": registry_sha,
"surface_registry_key_sha256": key_sha,
"surface_inventory_digest": inventory,
"required_population_source_sha256": sha(source),
"required_population_digest": sha(items),
"required_population_count": len(items),
}
def secret(name: str) -> bytes:
value = read_stable(KEY_PATHS[name], 64, private=True)
if not 32 <= len(value) <= 64:
raise EvidenceFault(f"{name} HMAC key must contain 32..64 bytes")
return value
def envelope(path: str, schema: str, fields: set[str], key_name: str,
policy: dict, deterministic: bool = False) -> dict:
document = decode(read_stable(Path(path), policy["input_artifact_max_bytes"]), schema)
if not isinstance(document, dict) or set(document) != {"payload", "hmac_sha256"}:
raise EvidenceFault(f"{schema} envelope schema differs")
payload, supplied = document["payload"], document["hmac_sha256"]
if (not isinstance(payload, dict) or set(payload) != fields
or payload.get("schema") != schema
or not isinstance(supplied, str) or HEX64.fullmatch(supplied) is None):
raise EvidenceFault(f"{schema} payload schema differs")
message = canonical(payload)
if schema == "aidefend.dv004.activation.v2":
message = ACTIVATION_ENVELOPE_DOMAIN + message
expected = hmac.new(secret(key_name), message, hashlib.sha256).hexdigest()
if not hmac.compare_digest(expected, supplied):
raise EvidenceFault(f"{schema} HMAC verification failed")
if deterministic:
base = {name: value for name, value in payload.items() if name != "event_id"}
expected_id = "dv004-" + hashlib.sha256(canonical(base)).hexdigest()
if (not isinstance(payload.get("event_id"), str)
or not hmac.compare_digest(payload["event_id"], expected_id)):
raise EvidenceFault(f"{schema} deterministic event ID differs")
elif schema == "aidefend.dv004.activation.v2":
try:
parsed_id = uuid.UUID(payload["event_id"])
except (KeyError, TypeError, ValueError, AttributeError) as exc:
raise EvidenceFault("activation event ID is not a UUIDv4") from exc
if (parsed_id.version != 4 or parsed_id.variant != uuid.RFC_4122
or str(parsed_id) != payload["event_id"]):
raise EvidenceFault("activation event ID is not canonical UUIDv4")
else:
bounded_text(payload["check_id"], "watermark check ID", 128)
return payload
def timestamp(record: dict, field: str, policy: dict, label: str) -> datetime:
try:
value = record[field]
observed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except (KeyError, TypeError, AttributeError, ValueError) as exc:
raise EvidenceFault(f"{label} timestamp is invalid") from exc
if not isinstance(value, str) or observed.tzinfo is None:
raise EvidenceFault(f"{label} timestamp lacks timezone")
normalized = observed.astimezone(timezone.utc)
if (normalized - datetime.now(timezone.utc)).total_seconds() > policy["max_future_skew_seconds"]:
raise EvidenceFault(f"{label} timestamp is in the future")
return normalized
def is_fresh(observed: datetime, maximum: int) -> bool:
return (datetime.now(timezone.utc) - observed).total_seconds() <= maximum
def correlate(args: argparse.Namespace, policy: dict, registry: dict,
commitments: dict) -> dict:
matches = [(marker, meta) for marker, meta in registry["markers"].items()
if meta["marker_id"] == args.marker_id]
if len(matches) != 1:
raise EvidenceFault("logical marker ID is absent or ambiguous")
marker, meta = matches[0]
scope = registry["scope"]
identity = {
"tenant_id": scope["tenant_id"], "application_id": scope["application_id"],
"marker_id": meta["marker_id"],
"marker_sha256": hashlib.sha256(marker.encode("ascii")).hexdigest(),
"trap_id": meta["trap_id"], "registry_digest": meta["activation_registry_digest"],
"surface_registry_sha256": commitments["surface_registry_sha256"],
}
supplied_runtime = bool(args.scan or args.hit or args.activation or args.watermark)
if not meta["active"]:
if supplied_runtime:
raise EvidenceFault("inactive marker has contradictory runtime evidence")
return {**identity, "schema": "aidefend.dv004.investigation.v1",
"outcome": "NOT_APPLICABLE", "finding_state": "marker_inactive",
"evidence_refs": [commitments["surface_registry_sha256"]]}
expected = {**identity, **commitments}
scan = None
coverage_complete = False
coverage_gaps = []
if args.scan:
scan = envelope(args.scan, "aidefend.dv004.external-scan.v1",
SCAN_FIELDS, "scanner", policy, True)
bound = ("tenant_id", "application_id", "marker_id", "marker_sha256", "trap_id",
"registry_digest", "surface_registry_sha256", "surface_registry_key_sha256",
"surface_inventory_digest", "required_population_source_sha256",
"required_population_digest", "required_population_count")
if any(scan[name] != expected[name] for name in bound):
raise EvidenceFault("scan-to-registry binding differs")
digest_fields = (
"marker_sha256", "registry_digest", "surface_registry_sha256",
"surface_registry_key_sha256", "surface_inventory_digest",
"required_population_source_sha256", "required_population_digest",
"completed_population_digest", "remote_fetch_receipts_digest",
"local_population_digest",
)
if any(not isinstance(scan[name], str) or HEX64.fullmatch(scan[name]) is None
for name in digest_fields):
raise EvidenceFault("scan digest field is invalid")
counters = ("required_population_count", "completed_population_count", "marker_count",
"verified_hit_count", "remote_entries", "remote_bytes",
"local_entries", "local_bytes")
if any(type(scan[name]) is not int or scan[name] < 0 for name in counters):
raise EvidenceFault("scan counter is invalid")
limits = registry["scanner_limits"]
if (scan["remote_entries"] > limits["maximum_remote_entries"]
or scan["remote_bytes"] > limits["maximum_remote_aggregate_bytes"]
or scan["local_entries"] > limits["maximum_local_entries"]
or scan["local_bytes"] > limits["maximum_local_aggregate_bytes"]):
raise EvidenceFault("scan counter exceeds the signed registry bound")
active_count = sum(1 for value in registry["markers"].values() if value["active"])
if scan["marker_count"] != active_count:
raise EvidenceFault("scan active-marker count differs from the registry")
local = registry["surfaces"]["local_archives"]
expected_local = (
(local["expected_population_sha256"], local["expected_file_count"],
local["expected_aggregate_bytes"])
if local["enabled"] else (sha([]), 0, 0)
)
if (scan["local_population_digest"], scan["local_entries"], scan["local_bytes"]) != expected_local:
raise EvidenceFault("local archive readback differs from the signed enabled/disabled state")
complete_measurement = (
scan["completed_population_count"] == scan["required_population_count"]
and hmac.compare_digest(
scan["completed_population_digest"], scan["required_population_digest"])
)
if scan["population_complete"] is True and not complete_measurement:
raise EvidenceFault("scan claims completeness with a contradictory population")
if scan["population_complete"] is not True or not complete_measurement:
coverage_gaps.append("scan_population_incomplete")
scanned_at = timestamp(scan, "scanned_at", policy, "scan")
if not is_fresh(scanned_at, policy["max_scan_age_seconds"]):
coverage_gaps.append("scan_stale")
coverage_complete = not coverage_gaps
else:
coverage_gaps.append("scan_absent")
hits = [envelope(path, "aidefend.dv004.external-hit.v1",
HIT_FIELDS, "scanner", policy, True) for path in args.hit]
if len({hit["event_id"] for hit in hits}) != len(hits):
raise EvidenceFault("duplicate hit event ID")
hit_bound = ("tenant_id", "application_id", "marker_id", "marker_sha256", "trap_id",
"registry_digest", "surface_registry_sha256", "surface_inventory_digest")
hit_times = []
for hit in hits:
if any(hit[name] != expected[name] for name in hit_bound):
raise EvidenceFault("hit-to-registry binding differs")
if (hit["surface"] not in HIT_SURFACES
or not isinstance(hit["artifact_digest"], str)
or HEX64.fullmatch(hit["artifact_digest"]) is None
or not isinstance(hit["location"], str) or not hit["location"]):
raise EvidenceFault("hit surface, artifact, or location is invalid")
observed = timestamp(hit, "observed_at", policy, "hit")
hit_times.append(observed)
if scan is not None and hit["observed_at"] != scan["scanned_at"]:
raise EvidenceFault("hit timestamp differs from its scanner record")
if scan is not None:
hit_count = scan["verified_hit_count"]
if scan["outcome"] != ("FAIL" if hit_count else "PASS"):
raise EvidenceFault("scan outcome contradicts its verified-hit count")
if len(hits) != hit_count:
raise EvidenceFault("supplied hit population differs from the signed scan count")
activations = [envelope(path, "aidefend.dv004.activation.v2",
ACTIVATION_FIELDS, "activation", policy) for path in args.activation]
if len({item["event_id"] for item in activations}) != len(activations):
raise EvidenceFault("duplicate activation event ID")
activation_bound = ("tenant_id", "application_id", "marker_id", "marker_sha256",
"trap_id", "registry_digest")
activation_refs = []
fresh_activation_ids = []
after_observation_ids = []
correlation_gaps = []
observation_times = list(hit_times)
if scan is not None:
observation_times.append(timestamp(scan, "scanned_at", policy, "scan"))
earliest_observation = min(observation_times) if observation_times else None
for activation in activations:
if any(activation[name] != identity[name] for name in activation_bound):
raise EvidenceFault("activation-to-registry binding differs")
if (type(activation["model_invocations"]) is not int
or activation["model_invocations"] != 0
or not isinstance(activation["request_digest"], str)
or HEX64.fullmatch(activation["request_digest"]) is None):
raise EvidenceFault("activation v2 execution evidence is invalid")
activated = timestamp(activation, "activated_at", policy, "activation")
after_observation = (
earliest_observation is not None and activated > earliest_observation
)
fresh = (
not after_observation
and is_fresh(activated, policy["max_correlation_age_seconds"])
)
activation_refs.append({
"event_id": activation["event_id"],
"attribution": (
"after_observation" if after_observation else (
"fresh" if fresh else "attribution_stale"
)
),
})
if after_observation:
after_observation_ids.append(activation["event_id"])
if fresh:
fresh_activation_ids.append(activation["event_id"])
if after_observation_ids:
correlation_gaps.append("activation_after_observation")
if hits and not fresh_activation_ids:
correlation_gaps.append("matching_fresh_activation_absent")
if hits and any(not is_fresh(item, policy["max_hit_age_seconds"]) for item in hit_times):
correlation_gaps.append("hit_outside_freshness_window")
watermark = None
if args.watermark:
watermark = envelope(args.watermark, "aidefend.h013.detector-receipt.v1",
WATERMARK_FIELDS, "h013", policy)
if (not hits or watermark["tenant_id"] != scope["tenant_id"]
or watermark["artifact_digest"] not in {hit["artifact_digest"] for hit in hits}
or any(not isinstance(watermark[name], str) or not watermark[name]
for name in WATERMARK_FIELDS)):
raise EvidenceFault("H-013.001 reference is not bound to an external hit")
base_refs = ([scan["event_id"]] if scan is not None else [])
base_refs.extend(hit["event_id"] for hit in hits)
base_refs.extend(item["event_id"] for item in activation_refs)
if watermark is not None:
base_refs.append(watermark["check_id"])
if hits:
return {**identity, "schema": "aidefend.dv004.investigation.v1",
"outcome": "FAIL", "finding_state": "verified_external_marker_hit",
"scan_coverage_complete": coverage_complete,
"coverage_gaps": sorted(set(coverage_gaps)),
"correlation_gaps": sorted(set(correlation_gaps)),
"activation_refs": activation_refs,
"attribution_scope": "marker-level correlation; no unique request attribution",
"external_locations": sorted({hit["location"] for hit in hits}),
"artifact_digests": sorted({hit["artifact_digest"] for hit in hits}),
"evidence_refs": base_refs}
if args.watermark:
raise EvidenceFault("watermark reference supplied without an external hit")
if not coverage_complete:
return {**identity, "schema": "aidefend.dv004.investigation.v1",
"outcome": "INSUFFICIENT_DATA",
"finding_state": "scan_coverage_incomplete",
"scan_coverage_complete": False,
"coverage_gaps": sorted(set(coverage_gaps)),
"correlation_gaps": sorted(set(correlation_gaps)),
"population_binding": commitments,
"activation_refs": activation_refs,
"evidence_refs": base_refs}
if after_observation_ids:
return {**identity, "schema": "aidefend.dv004.investigation.v1",
"outcome": "INSUFFICIENT_DATA",
"finding_state": "post_scan_activation_unmeasured",
"scan_coverage_complete": coverage_complete,
"coverage_gaps": sorted(set(coverage_gaps)),
"correlation_gaps": sorted(set(correlation_gaps)),
"population_binding": commitments,
"activation_refs": activation_refs,
"evidence_refs": base_refs}
if not activation_refs:
return {**identity, "schema": "aidefend.dv004.investigation.v1",
"outcome": "NOT_APPLICABLE",
"finding_state": "marker_not_exposed_in_observation_window",
"scan_coverage_complete": coverage_complete,
"population_binding": commitments,
"activation_refs": [], "evidence_refs": base_refs}
if not fresh_activation_ids:
return {**identity, "schema": "aidefend.dv004.investigation.v1",
"outcome": "INSUFFICIENT_DATA",
"finding_state": "activation_outside_correlation_window",
"scan_coverage_complete": coverage_complete,
"population_binding": commitments,
"activation_refs": activation_refs, "evidence_refs": base_refs}
return {**identity, "schema": "aidefend.dv004.investigation.v1",
"outcome": "PASS", "finding_state": "complete_exposed_population_no_hit",
"activation_refs": activation_refs, "evidence_refs": base_refs}
def write_private_snapshot(path: Path, payload: bytes) -> None:
flags = (os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC
| getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_NONBLOCK", 0))
try:
fd = os.open(path, flags, 0o400)
except OSError as exc:
raise EvidenceFault(f"cannot create private TLS snapshot: {exc}") from exc
try:
os.fchmod(fd, 0o400)
offset = 0
while offset < len(payload):
written = os.write(fd, payload[offset:])
if written <= 0:
raise EvidenceFault("private TLS snapshot write made no progress")
offset += written
os.fsync(fd)
metadata = os.fstat(fd)
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != 0
or metadata.st_nlink != 1 or stat.S_IMODE(metadata.st_mode) != 0o400
or metadata.st_size != len(payload)):
raise EvidenceFault("private TLS snapshot metadata differs")
finally:
os.close(fd)
if not hmac.compare_digest(
read_stable(path, len(payload), private=True), payload):
raise EvidenceFault("private TLS snapshot readback differs")
def persist(record: dict, policy: dict) -> str:
signed = {"payload": record, "hmac_sha256": hmac.new(
secret("result"), canonical(record), hashlib.sha256
).hexdigest()}
body = canonical(signed)
body_sha = hashlib.sha256(body).hexdigest()
if len(body) > policy["result_body_max_bytes"]:
raise EvidenceFault("signed investigation exceeds the signed result-body bound")
ca_bytes = read_stable(
EVIDENCE_CA_PATH, TLS_MATERIAL_BYTES_CAP, privileged=True
)
writer_cert_bytes = read_stable(
EVIDENCE_WRITER_CERT_PATH, TLS_MATERIAL_BYTES_CAP, privileged=True
)
writer_key_bytes = read_stable(
EVIDENCE_WRITER_KEY_PATH, TLS_MATERIAL_BYTES_CAP, private=True
)
reader_cert_bytes = read_stable(
EVIDENCE_READER_CERT_PATH, TLS_MATERIAL_BYTES_CAP, privileged=True
)
reader_key_bytes = read_stable(
EVIDENCE_READER_KEY_PATH, TLS_MATERIAL_BYTES_CAP, private=True
)
writer_cert_sha = hashlib.sha256(writer_cert_bytes).digest()
reader_cert_sha = hashlib.sha256(reader_cert_bytes).digest()
writer_key_sha = hashlib.sha256(writer_key_bytes).digest()
reader_key_sha = hashlib.sha256(reader_key_bytes).digest()
if (hmac.compare_digest(writer_cert_sha, reader_cert_sha)
or hmac.compare_digest(writer_key_sha, reader_key_sha)):
raise EvidenceFault("writer and reader mTLS identities are identical or aliased")
def mtls_opener(label: str, cert_bytes: bytes, key_bytes: bytes):
with tempfile.TemporaryDirectory(prefix=f"aidefend-dv004-{label}-", dir="/run") as directory:
root = Path(directory)
root.chmod(0o700)
ca_snapshot = root / "ca.pem"
cert_snapshot = root / "client.crt"
key_snapshot = root / "client.key"
for snapshot, payload in (
(ca_snapshot, ca_bytes),
(cert_snapshot, cert_bytes),
(key_snapshot, key_bytes),
):
write_private_snapshot(snapshot, payload)
context = ssl.create_default_context(cafile=str(ca_snapshot))
context.load_cert_chain(str(cert_snapshot), str(key_snapshot))
for snapshot, payload in (
(ca_snapshot, ca_bytes),
(cert_snapshot, cert_bytes),
(key_snapshot, key_bytes),
):
if not hmac.compare_digest(
read_stable(snapshot, len(payload), private=True), payload):
raise EvidenceFault("TLS snapshot changed while SSLContext loaded")
return build_opener(
ProxyHandler({}), HTTPSHandler(context=context), NoRedirect()
)
writer = mtls_opener("writer", writer_cert_bytes, writer_key_bytes)
reader = mtls_opener("reader", reader_cert_bytes, reader_key_bytes)
collection = policy["evidence_endpoint"] + "/v1/deception-investigations"
def exchange(opener, request: Request, expected_status: int, label: str) -> bytes:
try:
with opener.open(request, timeout=policy["evidence_timeout_seconds"]) as response:
if response.status != expected_status:
raise EvidenceFault(f"{label} returned HTTP {response.status}")
raw = response.read(policy["receipt_max_bytes"] + 1)
except EvidenceFault:
raise
except Exception as exc:
raise EvidenceFault(f"{label} failed: {exc}") from exc
if not raw or len(raw) > policy["receipt_max_bytes"]:
raise EvidenceFault(f"{label} response exceeds the signed bound")
return raw
posted = exchange(writer, Request(collection, data=body, method="POST", headers={
"Content-Type": "application/json", "Idempotency-Key": body_sha,
"Accept": "application/json", "Accept-Encoding": "identity",
}), 201, "evidence create")
receipt = decode(posted, "create receipt")
receipt_fields = {"schema", "record_id", "object_version", "payload_sha256"}
if (not isinstance(receipt, dict) or set(receipt) != receipt_fields
or receipt["schema"] != "aidefend.evidence.create-receipt.v1"
or not isinstance(receipt["record_id"], str)
or re.fullmatch(r"[A-Za-z0-9_-]{1,128}", receipt["record_id"]) is None
or not isinstance(receipt["object_version"], str)
or re.fullmatch(r"[A-Za-z0-9._-]{1,128}", receipt["object_version"]) is None
or not isinstance(receipt["payload_sha256"], str)
or not hmac.compare_digest(receipt["payload_sha256"], body_sha)):
raise EvidenceFault("evidence create receipt differs")
item_url = (collection + "/" + quote(receipt["record_id"], safe="")
+ "/versions/" + quote(receipt["object_version"], safe=""))
stored_raw = exchange(reader, Request(item_url, method="GET", headers={
"Accept": "application/json", "Accept-Encoding": "identity",
"Cache-Control": "no-cache",
}), 200, "evidence readback")
stored = decode(stored_raw, "stored investigation")
if (not isinstance(stored, dict)
or not hmac.compare_digest(stored_raw, body)):
raise EvidenceFault("independent evidence readback is not the exact signed bytes")
return body_sha
def bounded_error_reason(exc: Exception) -> str:
reason = " ".join(str(exc)[:2048].split())
return (reason or "unspecified failure")[:512]
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--marker-id", required=True)
parser.add_argument("--scan")
parser.add_argument("--hit", action="append", default=[])
parser.add_argument("--activation", action="append", default=[])
parser.add_argument("--watermark")
args = parser.parse_args()
try:
policy, policy_sha = load_policy()
registry, commitments = load_registry(args.marker_id)
except Exception as exc:
print(json.dumps({
"outcome": "ERROR", "error_code": "bootstrap_validation_failed",
"reason": bounded_error_reason(exc),
}, sort_keys=True))
return 4
try:
record = correlate(args, policy, registry, commitments)
except Exception as exc:
record = {
"schema": "aidefend.dv004.investigation.v1",
"outcome": "ERROR", "error_code": "correlation_validation_failed",
"reason": bounded_error_reason(exc),
"requested_marker_id": args.marker_id,
"tenant_id": registry["scope"]["tenant_id"],
"application_id": registry["scope"]["application_id"],
"surface_registry_sha256": commitments["surface_registry_sha256"],
}
record["policy_digest"] = policy_sha
try:
receipt = persist(record, policy)
except EvidenceFault as exc:
print(json.dumps({
"outcome": "ERROR", "error_code": "evidence_persistence_failed",
"reason": bounded_error_reason(exc),
}, sort_keys=True))
return 4
print(json.dumps({"outcome": record["outcome"], "receipt_digest": receipt}, sort_keys=True))
return 2 if record["outcome"] == "FAIL" else (
0 if record["outcome"] in {"PASS", "NOT_APPLICABLE"} else 3
)
if __name__ == "__main__":
raise SystemExit(main())Operational response
Action: Pass the logical marker ID and independently obtained scanner evidence. Repeat --hit for the complete authenticated hit population and --activation for every retained activation envelope for that marker. A marker can have multiple activations, so the record preserves every authenticated activation reference, labels stale ones attribution_stale and later activations after_observation, and makes only marker-level correlation; it never claims unique request attribution.
An authenticated external hit is always FAIL; a later activation remains an activation_after_observation correlation gap and never suppresses that finding. A missing or stale activation is recorded as a correlation gap on a positive finding. For a zero-hit result, incomplete scan coverage is INSUFFICIENT_DATA; an active marker with no retained exposure in the observation window is NOT_APPLICABLE; and exposure that is present but stale or later than the scan is INSUFFICIENT_DATA. PASS requires complete scan coverage plus at least one fresh authenticated activation that the scan window covers. An inactive signed marker with no runtime evidence is also NOT_APPLICABLE.
Malformed signatures, unsafe bounds, schema drift, contradictory populations, future timestamps, mismatched supplied evidence, or failed bounded no-proxy mTLS POST with the independently prevalidated writer identity followed by version-addressed GET exact-byte readback with the separately prevalidated reader identity; CA, certificate, and key bytes are loaded only from verified private snapshots and identical writer/reader material is rejected are ERROR; automation receives a stable error_code plus a bounded diagnostic reason. Optional AID-H-013.001 detector evidence remains a bound reference and never changes the DV-004 outcome.
Open-source tools
- Canarytokens
- FastAPI
- OpenTelemetry
- Beautiful Soup 4
- YARA
Commercial / hosted tools
- Hugging Face Hub (hosted service)
- GitHub Code Search API
- Thinkst Canary
- Splunk Enterprise Security
- Microsoft Sentinel
- Google Security Operations
Threat mappings
MITRE ATLAS
- AML.T0024.002 Exfiltration via AI Inference API: Extract AI Model (trap markers reveal copied or replayed model outputs)
- AML.T0048.004 External Harms: AI Intellectual Property Theft
- AML.T0057 LLM Data Leakage (external marker hits reveal leaked outputs)
- AML.T0005 Create Proxy AI Model (trap markers reveal replicated model behavior)
- AML.T0005.001 Create Proxy AI Model: Train Proxy via Replication (trap markers reveal proxy models trained on captured inference outputs)
MAESTRO
- Model Stealing (L1) (trap markers reveal copied or replayed model behavior)
- Data Exfiltration (L2) (leak beacons reveal exfiltrated outputs or decoy data)
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure (external marker hits reveal leaked sensitive output)
- LLM06:2026 Unbounded Consumption (marker reuse reveals scraping, unauthorized replay, or model-extraction activity)
OWASP ML Top 10 2023
- ML05:2023 Model Theft (trap markers make stolen or replayed model behavior traceable)
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- NISTAML.031 Model Extraction (trap markers reveal copied or replayed model outputs)
- NISTAML.038 Data Extraction (leak beacons detect disclosed trap output or decoy data)
Cisco Integrated AI Security and Safety Framework
- AITech-8.2 Data Exfiltration / Exposure (detects external reuse of registered output markers)
- AITech-8.3 Information Disclosure (detects external reuse of registered output markers)
Google Secure AI Framework 2.0 - Risks
- MRE: Model Reverse Engineering (trap markers reveal copied or replayed model behavior)
- SDD: Sensitive Data Disclosure (external marker hits reveal leaked sensitive outputs)
Databricks AI Security Framework 3.0
- Model Management 8.2: Model theft (trap markers make stolen or replayed behavior traceable)
- Model Serving - Inference response 10.6: Sensitive data output from a model (leak beacons reveal disclosed trap output)
- Agents - Tools MCP Server 13.23: Data Exfiltration (leak beacons detect exfiltrated marker content)