AID-DV-002: Honey Data, Decoy Artifacts & Canary Tokens for AI
Seed and monitor approved AI assets, including training datasets, model repositories, configuration files, and API documentation, with synthetic decoy records, non-functional or instrumented model artifacts, or non-authoritative canary tokens. Access, exfiltration, or attempted use must generate a uniquely attributable alert and evidence without granting production access or disclosing sensitive data.
Pillars: data, infra, model, app
Lifecycle: building, operation
Scope boundary
Owns passive, pre-seeded honey records, decoy artifacts, fake credentials, and canary tokens placed in datasets, repositories, configuration, or documentation. They generate attributable evidence when accessed, copied, exported, or used, without requiring an agent to choose a decoy capability.
AID-DV-005: AID-DV-002 plants passive decoy material; AID-DV-005 presents active canary goals, tools, MCP servers, or sub-agents for agent invocation. Choose the control according to the trigger: access or attempted use of a pre-seeded artifact belongs here, while selection or invocation of a decoy agent capability belongs to AID-DV-005.
AID-DV-004: AID-DV-002 detects interaction with pre-seeded honey material; AID-DV-004 emits exact output markers from controlled trap prompts and investigates marker reuse. AID-DV-002 does not own output-triggered marker activation, external marker scanning, or leak-beacon correlation.
AID-DV-007: AID-DV-002 uses access-triggered operational decoys; AID-DV-007 inserts authorized training canaries and measures their behavior in an exact model version. Training-run insertion provenance, blinded recall measurement, and statistical evidence belong to AID-DV-007 rather than this passive artifact control.
Implementation guidance
- AID-DV-002-G001: Embed unique, synthetic honey records in operational databases or knowledge stores, and alert on access.
- AID-DV-002-G002: Publish fake/instrumented decoy model artifacts.
- AID-DV-002-G003: Create and embed decoy API keys/access tokens (Canary Tokens).
- AID-DV-002-G004: Embed trackable URLs / web bugs in fake sensitive documents.
- AID-DV-002-G005: Watermark synthetic data in honeypots / decoys.
- AID-DV-002-G006: Ensure honey elements are isolated and cannot impact production.
- AID-DV-002-G007: Integrate honey element alerts into security monitoring.
AID-DV-002-G001
Embed unique, synthetic honey records in operational databases or knowledge stores, and alert on access.
When this method applies
Place only synthetic, non-routable records in an explicitly governed operational decoy surface. The authoritative honey registry must be a signed, versioned artifact controlled by the deception team; a mutable local JSON file or a caller-supplied is_honey_record flag is not authoritative. This guidance proves that every registered record is present only in its declared store, is excluded from legitimate consumers, and produces a digest-bound event whenever a storage or retrieval path reads it.
Register and load the exact population
File: deception/honey_registry.json
{
"schema_version": "aidefend.honey-registry.v1",
"registry_version": "2026.07.16.1",
"records": [
{
"honey_id": "hny_crm_01J3F6M5S7Y9",
"surface": "crm.users",
"tenant_scope": "security-decoy",
"synthetic_email": "hny_crm_01J3F6M5S7Y9@example.invalid",
"alert_route": "soc-critical"
},
{
"honey_id": "hny_rag_01J3F70A2B8K",
"surface": "rag.customer-kb",
"tenant_scope": "security-decoy",
"synthetic_email": "hny_rag_01J3F70A2B8K@example.invalid",
"alert_route": "soc-critical"
}
]
}#!/usr/bin/env bash
set -euo pipefail
umask 077
registry=deception/honey_registry.json
bundle=deception/honey_registry.json.sigstore.json
for input_path in "$registry" "$bundle"; do
if [[ ! -f "$input_path" || -L "$input_path" ]]; then
echo "honey registry and bundle must be regular, non-symlink files" >&2
exit 2
fi
done
snapshot_root="$(mktemp -d)"
chmod 0700 "$snapshot_root"
trap 'rm -rf -- "$snapshot_root"' EXIT
registry_snapshot="$snapshot_root/honey_registry.json"
bundle_snapshot="$snapshot_root/honey_registry.json.sigstore.json"
install -m 0400 -- "$registry" "$registry_snapshot"
install -m 0400 -- "$bundle" "$bundle_snapshot"
registry_sha256="$(sha256sum -- "$registry_snapshot" | awk '{print $1}')"
bundle_sha256="$(sha256sum -- "$bundle_snapshot" | awk '{print $1}')"
cosign verify-blob --key /run/trust/honey-registry.pub \
--bundle "$bundle_snapshot" "$registry_snapshot" >/dev/null
if [[ "$(sha256sum -- "$registry_snapshot" | awk '{print $1}')" != "$registry_sha256" \
|| "$(sha256sum -- "$bundle_snapshot" | awk '{print $1}')" != "$bundle_sha256" ]]; then
echo "honey registry snapshot changed during verification" >&2
exit 1
fi
python -m deception.load_honey_registry \
--registry "$registry_snapshot" \
--expected-sha256 "$registry_sha256"Pass only the private verified snapshot and its pre-verification digest to the storage-specific loader; the loader must never reopen the mutable source registry. Use a deployment identity that can create only the registered synthetic rows. Reject real domains, real customer identifiers, duplicates, unknown surfaces, missing tenant scope, and any existing business-key collision. Persist the registry digest on each row/chunk. Direct database audit streams and vector-store/retrieval adapters must emit honey_id, registry digest, store object/version, authenticated actor/workload, request/trace ID, operation, and timestamp to the durable honey-event outbox; do not rely on application print calls.
Independent exact-set and trigger verification
Using read-only identities independent from the loader and alert producer, enumerate the complete registered and observed honey populations per store and require exact equality. Run one controlled read per registered record through every declared access path, then read back exactly one correlated outbox and SIEM event. Run negative controls against ordinary records and require no honey event. Reconcile legitimate analytics, billing, marketing, training, retrieval, and export outputs against the registry and require an empty intersection.
Action: Promote the signed registry and exact record population together, then continuously reconcile store, consumer, and alert populations.
Keep deliberate shared state from becoming a cross-run coordination channel
When a canary intentionally spans runs, bind the registered object or token to an allowed tenant, principal, producer run, consumer run or role, generation, operation, and expiry. Make the decoy payload non-authoritative and non-routable, and prevent workloads from rewriting it into a general mailbox, queue, note, tag, or registry entry. Alert on any read, write, replay, or consume edge outside that declared relationship, while treating immutable content-addressed caches as shared artifacts rather than mutable coordination state. This guidance supplies deception evidence only; authorization and containment remain with the shared-state access control.
AID-DV-002-G002
Publish fake/instrumented decoy model artifacts.
Concept:
A decoy bundle must look like an ordinary high-value model to the party browsing or downloading it. Do not place decoy, honeypot, owner-team, or intended-alert fields in the artifact-visible manifest, object tags, or adjacent files. Keep the trap identity in a separately administered, signed registry that the monitored storage path and alerting service can access but artifact readers cannot.
Step 1: Build a plausible, non-executable public bundle
# File: deception/build_decoy_model_bundle.py
from __future__ import annotations
import json
from pathlib import Path
import torch
from safetensors.torch import save_file
output = Path("deception/out/prod_financial_forecast_model")
output.mkdir(parents=True, exist_ok=True)
save_file(
{
"encoder.weight": torch.zeros((4, 4), dtype=torch.float32),
"encoder.bias": torch.zeros((4,), dtype=torch.float32),
},
str(output / "model.safetensors"),
)
public_manifest = {
"model_name": "prod_financial_forecast_model",
"format": "safetensors",
"task": "tabular-regression",
"framework": "pytorch",
"model_version": "2026.07.1",
}
(output / "manifest.json").write_text(
json.dumps(public_manifest, sort_keys=True, indent=2) + "\n",
encoding="utf-8",
)
Step 2: Register the trap privately and sign the registry entry
File: security-only/decoy_asset_registry.json
# File: deception/build_private_registry.py
from __future__ import annotations
import hashlib
import json
import os
from pathlib import Path
bundle = Path("deception/out/prod_financial_forecast_model")
names = ["manifest.json", "model.safetensors"]
if sorted(path.name for path in bundle.iterdir()) != sorted(names):
raise RuntimeError("public bundle population differs from the registered population")
objects = []
for name in names:
path = bundle / name
if path.is_symlink() or not path.is_file():
raise RuntimeError(f"unsafe decoy object: {name}")
raw = path.read_bytes()
objects.append({"name": name, "sha256": hashlib.sha256(raw).hexdigest(), "bytes": len(raw)})
registry = {
"schema_version": "aidefend.decoy-asset-registry.v3",
"registry_version": os.environ["DECOY_REGISTRY_VERSION"],
"assets": [{
"decoy_id": "decoy-model-7f2d",
"bucket": os.environ["DECOY_MODEL_BUCKET"],
"expected_bucket_owner": os.environ["DECOY_MODEL_BUCKET_OWNER"],
"key_prefix": "models/prod_financial_forecast_model/",
"expected_objects": objects,
"kms_key_arn": os.environ["DECOY_MODEL_KMS_KEY_ARN"],
"object_lock_mode": "COMPLIANCE",
"retain_until": os.environ["DECOY_OBJECT_RETAIN_UNTIL"],
"s3_connect_timeout_seconds": 5,
"s3_read_timeout_seconds": 15,
"s3_max_attempts": 3,
"conditional_conflict_attempts": 3,
"maximum_version_entries": 16,
"incident_severity": "CRITICAL",
"alert_slo_seconds": 300,
}],
}
Path("security-only/decoy_asset_registry.json").write_text(
json.dumps(registry, sort_keys=True, separators=(",", ":")) + "\n",
encoding="utf-8",
)
#!/usr/bin/env bash
set -euo pipefail
umask 077
# Producer job: emit the registry and its Cosign v3 Sigstore bundle.
registry=security-only/decoy_asset_registry.json
bundle=security-only/decoy_asset_registry.json.sigstore.json
cosign sign-blob --yes --key env://DECOY_REGISTRY_SIGNING_KEY \
--bundle "$bundle" "$registry"
#!/usr/bin/env bash
set -euo pipefail
umask 077
# Independent consumer job: verify private snapshots of those exact artifacts.
registry=security-only/decoy_asset_registry.json
bundle=security-only/decoy_asset_registry.json.sigstore.json
for input_path in "$registry" "$bundle"; do
if [[ ! -f "$input_path" || -L "$input_path" ]]; then
echo "decoy registry and bundle must be regular, non-symlink files" >&2
exit 2
fi
done
snapshot_root="$(mktemp -d)"
chmod 0700 "$snapshot_root"
trap 'rm -rf -- "$snapshot_root"' EXIT
registry_snapshot="$snapshot_root/decoy_asset_registry.json"
bundle_snapshot="$snapshot_root/decoy_asset_registry.json.sigstore.json"
install -m 0400 -- "$registry" "$registry_snapshot"
install -m 0400 -- "$bundle" "$bundle_snapshot"
registry_sha256="$(sha256sum -- "$registry_snapshot" | awk '{print $1}')"
bundle_sha256="$(sha256sum -- "$bundle_snapshot" | awk '{print $1}')"
cosign verify-blob --key /run/trust/decoy-registry.pub \
--bundle "$bundle_snapshot" "$registry_snapshot" >/dev/null
if [[ "$(sha256sum -- "$registry_snapshot" | awk '{print $1}')" != "$registry_sha256" \
|| "$(sha256sum -- "$bundle_snapshot" | awk '{print $1}')" != "$bundle_sha256" ]]; then
echo "decoy registry snapshot changed during verification" >&2
exit 1
fi
printf 'registry_sha256=%s bundle_sha256=%s\n' "$registry_sha256" "$bundle_sha256"
python deception/publish_decoy_model_bundle.py \
--registry "$registry_snapshot" \
--expected-registry-sha256 "$registry_sha256" \
--bundle-dir deception/out/prod_financial_forecast_model \
--receipt security-only/decoy-publication-receipt.json
Retain the exact registry and .json.sigstore.json bundle bytes verified by the independent consumer, bound to the printed registry and bundle digests, in a security-owned account or immutable evidence store. Do not upload either file into the decoy bucket or grant artifact-reader identities s3:GetObjectTagging or registry access.
Step 3: Publish only the plausible bundle
# File: deception/publish_decoy_model_bundle.py
from __future__ import annotations
import argparse
import base64
import hashlib
import json
import math
import os
import time
from datetime import datetime, timezone
from pathlib import Path
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
def unique(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate registry key: {key}")
value[key] = item
return value
def strict_json(raw: bytes):
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique,
parse_constant=lambda value: (_ for _ in ()).throw(ValueError("non-finite JSON: " + value)),
)
def parse_time(value: str) -> datetime:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("retention time lacks a timezone")
return parsed.astimezone(timezone.utc)
def write_once(path: Path, payload: bytes) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
try:
view = memoryview(payload)
while view:
view = view[os.write(descriptor, view):]
os.fsync(descriptor)
finally:
os.close(descriptor)
def verify_exact_version(s3, asset: dict, key: str, row: dict, version_id: str) -> dict:
checksum = base64.b64encode(bytes.fromhex(row["sha256"])).decode("ascii")
readback = s3.get_object(
Bucket=asset["bucket"], Key=key, VersionId=version_id,
ChecksumMode="ENABLED", ExpectedBucketOwner=asset["expected_bucket_owner"],
)
stream = readback["Body"]
try:
exact = stream.read(row["bytes"] + 1)
finally:
stream.close()
if (len(exact) != row["bytes"] or hashlib.sha256(exact).hexdigest() != row["sha256"]
or readback.get("ContentLength") != row["bytes"]
or readback.get("ChecksumSHA256") != checksum
or readback.get("VersionId") != version_id
or readback.get("ServerSideEncryption") != "aws:kms"
or readback.get("SSEKMSKeyId") != asset["kms_key_arn"]):
raise RuntimeError(f"exact-version S3 readback differs: {key}")
retention = s3.get_object_retention(
Bucket=asset["bucket"], Key=key, VersionId=version_id,
ExpectedBucketOwner=asset["expected_bucket_owner"],
).get("Retention", {})
retain_until = retention.get("RetainUntilDate")
required_until = parse_time(asset["retain_until"])
if (retention.get("Mode") != "COMPLIANCE" or not isinstance(retain_until, datetime)
or retain_until.utcoffset() is None
or retain_until.astimezone(timezone.utc) < required_until):
raise RuntimeError(f"S3 compliance retention readback differs: {key}")
return {
"key": key, "version_id": version_id, "sha256": row["sha256"],
"bytes": row["bytes"], "checksum_sha256": checksum,
"kms_key_arn": asset["kms_key_arn"],
"retain_until": retain_until.astimezone(timezone.utc).isoformat(),
}
def publish_or_reconcile(s3, asset: dict, key: str, row: dict, body: bytes) -> dict:
checksum = base64.b64encode(hashlib.sha256(body).digest()).decode("ascii")
for attempt in range(asset["conditional_conflict_attempts"]):
try:
result = s3.put_object(
Bucket=asset["bucket"], Key=key, Body=body, ContentLength=len(body),
ChecksumAlgorithm="SHA256", ChecksumSHA256=checksum, IfNoneMatch="*",
ServerSideEncryption="aws:kms", SSEKMSKeyId=asset["kms_key_arn"],
ObjectLockMode="COMPLIANCE",
ObjectLockRetainUntilDate=parse_time(asset["retain_until"]),
ExpectedBucketOwner=asset["expected_bucket_owner"],
)
except ClientError as exc:
response = exc.response if isinstance(exc.response, dict) else {}
status_code = response.get("ResponseMetadata", {}).get("HTTPStatusCode")
error_code = str(response.get("Error", {}).get("Code", ""))
if status_code == 412 or error_code in {"412", "PreconditionFailed"}:
current = s3.head_object(
Bucket=asset["bucket"], Key=key, ChecksumMode="ENABLED",
ExpectedBucketOwner=asset["expected_bucket_owner"],
)
version_id = current.get("VersionId")
if not isinstance(version_id, str) or not version_id:
raise RuntimeError(f"existing object lacks a version ID: {key}") from exc
evidence = verify_exact_version(s3, asset, key, row, version_id)
evidence["publication_state"] = "RECONCILED"
return evidence
if status_code == 409 or error_code in {"409", "ConditionalRequestConflict"}:
if attempt + 1 == asset["conditional_conflict_attempts"]:
raise RuntimeError(f"conditional publication remained conflicted: {key}") from exc
time.sleep(min(0.05 * (2 ** attempt), 0.5))
continue
raise
version_id = result.get("VersionId")
if not isinstance(version_id, str) or not version_id:
raise RuntimeError(f"S3 did not return a version ID: {key}")
evidence = verify_exact_version(s3, asset, key, row, version_id)
evidence["publication_state"] = "CREATED"
return evidence
raise RuntimeError(f"conditional publication exhausted without a result: {key}")
def reconcile_version_history(s3, asset: dict, expected: dict, published: list[dict]) -> list[dict]:
expected_keys = {asset["key_prefix"] + name: row for name, row in expected.items()}
published_by_key = {row["key"]: row for row in published}
entries, key_marker, version_marker = [], None, None
while True:
remaining = asset["maximum_version_entries"] + 1 - len(entries)
if remaining <= 0:
raise RuntimeError("S3 version history exceeds the signed population bound")
request = {
"Bucket": asset["bucket"], "Prefix": asset["key_prefix"],
"ExpectedBucketOwner": asset["expected_bucket_owner"],
"MaxKeys": min(remaining, 1_000),
}
if key_marker is not None:
request["KeyMarker"] = key_marker
if version_marker is not None:
request["VersionIdMarker"] = version_marker
page = s3.list_object_versions(**request)
page_entries = [("VERSION", row) for row in page.get("Versions", [])]
page_entries.extend(("DELETE_MARKER", row) for row in page.get("DeleteMarkers", []))
if not page_entries and page.get("IsTruncated"):
raise RuntimeError("S3 returned an empty truncated version-history page")
entries.extend(page_entries)
if len(entries) > asset["maximum_version_entries"]:
raise RuntimeError("S3 version history exceeds the signed population bound")
if not page.get("IsTruncated"):
break
key_marker = page.get("NextKeyMarker")
version_marker = page.get("NextVersionIdMarker")
if not isinstance(key_marker, str) or not key_marker:
raise RuntimeError("S3 version-history pagination lacks a key marker")
observed = {key: [] for key in expected_keys}
for kind, entry in entries:
key = entry.get("Key")
if kind != "VERSION" or key not in observed:
raise RuntimeError("unexpected key, delete marker, or object under the dedicated decoy prefix")
observed[key].append(entry)
history = []
for key in sorted(expected_keys):
versions = observed[key]
receipt_row = published_by_key.get(key)
if (len(versions) != 1 or receipt_row is None or versions[0].get("IsLatest") is not True
or versions[0].get("VersionId") != receipt_row["version_id"]
or versions[0].get("Size") != expected_keys[key]["bytes"]):
raise RuntimeError(f"version history is not the exact single-version population: {key}")
history.append({
"key": key, "version_id": receipt_row["version_id"],
"is_latest": True, "bytes": expected_keys[key]["bytes"],
})
return history
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--registry", type=Path, required=True)
parser.add_argument("--expected-registry-sha256", required=True)
parser.add_argument("--bundle-dir", type=Path, required=True)
parser.add_argument("--receipt", type=Path, required=True)
args = parser.parse_args()
registry_raw = args.registry.read_bytes()
registry_sha256 = hashlib.sha256(registry_raw).hexdigest()
if registry_sha256 != args.expected_registry_sha256:
raise RuntimeError("verified registry digest changed before publication")
registry = strict_json(registry_raw)
if not isinstance(registry, dict) or set(registry) != {"schema_version", "registry_version", "assets"}:
raise ValueError("registry schema differs")
if registry["schema_version"] != "aidefend.decoy-asset-registry.v3" or len(registry["assets"]) != 1:
raise ValueError("unsupported registry or ambiguous asset population")
if not isinstance(registry["registry_version"], str) or not registry["registry_version"].strip():
raise ValueError("registry version is invalid")
asset = registry["assets"][0]
required = {
"decoy_id", "bucket", "expected_bucket_owner", "key_prefix", "expected_objects",
"kms_key_arn", "object_lock_mode", "retain_until", "s3_connect_timeout_seconds",
"s3_read_timeout_seconds", "s3_max_attempts", "conditional_conflict_attempts",
"maximum_version_entries", "incident_severity", "alert_slo_seconds",
}
if not isinstance(asset, dict) or set(asset) != required or asset["object_lock_mode"] != "COMPLIANCE":
raise ValueError("registered asset schema differs")
for field in ("decoy_id", "bucket", "key_prefix", "kms_key_arn"):
if not isinstance(asset[field], str) or not asset[field].strip():
raise ValueError(f"registered asset identity is invalid: {field}")
if (not asset["key_prefix"].endswith("/") or asset["key_prefix"].startswith("/")
or "\\" in asset["key_prefix"] or ".." in asset["key_prefix"].split("/")):
raise ValueError("registered key prefix is unsafe")
if (not isinstance(asset["expected_bucket_owner"], str)
or len(asset["expected_bucket_owner"]) != 12 or not asset["expected_bucket_owner"].isdigit()):
raise ValueError("registered bucket owner is invalid")
if not asset["kms_key_arn"].startswith("arn:aws:kms:"):
raise ValueError("registered KMS key is not an ARN")
if asset["incident_severity"] not in {"LOW", "MEDIUM", "HIGH", "CRITICAL"}:
raise ValueError("registered incident severity is invalid")
retention = parse_time(asset["retain_until"])
if retention <= datetime.now(timezone.utc):
raise ValueError("signed object retention has already expired")
timeout_values = (asset["s3_connect_timeout_seconds"], asset["s3_read_timeout_seconds"])
if (any(type(value) not in (int, float) or not math.isfinite(float(value)) or value <= 0
for value in timeout_values) or float(timeout_values[0]) > 60 or float(timeout_values[1]) > 300):
raise ValueError("signed S3 timeouts are invalid")
if type(asset["s3_max_attempts"]) is not int or not 1 <= asset["s3_max_attempts"] <= 10:
raise ValueError("signed S3 retry bound is invalid")
if (type(asset["conditional_conflict_attempts"]) is not int
or not 1 <= asset["conditional_conflict_attempts"] <= 10):
raise ValueError("signed conditional-conflict retry bound is invalid")
if (type(asset["maximum_version_entries"]) is not int
or not 1 <= asset["maximum_version_entries"] <= 1_000):
raise ValueError("signed version-history bound is invalid")
if type(asset["alert_slo_seconds"]) is not int or not 1 <= asset["alert_slo_seconds"] <= 86_400:
raise ValueError("signed alert SLO is invalid")
if (not isinstance(asset["expected_objects"], list) or not asset["expected_objects"]
or len(asset["expected_objects"]) > asset["maximum_version_entries"]):
raise ValueError("signed object population is empty, invalid, or exceeds its history bound")
expected = {}
for row in asset["expected_objects"]:
if not isinstance(row, dict) or set(row) != {"name", "sha256", "bytes"}:
raise ValueError("registered object schema differs")
name = row["name"]
if (not isinstance(name, str) or not name or name in {".", ".."}
or "/" in name or "\\" in name or any(ord(character) < 32 for character in name)
or name in expected or type(row["bytes"]) is not int or row["bytes"] < 1
or not isinstance(row["sha256"], str) or len(row["sha256"]) != 64
or set(row["sha256"]) - set("0123456789abcdef")):
raise ValueError("registered object identity is invalid")
expected[name] = row
local_names = sorted(path.name for path in args.bundle_dir.iterdir())
if local_names != sorted(expected):
raise RuntimeError("local decoy population differs from the signed registry")
s3 = boto3.client("s3", config=Config(
connect_timeout=float(asset["s3_connect_timeout_seconds"]),
read_timeout=float(asset["s3_read_timeout_seconds"]),
retries={"total_max_attempts": asset["s3_max_attempts"], "mode": "standard"},
))
owner = asset["expected_bucket_owner"]
if s3.get_bucket_versioning(Bucket=asset["bucket"], ExpectedBucketOwner=owner).get("Status") != "Enabled":
raise RuntimeError("S3 versioning is not enabled")
if s3.get_object_lock_configuration(Bucket=asset["bucket"], ExpectedBucketOwner=owner).get("ObjectLockConfiguration", {}).get("ObjectLockEnabled") != "Enabled":
raise RuntimeError("S3 Object Lock is not enabled")
published = []
for name in sorted(expected):
path = args.bundle_dir / name
if path.is_symlink() or not path.is_file():
raise RuntimeError("local object is not a regular non-symlink file")
body = path.read_bytes()
row = expected[name]
sha = hashlib.sha256(body).hexdigest()
if len(body) != row["bytes"] or sha != row["sha256"]:
raise RuntimeError("local object differs from the signed registry")
key = asset["key_prefix"] + name
published.append(publish_or_reconcile(s3, asset, key, row, body))
version_history = reconcile_version_history(s3, asset, expected, published)
receipt = {
"schema_version": "aidefend.decoy-publication.v2",
"registry_sha256": registry_sha256,
"registry_version": registry["registry_version"],
"bucket": asset["bucket"],
"published": published,
"version_history": version_history,
"verified_at": datetime.now(timezone.utc).isoformat(),
}
write_once(args.receipt, (json.dumps(receipt, sort_keys=True, separators=(",", ":")) + "\n").encode())
if __name__ == "__main__":
main()
Step 4: Alert from the private deployment configuration
# File: deception/alert_on_decoy_model_access.py
from __future__ import annotations
import hashlib
import json
import os
import stat
import subprocess
from pathlib import Path
REGISTRY_PATH = Path("/opt/aidefend/contracts/decoy_asset_registry.json")
BUNDLE_PATH = Path("/opt/aidefend/contracts/decoy_asset_registry.json.sigstore.json")
PUBLIC_KEY_PATH = Path("/opt/aidefend/trust/decoy-registry.pub")
COSIGN_PATH = Path("/opt/aidefend/bin/cosign") # Lambda layer pins Cosign v3.1.2.
BOOTSTRAP_ARTIFACT_CEILING = 1_048_576
COSIGN_DEADLINE_SECONDS = 10
ASSET_FIELDS = {
"decoy_id", "bucket", "expected_bucket_owner", "key_prefix", "expected_objects",
"kms_key_arn", "object_lock_mode", "retain_until", "s3_connect_timeout_seconds",
"s3_read_timeout_seconds", "s3_max_attempts", "conditional_conflict_attempts",
"maximum_version_entries", "incident_severity", "alert_slo_seconds",
}
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate registry key: {key}")
result[key] = value
return result
def strict_json(raw: bytes):
return json.loads(
raw.decode("utf-8", errors="strict"), object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(ValueError("non-finite JSON: " + value)),
)
def read_privileged(path: Path) -> bytes:
if not hasattr(os, "O_NOFOLLOW"):
raise OSError("secure no-follow reads are unavailable")
descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
try:
before = os.fstat(descriptor)
if (not stat.S_ISREG(before.st_mode) or before.st_uid != 0 or before.st_mode & 0o022
or not 1 <= before.st_size <= BOOTSTRAP_ARTIFACT_CEILING):
raise PermissionError(f"untrusted Lambda-layer artifact: {path}")
chunks, observed = [], 0
while observed <= BOOTSTRAP_ARTIFACT_CEILING:
chunk = os.read(descriptor, min(65_536, BOOTSTRAP_ARTIFACT_CEILING + 1 - observed))
if not chunk:
break
chunks.append(chunk)
observed += len(chunk)
after = os.fstat(descriptor)
identity = lambda item: (item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns)
if observed != before.st_size or observed > BOOTSTRAP_ARTIFACT_CEILING or identity(before) != identity(after):
raise OSError(f"Lambda-layer artifact changed during read: {path}")
return b"".join(chunks)
finally:
os.close(descriptor)
def validate_cosign() -> None:
metadata = os.lstat(COSIGN_PATH)
if (not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode)
or metadata.st_uid != 0 or metadata.st_mode & 0o022 or not os.access(COSIGN_PATH, os.X_OK)):
raise PermissionError("untrusted Cosign v3.1.2 executable")
def load_alert_contract() -> dict:
registry_raw = read_privileged(REGISTRY_PATH)
bundle_raw = read_privileged(BUNDLE_PATH)
key_raw = read_privileged(PUBLIC_KEY_PATH)
snapshots = tuple(hashlib.sha256(value).hexdigest() for value in (registry_raw, bundle_raw, key_raw))
validate_cosign()
subprocess.run(
[str(COSIGN_PATH), "verify-blob", "--key", str(PUBLIC_KEY_PATH),
"--bundle", str(BUNDLE_PATH), str(REGISTRY_PATH)],
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
check=True, timeout=COSIGN_DEADLINE_SECONDS,
)
reread = tuple(hashlib.sha256(read_privileged(path)).hexdigest()
for path in (REGISTRY_PATH, BUNDLE_PATH, PUBLIC_KEY_PATH))
if snapshots != reread:
raise RuntimeError("registry, bundle, or trust key changed during Cosign verification")
registry = strict_json(registry_raw)
if (not isinstance(registry, dict) or set(registry) != {"schema_version", "registry_version", "assets"}
or registry.get("schema_version") != "aidefend.decoy-asset-registry.v3"
or not isinstance(registry.get("registry_version"), str)
or not registry["registry_version"].strip()
or not isinstance(registry.get("assets"), list) or len(registry["assets"]) != 1):
raise RuntimeError("signed decoy registry schema differs")
asset = registry["assets"][0]
if not isinstance(asset, dict) or set(asset) != ASSET_FIELDS:
raise RuntimeError("signed decoy asset schema differs")
for field in ("decoy_id", "bucket", "key_prefix"):
if not isinstance(asset[field], str) or not asset[field].strip():
raise RuntimeError(f"signed alert identity is invalid: {field}")
if (not asset["key_prefix"].endswith("/") or asset["key_prefix"].startswith("/")
or "\\" in asset["key_prefix"] or ".." in asset["key_prefix"].split("/")):
raise RuntimeError("signed alert key prefix is unsafe")
owner = asset["expected_bucket_owner"]
if not isinstance(owner, str) or len(owner) != 12 or not owner.isdigit():
raise RuntimeError("signed bucket owner is invalid")
if asset["incident_severity"] not in {"LOW", "MEDIUM", "HIGH", "CRITICAL"}:
raise RuntimeError("signed alert severity is invalid")
if type(asset["alert_slo_seconds"]) is not int or not 1 <= asset["alert_slo_seconds"] <= 86_400:
raise RuntimeError("signed alert SLO is invalid")
if (not isinstance(asset["expected_objects"], list) or not asset["expected_objects"]
or type(asset["maximum_version_entries"]) is not int
or len(asset["expected_objects"]) > asset["maximum_version_entries"]):
raise RuntimeError("signed alert population is invalid")
exact_keys = set()
for row in asset["expected_objects"]:
if not isinstance(row, dict) or set(row) != {"name", "sha256", "bytes"}:
raise RuntimeError("signed alert object schema differs")
name = row["name"]
if (not isinstance(name, str) or not name or name in {".", ".."}
or "/" in name or "\\" in name or any(ord(character) < 32 for character in name)
or asset["key_prefix"] + name in exact_keys or type(row["bytes"]) is not int
or row["bytes"] < 1 or not isinstance(row["sha256"], str)
or len(row["sha256"]) != 64 or set(row["sha256"]) - set("0123456789abcdef")):
raise RuntimeError("signed alert object name is invalid")
exact_keys.add(asset["key_prefix"] + name)
return {
"registry_version": registry["registry_version"],
"registry_sha256": snapshots[0], "bundle_sha256": snapshots[1],
"public_key_sha256": snapshots[2], "decoy_id": asset["decoy_id"],
"bucket": asset["bucket"], "expected_bucket_owner": owner,
"exact_keys": frozenset(exact_keys), "severity": asset["incident_severity"],
"alert_slo_seconds": asset["alert_slo_seconds"],
}
CONTRACT = load_alert_contract() # Cold start fails before any event can be acknowledged.
def bounded_text(value: object, field: str, maximum_bytes: int = 2_048) -> str:
if (not isinstance(value, str) or not value or len(value.encode("utf-8")) > maximum_bytes
or any(ord(character) < 32 for character in value)):
raise RuntimeError(f"malformed CloudTrail field: {field}")
return value
def lambda_handler(event, context):
if not isinstance(event, dict) or not isinstance(event.get("detail"), dict):
raise RuntimeError("malformed EventBridge CloudTrail envelope")
detail = event["detail"]
if detail.get("eventSource") != "s3.amazonaws.com" or detail.get("eventName") != "GetObject":
return {"statusCode": 200, "ignored": True}
params = detail.get("requestParameters")
if not isinstance(params, dict):
raise RuntimeError("matching CloudTrail event lacks request parameters")
bucket = bounded_text(params.get("bucketName"), "bucketName", 63)
key = bounded_text(params.get("key"), "key", 1_024)
if bucket != CONTRACT["bucket"] or key not in CONTRACT["exact_keys"]:
return {"statusCode": 200, "ignored": True}
if (detail.get("eventCategory") != "Data" or detail.get("readOnly") is not True
or detail.get("recipientAccountId") != CONTRACT["expected_bucket_owner"]):
raise RuntimeError("matching S3 event is not the expected read-only data event")
identity = detail.get("userIdentity")
if not isinstance(identity, dict):
raise RuntimeError("matching S3 event lacks an authenticated identity")
alert = {
"schema_version": "aidefend.decoy-model-access.v2",
"severity": CONTRACT["severity"],
"event_type": "decoy_model_access",
"decoy_id": CONTRACT["decoy_id"],
"registry_version": CONTRACT["registry_version"],
"registry_sha256": CONTRACT["registry_sha256"],
"cloudtrail_event_id": bounded_text(detail.get("eventID"), "eventID", 256),
"cloudtrail_event_time": bounded_text(detail.get("eventTime"), "eventTime", 64),
"bucket": bucket, "key": key,
"version_id": params.get("versionId"),
"actor": bounded_text(identity.get("arn"), "userIdentity.arn"),
"source_ip": bounded_text(detail.get("sourceIPAddress"), "sourceIPAddress", 256),
"alert_slo_seconds": CONTRACT["alert_slo_seconds"],
}
if alert["version_id"] is not None:
alert["version_id"] = bounded_text(alert["version_id"], "versionId", 1_024)
print(json.dumps(alert, sort_keys=True))
return {"statusCode": 200, "alerted": True, "cloudtrail_event_id": alert["cloudtrail_event_id"]}
Build the alert Lambda layer with the exact registry, Sigstore bundle, public key, and a root-owned non-writable Cosign v3.1.2 binary at the fixed paths above. Cold start verifies those exact bytes and derives bucket, account, exact keys, decoy ID, severity, and SLO exclusively from the signed registry; do not mirror them into environment variables. Restrict the event rule to CloudTrail S3 data events and grant the publisher only the registered bucket/prefix operations, including s3:ListBucketVersions for the final exact-population reconciliation. This handler intentionally accepts only GetObject; add CopyObject only with a separate fixture-tested parser for its source-object fields.
Verification: Reserve the signed key prefix for this one registry population, enable CloudTrail S3 data events for those exact keys, and route matching events to the alert function. A dedicated drill identity reads each expected object. Require the corresponding CloudTrail event IDs and independently read-back SIEM alert IDs within the signed SLO, while confirming that the downloaded bundle, object metadata visible to that identity, and list results contain no deception labels. Exercise fresh publication, interruption after the first key, safe resume, concurrent 409, exact-existing 412, altered existing bytes/KMS/retention, extra version, delete marker, and unexpected-prefix-object fixtures.
Action: Keep trap classification exclusively in the signed private registry and security deployment configuration. Any artifact-visible decoy/honeypot marker is a failed deception review, while a missing end-to-end read alert is ERROR, not a pass.
The producer signs the computed object population, while the independent deploy job publishes only from that verified snapshot. Each key is conditionally create-once, versioned, compliance-retained, checksum/KMS-bound, and read back by its exact VersionId. A 412 is accepted only after exact current-version reconciliation; the final receipt also binds the bounded complete version history and is forwarded to the security-owned append-only evidence store before advertising the lure.
AID-DV-002-G003
Create and embed decoy API keys/access tokens (Canary Tokens).
Concept:
A credential canary must have an observable alert path. A self-invented string that only looks like an AWS access key is not enough: if the key does not exist anywhere, a failed validation attempt is not reliably visible to the defender. Use either a provider-issued AWS API key Canarytoken or a real, zero-permission IAM user whose access key is watched by CloudTrail and EventBridge.
Option A: Use a provider-issued AWS API key Canarytoken
Create an AWS API Key Canarytoken in the Canarytokens/Thinkst console or API, copy the issued credential pair, and place it where stolen credentials would plausibly appear. Do not edit the issued key material. The value is useful because the provider controls the backing AWS-side alert path.
#!/usr/bin/env bash
# File: decoys/archive-model-registry.env
# Provider-issued canary credential. Never route this to production code.
ARCHIVE_AWS_ACCESS_KEY_ID="AKIA_PROVIDER_ISSUED_CANARY"
ARCHIVE_AWS_SECRET_ACCESS_KEY="provider-issued-secret-value"
ARCHIVE_AWS_CANARY_MEMO="dropped in decoy archive model registry config"
Option B: Self-managed zero-permission IAM canary
If you cannot use a canary provider, create a dedicated IAM user in a security-owned account with no business permissions. Alert on any CloudTrail event where that access key appears. A successful sts:GetCallerIdentity or an AccessDenied event both mean the credential was used.
data "aws_caller_identity" "current" {}
# File: infrastructure/aws_access_key_canary.tf
resource "aws_iam_user" "archive_key_canary" {
name = "aidefend-archive-key-canary"
path = "/canary/"
tags = {
aidefend_control = "AID-DV-002"
purpose = "credential-canary"
}
}
# Do not attach allow policies. Store the secret value once in a security-owned
# secret store or canary registry, then seed only the intended decoy locations.
resource "aws_iam_access_key" "archive_key_canary" {
user = aws_iam_user.archive_key_canary.name
}
resource "aws_sns_topic" "credential_canary_alerts" {
name = "aidefend-credential-canary-alerts"
}
# This queue is the confirmed delivery/readback endpoint. A security-owned
# consumer may forward from it to the SIEM, but the queue remains available for
# an independent end-to-end drill readback.
resource "aws_sqs_queue" "credential_canary_readback" {
name = "aidefend-credential-canary-readback"
sqs_managed_sse_enabled = true
}
resource "aws_sqs_queue_policy" "allow_canary_topic" {
queue_url = aws_sqs_queue.credential_canary_readback.url
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Sid = "AllowCanaryTopic"
Effect = "Allow"
Principal = { Service = "sns.amazonaws.com" }
Action = "sqs:SendMessage"
Resource = aws_sqs_queue.credential_canary_readback.arn
Condition = {
StringEquals = { "aws:SourceAccount" = data.aws_caller_identity.current.account_id }
ArnEquals = { "aws:SourceArn" = aws_sns_topic.credential_canary_alerts.arn }
}
}
]
})
}
resource "aws_sns_topic_subscription" "credential_canary_readback" {
topic_arn = aws_sns_topic.credential_canary_alerts.arn
protocol = "sqs"
endpoint = aws_sqs_queue.credential_canary_readback.arn
raw_message_delivery = true
depends_on = [aws_sqs_queue_policy.allow_canary_topic]
}
resource "aws_cloudwatch_event_rule" "canary_access_key_used" {
name = "aidefend-canary-access-key-used"
description = "Alert whenever the canary access key calls STS GetCallerIdentity"
state = "ENABLED_WITH_ALL_CLOUDTRAIL_MANAGEMENT_EVENTS"
event_pattern = jsonencode({
source = ["aws.sts"]
"detail-type" = ["AWS API Call via CloudTrail"]
detail = {
eventSource = ["sts.amazonaws.com"]
eventName = ["GetCallerIdentity"]
userIdentity = {
accessKeyId = [aws_iam_access_key.archive_key_canary.id]
}
}
})
}
resource "aws_sns_topic_policy" "allow_eventbridge_publish" {
arn = aws_sns_topic.credential_canary_alerts.arn
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Sid = "AllowEventBridgePublish"
Effect = "Allow"
Principal = {
Service = "events.amazonaws.com"
}
Action = "sns:Publish"
Resource = aws_sns_topic.credential_canary_alerts.arn
Condition = {
StringEquals = { "AWS:SourceAccount" = data.aws_caller_identity.current.account_id }
ArnEquals = { "aws:SourceArn" = aws_cloudwatch_event_rule.canary_access_key_used.arn }
}
}
]
})
}
resource "aws_cloudwatch_event_target" "canary_access_key_sns" {
rule = aws_cloudwatch_event_rule.canary_access_key_used.name
target_id = "send-to-confirmed-readback"
arn = aws_sns_topic.credential_canary_alerts.arn
depends_on = [
aws_sns_topic_policy.allow_eventbridge_publish,
aws_sns_topic_subscription.credential_canary_readback,
]
}
output "credential_canary_readback_queue_url" {
value = aws_sqs_queue.credential_canary_readback.url
}Step 3: Prove the alert path with a synthetic trigger
Run a controlled drill after deployment and after every logging/alerting change. The drill should use the seeded key from a security test workstation, verify that EventBridge or the canary provider fires within the expected SLO, and record the CloudTrail event ID, alert ID, key ID, source IP, and decoy placement memo.
#!/usr/bin/env bash
set -euo pipefail
umask 077
required=(
CANARY_AWS_ACCESS_KEY_ID
CANARY_AWS_SECRET_ACCESS_KEY
CANARY_AWS_REGION
CANARY_EXPECTED_ACCOUNT_ID
CANARY_EXPECTED_ARN
CANARY_DRILL_ID
CANARY_DRILL_CONNECT_TIMEOUT_SECONDS
CANARY_DRILL_READ_TIMEOUT_SECONDS
CANARY_AWS_MAX_ATTEMPTS
)
for name in "${required[@]}"; do
if [[ -z "${!name:-}" || "${!name}" =~ [[:cntrl:]] ]]; then
echo "missing or invalid drill input: $name" >&2
exit 2
fi
done
if [[ ! "$CANARY_AWS_ACCESS_KEY_ID" =~ ^[A-Z0-9]{16,128}$ ]]; then
echo "canary access-key ID has an invalid shape" >&2
exit 2
fi
if [[ ! "$CANARY_AWS_REGION" =~ ^[a-z0-9][a-z0-9-]{0,62}$ ]]; then
echo "canary region has an invalid shape" >&2
exit 2
fi
if [[ ! "$CANARY_EXPECTED_ACCOUNT_ID" =~ ^[0-9]{12}$ ]]; then
echo "expected AWS account ID must contain exactly 12 digits" >&2
exit 2
fi
if [[ ! "$CANARY_DRILL_ID" =~ ^[a-zA-Z0-9][a-zA-Z0-9._:-]{0,127}$ ]]; then
echo "drill ID has an invalid shape" >&2
exit 2
fi
for name in CANARY_DRILL_CONNECT_TIMEOUT_SECONDS CANARY_DRILL_READ_TIMEOUT_SECONDS; do
value="${!name}"
if [[ ! "$value" =~ ^[0-9]{1,3}$ ]] || (( 10#$value < 1 || 10#$value > 300 )); then
echo "$name must be an integer in 1..300 from the signed drill policy" >&2
exit 2
fi
done
if [[ ! "$CANARY_AWS_MAX_ATTEMPTS" =~ ^[0-9]{1,2}$ ]] \
|| (( 10#$CANARY_AWS_MAX_ATTEMPTS < 1 || 10#$CANARY_AWS_MAX_ATTEMPTS > 10 )); then
echo "CANARY_AWS_MAX_ATTEMPTS must be an integer in 1..10 from the signed drill policy" >&2
exit 2
fi
aws_binary=/usr/local/bin/aws
if [[ ! -f "$aws_binary" || -L "$aws_binary" || ! -x "$aws_binary" ]]; then
echo "fixed AWS CLI v2 binary is absent, linked, or not executable" >&2
exit 2
fi
aws_owner="$(stat -Lc '%u' -- "$aws_binary")"
aws_mode="$(stat -Lc '%a' -- "$aws_binary")"
if (( 10#$aws_owner != 0 || (8#$aws_mode & 022) != 0 )); then
echo "fixed AWS CLI v2 binary is not root-controlled" >&2
exit 2
fi
aws_version="$($aws_binary --version 2>&1)"
if [[ ! "$aws_version" =~ ^aws-cli/2\. ]]; then
echo "the protected drill image must contain AWS CLI v2" >&2
exit 2
fi
scratch="$(mktemp -d)"
chmod 0700 "$scratch"
mkdir -m 0700 "$scratch/home"
: >"$scratch/config"
: >"$scratch/credentials"
chmod 0400 "$scratch/config" "$scratch/credentials"
cleanup() {
unset CANARY_AWS_ACCESS_KEY_ID CANARY_AWS_SECRET_ACCESS_KEY AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN
rm -rf -- "$scratch"
}
trap cleanup EXIT
started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
set +e
env -i \
PATH=/usr/local/bin:/usr/bin:/bin \
HOME="$scratch/home" \
LC_ALL=C \
AWS_ACCESS_KEY_ID="$CANARY_AWS_ACCESS_KEY_ID" \
AWS_SECRET_ACCESS_KEY="$CANARY_AWS_SECRET_ACCESS_KEY" \
AWS_SESSION_TOKEN= \
AWS_REGION="$CANARY_AWS_REGION" \
AWS_DEFAULT_REGION="$CANARY_AWS_REGION" \
AWS_CONFIG_FILE="$scratch/config" \
AWS_SHARED_CREDENTIALS_FILE="$scratch/credentials" \
AWS_EC2_METADATA_DISABLED=true \
AWS_MAX_ATTEMPTS="$CANARY_AWS_MAX_ATTEMPTS" \
AWS_RETRY_MODE=standard \
AWS_CLI_AUTO_PROMPT=off \
AWS_PAGER= \
HTTPS_PROXY= HTTP_PROXY= ALL_PROXY= NO_PROXY='*' \
"$aws_binary" sts get-caller-identity \
--region "$CANARY_AWS_REGION" \
--query '[Account,Arn]' \
--output text \
--cli-connect-timeout "$CANARY_DRILL_CONNECT_TIMEOUT_SECONDS" \
--cli-read-timeout "$CANARY_DRILL_READ_TIMEOUT_SECONDS" \
--no-cli-pager \
>"$scratch/stdout" 2>"$scratch/stderr"
aws_status=$?
set -e
completed_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
stdout_sha256="$(sha256sum -- "$scratch/stdout" | awk '{print $1}')"
stderr_sha256="$(sha256sum -- "$scratch/stderr" | awk '{print $1}')"
aws_binary_sha256="$(sha256sum -- "$aws_binary" | awk '{print $1}')"
case "$aws_status" in
0)
observed_account=""
observed_arn=""
unexpected=""
if ! read -r observed_account observed_arn unexpected <"$scratch/stdout"; then
printf 'status=ERROR\nreason=caller_identity_unreadable\ndrill_id=%s\naws_cli_exit_code=%s\nstdout_sha256=%s\nstderr_sha256=%s\n' \
"$CANARY_DRILL_ID" "$aws_status" "$stdout_sha256" "$stderr_sha256" >&2
exit 2
fi
if [[ "$observed_account" != "$CANARY_EXPECTED_ACCOUNT_ID" \
|| "$observed_arn" != "$CANARY_EXPECTED_ARN" \
|| -n "$unexpected" ]]; then
printf 'status=ERROR\nreason=caller_identity_mismatch\ndrill_id=%s\naws_cli_exit_code=%s\nstdout_sha256=%s\nstderr_sha256=%s\n' \
"$CANARY_DRILL_ID" "$aws_status" "$stdout_sha256" "$stderr_sha256" >&2
exit 2
fi
;;
254)
# AWS CLI v2 documents 254 as a service-returned error. This is only a
# trigger attempt; the independent CloudTrail/SQS replay must prove receipt.
;;
*)
printf 'status=ERROR\nreason=aws_cli_operational_failure\ndrill_id=%s\naws_cli_exit_code=%s\nstdout_sha256=%s\nstderr_sha256=%s\n' \
"$CANARY_DRILL_ID" "$aws_status" "$stdout_sha256" "$stderr_sha256" >&2
exit 2
;;
esac
printf 'status=TRIGGER_ATTEMPT\ndrill_id=%s\naccess_key_id=%s\nstarted_at=%s\ncompleted_at=%s\naws_cli_exit_code=%s\naws_cli_v2_sha256=%s\nstdout_sha256=%s\nstderr_sha256=%s\n' \
"$CANARY_DRILL_ID" \
"$CANARY_AWS_ACCESS_KEY_ID" \
"$started_at" \
"$completed_at" \
"$aws_status" \
"$aws_binary_sha256" \
"$stdout_sha256" \
"$stderr_sha256"Runtime exclusion check
#!/usr/bin/env bash
set -euo pipefail
umask 077
roots=()
for candidate in deploy src infra; do
[[ -e "$candidate" ]] && roots+=("$candidate")
done
if (( ${#roots[@]} == 0 )); then
echo "none of the declared runtime-source roots exists" >&2
exit 2
fi
report="$(mktemp)"
trap 'rm -f -- "$report"' EXIT
set +e
rg -n --hidden --glob '!**/.git/**' \
'ARCHIVE_AWS_ACCESS_KEY_ID|ARCHIVE_AWS_SECRET_ACCESS_KEY' \
"${roots[@]}" >"$report"
rg_status=$?
set -e
case "$rg_status" in
0)
echo "canary credential names leaked into a runtime-source root:" >&2
cat "$report" >&2
exit 1
;;
1)
echo "PASS: no canary credential names in declared runtime-source roots"
;;
*)
echo "ERROR: runtime exclusion scan failed with rg status $rg_status" >&2
exit 2
;;
esacAction: Place only trackable credential canaries. Treat any canary key usage as incident-grade evidence, and fail the control if the synthetic drill cannot prove an end-to-end alert.
Keep an active CloudTrail trail that records management events in every governed account and region. The rule state shown above is required because sts:GetCallerIdentity is read-only; after each controlled drill, retrieve the matching CloudTrail event ID and SNS delivery record and fail the drill if either is absent.
Terraform state contains the canary access-key secret and therefore belongs in encrypted, access-logged, security-owned state storage. The topic has a confirmed security-owned SQS subscription so the drill has a concrete delivery/readback endpoint. Run the trigger from an isolated workstation; the script validates a fixed AWS CLI v2 binary, clears ambient profiles, endpoints, CA overrides, and proxies with env -i, and emits only TRIGGER_ATTEMPT or bounded ERROR evidence. An independent security identity must locate the exact CloudTrail event by access-key ID and drill time, read the matching raw SQS event, compare its event ID, event source, event name, and access-key ID, and retain the queue message ID plus the SIEM ingestion receipt when a SIEM consumer is configured. A trigger with a missing CloudTrail or SQS delivery after the signed drill window is FAIL; a query or service error is ERROR; incomplete retained evidence is INSUFFICIENT_DATA; only exact event and delivery readback is PASS. Scan the built runtime image, rendered configuration, and mounted secret population as well as source.
AID-DV-002-G004
Embed trackable URLs / web bugs in fake sensitive documents.
Limits and safe use
Use a provider-issued, unique HTTPS canary URL in a synthetic document whose content contains no real confidential data. A remote fetch is a corroborating access signal, not proof of the original reader: mail gateways, document scanners, privacy proxies, caches, or disabled remote-content settings can create false hits or suppress the request. Keep a signed private registry that binds document digest, canary ID, intended placement, owner, expected scanners, and alert route.
Build the document from the issued URL
{
"schema_version": "aidefend.dv002.url-canary-registry.v1",
"registry_version": "2026.07.20.1",
"entries": [
{
"canary_id": "url-canary-acquisition-7f2d",
"placement_id": "synthetic-acquisition-scenario-v1",
"provider": "approved-canary-provider",
"url": "https://canary.example.invalid/t/7f2d9a",
"allowed_hostname": "canary.example.invalid",
"document_filename": "2026-acquisition-scenario.md",
"not_after": "2026-08-20T00:00:00Z",
"expected_document_sha256": "ce156383bbcb101d63a2e09161da0a93e37d438dc498309269ae293ecc562418",
"owner_id": "security-deception-team",
"expected_scanner_ids": [
"approved-email-gateway",
"approved-web-proxy"
],
"alert_route_id": "soc-url-canary-sqs"
}
],
"builder_sha256": "961dd2f3c010074f1ff654736af2a4243fb8e11e979e705e29b785efc7a48553"
}# File: deception/build_url_canary_document.py
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import re
import stat
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit
REGISTRY_MAXIMUM_BYTES = 1_048_576
BUILDER_MAXIMUM_BYTES = 262_144
DOCUMENT_MAXIMUM_BYTES = 1_048_576
RECEIPT_MAXIMUM_BYTES = 1_048_576
OUTPUT_ROOT = Path("deception/out/url-canary-documents")
RECEIPT_ROOT = Path("security-only/url-canary-build-receipts")
HEX_SHA256 = re.compile(r"[0-9a-f]{64}")
IDENTIFIER = re.compile(r"[a-z0-9][a-z0-9._-]{0,127}")
HOSTNAME = re.compile(
r"(?=.{1,253}\Z)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*"
r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?"
)
REGISTRY_FIELDS = {"schema_version", "registry_version", "builder_sha256", "entries"}
ENTRY_FIELDS = {
"canary_id",
"placement_id",
"provider",
"url",
"allowed_hostname",
"document_filename",
"not_after",
"expected_document_sha256",
"owner_id",
"expected_scanner_ids",
"alert_route_id",
}
def canonical(value: object) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def reject_duplicates(pairs: list[tuple[str, object]]) -> dict:
value: dict = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate registry key: {key}")
value[key] = item
return value
def reject_nonfinite(value: str) -> None:
raise ValueError(f"non-finite registry value: {value}")
def read_stable(path: Path, maximum_bytes: int) -> bytes:
flags = (
os.O_RDONLY
| getattr(os, "O_CLOEXEC", 0)
| getattr(os, "O_BINARY", 0)
)
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode) or not 1 <= before.st_size <= maximum_bytes:
raise ValueError("artifact must be a bounded regular file")
chunks: list[bytes] = []
observed = 0
while True:
chunk = os.read(descriptor, min(65_536, maximum_bytes + 1 - observed))
if not chunk:
break
chunks.append(chunk)
observed += len(chunk)
if observed > maximum_bytes:
raise ValueError("registry exceeds the bootstrap byte ceiling")
after = os.fstat(descriptor)
if (
observed != before.st_size
or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
!= (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
):
raise RuntimeError("registry changed while being read")
return b"".join(chunks)
finally:
os.close(descriptor)
def parse_time(value: object) -> datetime:
if not isinstance(value, str) or not value:
raise ValueError("not_after must be a nonempty timestamp")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("not_after must include a timezone")
return parsed.astimezone(timezone.utc)
def validated_url(entry: dict) -> str:
raw = entry["url"]
hostname = entry["allowed_hostname"]
if (
not isinstance(raw, str)
or not 1 <= len(raw.encode("utf-8")) <= 2048
or any(ord(character) <= 0x20 or character in "<>\\" for character in raw)
):
raise ValueError("canary URL contains an unsafe character or length")
if not isinstance(hostname, str) or not HOSTNAME.fullmatch(hostname):
raise ValueError("allowed_hostname is not canonical ASCII DNS")
parsed = urlsplit(raw)
try:
port = parsed.port
except ValueError as exc:
raise ValueError("canary URL port is invalid") from exc
if (
parsed.scheme != "https"
or parsed.hostname != hostname
or parsed.netloc not in {hostname, f"{hostname}:443"}
or port not in {None, 443}
or parsed.username is not None
or parsed.password is not None
or parsed.fragment
or not parsed.path.startswith("/")
):
raise ValueError("canary URL is outside the signed canonical HTTPS origin")
return raw
def load_registry(path: Path, expected_sha256: str) -> tuple[dict, str]:
if not HEX_SHA256.fullmatch(expected_sha256):
raise ValueError("expected registry digest is not lower-case SHA-256")
raw = read_stable(path, REGISTRY_MAXIMUM_BYTES)
digest = hashlib.sha256(raw).hexdigest()
if digest != expected_sha256:
raise RuntimeError("registry digest differs from the verified snapshot")
registry = json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicates,
parse_constant=reject_nonfinite,
)
if not isinstance(registry, dict) or set(registry) != REGISTRY_FIELDS:
raise ValueError("URL-canary registry root schema differs")
if registry["schema_version"] != "aidefend.dv002.url-canary-registry.v1":
raise ValueError("unsupported URL-canary registry schema")
if not isinstance(registry["builder_sha256"], str) or not HEX_SHA256.fullmatch(registry["builder_sha256"]):
raise ValueError("builder_sha256 must be lower-case SHA-256")
if not isinstance(registry["registry_version"], str) or not registry["registry_version"]:
raise ValueError("registry_version must be nonempty")
entries = registry["entries"]
if not isinstance(entries, list) or not entries:
raise ValueError("URL-canary registry population is empty")
seen_ids: set[str] = set()
seen_placements: set[str] = set()
seen_documents: set[str] = set()
seen_urls: set[str] = set()
now = datetime.now(timezone.utc)
for entry in entries:
if not isinstance(entry, dict) or set(entry) != ENTRY_FIELDS:
raise ValueError("URL-canary entry schema differs")
for field in ("canary_id", "placement_id", "owner_id", "alert_route_id"):
if not isinstance(entry[field], str) or not IDENTIFIER.fullmatch(entry[field]):
raise ValueError(f"invalid URL-canary {field}")
if (
not isinstance(entry["provider"], str)
or not entry["provider"].strip()
or len(entry["provider"].encode("utf-8")) > 128
):
raise ValueError("provider must be a bounded nonempty string")
scanners = entry["expected_scanner_ids"]
if (
not isinstance(scanners, list)
or not 1 <= len(scanners) <= 64
or any(not isinstance(item, str) or not IDENTIFIER.fullmatch(item) for item in scanners)
or len(set(scanners)) != len(scanners)
):
raise ValueError("expected_scanner_ids must be a bounded unique identifier list")
filename = entry["document_filename"]
if (
not isinstance(filename, str)
or not IDENTIFIER.fullmatch(filename)
or not filename.endswith(".md")
):
raise ValueError("document filename is not a safe Markdown basename")
url = validated_url(entry)
if url in seen_urls:
raise ValueError("duplicate canary URL")
expected_document_sha256 = entry["expected_document_sha256"]
if not isinstance(expected_document_sha256, str) or not HEX_SHA256.fullmatch(expected_document_sha256):
raise ValueError("expected_document_sha256 must be lower-case SHA-256")
if parse_time(entry["not_after"]) <= now:
raise ValueError("URL-canary registry entry has expired")
if entry["canary_id"] in seen_ids:
raise ValueError("duplicate canary_id")
if entry["placement_id"] in seen_placements:
raise ValueError("duplicate placement_id")
if filename in seen_documents:
raise ValueError("duplicate document filename")
seen_ids.add(entry["canary_id"])
seen_placements.add(entry["placement_id"])
seen_documents.add(filename)
seen_urls.add(url)
return registry, digest
def ensure_private_directory(path: Path) -> None:
path.mkdir(parents=True, exist_ok=True, mode=0o700)
if path.is_symlink() or not path.is_dir():
raise PermissionError(f"output root is not a real directory: {path}")
def write_once(path: Path, payload: bytes) -> None:
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0),
0o600,
)
try:
view = memoryview(payload)
while view:
written = os.write(descriptor, view)
if written <= 0:
raise OSError("short write")
view = view[written:]
os.fsync(descriptor)
finally:
os.close(descriptor)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--registry", type=Path, required=True)
parser.add_argument("--expected-registry-sha256", required=True)
parser.add_argument("--canary-id", required=True)
args = parser.parse_args()
registry, registry_sha256 = load_registry(
args.registry, args.expected_registry_sha256
)
builder_sha256 = hashlib.sha256(read_stable(Path(__file__), BUILDER_MAXIMUM_BYTES)).hexdigest()
if builder_sha256 != registry["builder_sha256"]:
raise RuntimeError("running builder bytes differ from the signed registry")
matches = [
entry for entry in registry["entries"] if entry["canary_id"] == args.canary_id
]
if len(matches) != 1:
raise ValueError("requested canary_id is absent or ambiguous")
entry = matches[0]
url = validated_url(entry)
document = (
"# 2026 Acquisition Scenario - Synthetic Exercise\n\n"
"This is a controlled synthetic planning artifact.\n\n"
f"\n"
).encode("utf-8")
document_sha256 = hashlib.sha256(document).hexdigest()
if document_sha256 != entry["expected_document_sha256"]:
raise RuntimeError("generated document differs from the signed expected digest")
ensure_private_directory(OUTPUT_ROOT)
ensure_private_directory(RECEIPT_ROOT)
output = OUTPUT_ROOT / entry["document_filename"]
receipt_path = RECEIPT_ROOT / f"{entry['canary_id']}.json"
if output.exists() or receipt_path.exists():
raise FileExistsError("document or build receipt already exists")
write_once(output, document)
receipt = {
"schema_version": "aidefend.dv002.url-canary-build.v1",
"registry_version": registry["registry_version"],
"registry_sha256": registry_sha256,
"registry_entry_sha256": hashlib.sha256(canonical(entry)).hexdigest(),
"builder_sha256": builder_sha256,
"canary_id": entry["canary_id"],
"placement_id": entry["placement_id"],
"owner_id": entry["owner_id"],
"expected_scanner_ids": entry["expected_scanner_ids"],
"alert_route_id": entry["alert_route_id"],
"url_sha256": hashlib.sha256(url.encode("utf-8")).hexdigest(),
"document_sha256": document_sha256,
"document_path": output.as_posix(),
"built_at": datetime.now(timezone.utc).isoformat(),
}
receipt_bytes = canonical(receipt) + b"\n"
write_once(receipt_path, receipt_bytes)
if (
read_stable(output, DOCUMENT_MAXIMUM_BYTES) != document
or read_stable(receipt_path, RECEIPT_MAXIMUM_BYTES) != receipt_bytes
):
raise RuntimeError("document/receipt readback differs; quarantine the pair")
print(json.dumps(receipt, sort_keys=True))
if __name__ == "__main__":
main()Verify the exact registry before building
Deploy the registry, Sigstore bundle, fixed public key, and Cosign v3.1.2 binary through the protected build image. The 15-second value below is a compiled bootstrap ceiling for this example, not a framework-wide policy default. The builder receives only the private verified snapshot and its observed digest.
#!/usr/bin/env bash
set -euo pipefail
umask 077
: "${URL_CANARY_ID:?required}"
registry=/run/policy/dv002-url-canary-registry.json
bundle=/run/policy/dv002-url-canary-registry.sigstore.json
trust_key=/run/trust/dv002-url-canary-registry.pub
cosign_binary=/usr/local/bin/cosign
for artifact in "$registry" "$bundle" "$trust_key" "$cosign_binary"; do
if [[ ! -f "$artifact" || -L "$artifact" ]]; then
echo "required URL-canary artifact is absent, non-regular, or a symlink: $artifact" >&2
exit 2
fi
done
for artifact in "$trust_key" "$cosign_binary"; do
owner="$(stat -Lc '%u' -- "$artifact")"
mode="$(stat -Lc '%a' -- "$artifact")"
if (( 10#$owner != 0 || (8#$mode & 022) != 0 )); then
echo "trust material must be root-owned and not group/world writable: $artifact" >&2
exit 2
fi
done
[[ -x "$cosign_binary" ]] || { echo "fixed Cosign binary is not executable" >&2; exit 2; }
snapshot_root="$(mktemp -d)"
chmod 0700 "$snapshot_root"
trap 'rm -rf -- "$snapshot_root"' EXIT
registry_snapshot="$snapshot_root/registry.json"
bundle_snapshot="$snapshot_root/registry.sigstore.json"
key_snapshot="$snapshot_root/verify.pub"
install -m 0400 -- "$registry" "$registry_snapshot"
install -m 0400 -- "$bundle" "$bundle_snapshot"
install -m 0400 -- "$trust_key" "$key_snapshot"
registry_sha256="$(sha256sum -- "$registry_snapshot" | awk '{print $1}')"
bundle_sha256="$(sha256sum -- "$bundle_snapshot" | awk '{print $1}')"
key_sha256="$(sha256sum -- "$key_snapshot" | awk '{print $1}')"
"$cosign_binary" verify-blob --timeout 15s \
--key "$key_snapshot" \
--bundle "$bundle_snapshot" \
"$registry_snapshot" >/dev/null
if [[ "$(sha256sum -- "$registry_snapshot" | awk '{print $1}')" != "$registry_sha256" \
|| "$(sha256sum -- "$bundle_snapshot" | awk '{print $1}')" != "$bundle_sha256" \
|| "$(sha256sum -- "$key_snapshot" | awk '{print $1}')" != "$key_sha256" ]]; then
echo "URL-canary verification snapshot changed" >&2
exit 1
fi
python deception/build_url_canary_document.py \
--registry "$registry_snapshot" \
--expected-registry-sha256 "$registry_sha256" \
--canary-id "$URL_CANARY_ID"Verify placement and the end-to-end signal
Sign the registry before placement. Independently read back the exact file bytes and metadata from each registered location, verify its digest, and prove it is excluded from search results and workflows used by legitimate staff unless the signed scenario intentionally exposes it. From a controlled drill host, open the exact document and require one provider event and one SIEM event bound to canary ID, document digest, placement, drill ID, timestamp, proxy metadata, and alert ID. Baseline known security scanners separately and never suppress an unexplained production hit automatically.
Action: Place only digest-registered synthetic documents, run the controlled drill, and treat live hits as investigation signals requiring correlation rather than sole attribution.
The example.invalid entry is deliberately non-routable. Promotion replaces it only with a provider-issued organization-owned URL in the authenticated registry, runs the builder in an isolated private workspace, uses create-once writes, verifies both artifacts by exact readback, and emits a placement-eligible receipt only after both succeed. Treat any partial pair as ERROR, quarantine it before a controlled retry, and never place the document externally without that receipt.
AID-DV-002-G005
Watermark synthetic data in honeypots / decoys.
Runtime policy: Load minimum cohort size, significance level, baseline rate, and minimum effect from the signed, versioned test plan bound to this generator and watermark design, bind its version/digest to observations and replay, and return ERROR if the plan or any value is absent, invalid, or unverifiable.
Scope and safeguards
Embed a pre-registered statistical fingerprint only in synthetic decoy data. The watermark design, baseline, minimum population, effect size, significance level, secret key ID, generator digest, and authorized surfaces must be signed before generation. A watermark match is corroborating provenance evidence; it does not identify the actor, transfer path, or prove theft by itself.
Generate a deterministic keyed population
# File: deception/watermark_control.py
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import math
import os
import re
import stat
import sys
from datetime import datetime, timezone
from pathlib import Path
import scipy
from scipy.stats import binomtest
POLICY_MAXIMUM_BYTES = 1_048_576
BOOTSTRAP_POPULATION_MAXIMUM_BYTES = 268_435_456
BOOTSTRAP_ROW_MAXIMUM_BYTES = 1_048_576
KEY_PATH = Path("/run/secrets/deception/dv002-watermark-hmac.key")
KEY_ID_PATH = Path("/run/secrets/deception/dv002-watermark-hmac.key-id")
SHA256 = re.compile(r"[0-9a-f]{64}")
IDENTIFIER = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9._:/-]{0,255}")
PLAN_FIELDS = {
"schema_version",
"plan_version",
"watermark_population_id",
"authorized_surface",
"source_population_sha256",
"expected_source_record_count",
"eligible_month",
"target_zip_digit",
"selection_numerator",
"selection_denominator",
"baseline_rate",
"minimum_samples",
"alpha",
"minimum_effect",
"hypothesis_count",
"key_id",
"key_sha256",
"selection_algorithm",
"scipy_version",
"generator_sha256",
"maximum_population_bytes",
"maximum_row_bytes",
"maximum_evidence_age_seconds",
"maximum_future_skew_seconds",
}
RECEIPT_FIELDS = {
"schema_version",
"receipt_version",
"plan_sha256",
"authorized_surface",
"collection_complete",
"collected_at",
"population_sha256",
"record_count",
"source_checkpoint",
}
def canonical(value: object) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def reject_duplicates(pairs: list[tuple[str, object]]) -> dict:
result: dict = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON key: {key}")
result[key] = value
return result
def reject_nonfinite(value: str) -> None:
raise ValueError(f"non-finite JSON value: {value}")
def read_stable(
path: Path, maximum_bytes: int, *, allow_empty: bool = False,
require_root_controlled: bool = False, require_private: bool = False,
) -> bytes:
flags = (
os.O_RDONLY
| getattr(os, "O_CLOEXEC", 0)
| getattr(os, "O_BINARY", 0)
)
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = os.open(path, flags)
try:
before = os.fstat(descriptor)
minimum_bytes = 0 if allow_empty else 1
if (
not stat.S_ISREG(before.st_mode)
or not minimum_bytes <= before.st_size <= maximum_bytes
):
raise ValueError(f"artifact is not a bounded regular file: {path}")
if require_root_controlled and (before.st_uid != 0 or before.st_mode & 0o022):
raise PermissionError(f"artifact is not root-controlled: {path}")
if require_private and before.st_mode & 0o077:
raise PermissionError(f"secret is group/world accessible: {path}")
chunks: list[bytes] = []
observed = 0
while True:
chunk = os.read(descriptor, min(65_536, maximum_bytes + 1 - observed))
if not chunk:
break
chunks.append(chunk)
observed += len(chunk)
if observed > maximum_bytes:
raise ValueError(f"artifact exceeds byte ceiling: {path}")
after = os.fstat(descriptor)
if (
observed != before.st_size
or (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
!= (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
):
raise RuntimeError(f"artifact changed while being read: {path}")
return b"".join(chunks)
finally:
os.close(descriptor)
def strict_json(raw: bytes) -> object:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicates,
parse_constant=reject_nonfinite,
)
def load_verified_json(
path: Path, expected_sha256: str, maximum_bytes: int
) -> tuple[dict, str]:
if not SHA256.fullmatch(expected_sha256):
raise ValueError("verified snapshot digest is not lower-case SHA-256")
raw = read_stable(path, maximum_bytes)
digest = hashlib.sha256(raw).hexdigest()
if digest != expected_sha256:
raise RuntimeError("artifact differs from the signature-verified snapshot")
value = strict_json(raw)
if not isinstance(value, dict):
raise ValueError("signed JSON root must be an object")
return value, digest
def positive_integer(value: object, label: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError(f"{label} must be a positive integer")
return value
def finite_number(value: object, label: str) -> float:
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or not math.isfinite(float(value))
):
raise ValueError(f"{label} must be a finite number")
return float(value)
def validate_identifier(value: object, label: str) -> str:
if not isinstance(value, str) or not IDENTIFIER.fullmatch(value):
raise ValueError(f"{label} has an invalid shape")
return value
def validate_plan(plan: dict) -> None:
if set(plan) != PLAN_FIELDS:
raise ValueError("watermark plan fields differ")
if plan["schema_version"] != "aidefend.dv002.watermark-plan.v1":
raise ValueError("unsupported watermark plan schema")
for field in (
"plan_version",
"watermark_population_id",
"authorized_surface",
"key_id",
"scipy_version",
):
validate_identifier(plan[field], field)
for field in ("source_population_sha256", "generator_sha256", "key_sha256"):
if not isinstance(plan[field], str) or not SHA256.fullmatch(plan[field]):
raise ValueError(f"invalid plan digest: {field}")
for field in (
"expected_source_record_count",
"selection_numerator",
"selection_denominator",
"minimum_samples",
"hypothesis_count",
"maximum_population_bytes",
"maximum_row_bytes",
"maximum_evidence_age_seconds",
):
positive_integer(plan[field], field)
if (
isinstance(plan["maximum_future_skew_seconds"], bool)
or not isinstance(plan["maximum_future_skew_seconds"], int)
or plan["maximum_future_skew_seconds"] < 0
):
raise ValueError("maximum_future_skew_seconds must be a nonnegative integer")
if (
isinstance(plan["eligible_month"], bool)
or not isinstance(plan["eligible_month"], int)
or not 1 <= plan["eligible_month"] <= 12
):
raise ValueError("eligible_month must be an integer in 1..12")
if (
not isinstance(plan["target_zip_digit"], str)
or len(plan["target_zip_digit"]) != 1
or plan["target_zip_digit"] not in "0123456789"
):
raise ValueError("target_zip_digit must be one ASCII digit")
if not 1 <= plan["selection_numerator"] < plan["selection_denominator"] <= (1 << 256):
raise ValueError("selection ratio is invalid or exceeds the HMAC-SHA256 domain")
if plan["selection_algorithm"] != "hmac-sha256-threshold-v1":
raise ValueError("unsupported deterministic selection algorithm")
baseline = finite_number(plan["baseline_rate"], "baseline_rate")
alpha = finite_number(plan["alpha"], "alpha")
effect = finite_number(plan["minimum_effect"], "minimum_effect")
if not 0 < baseline < 1 or not 0 < alpha < 1 or not 0 < effect < 1 - baseline:
raise ValueError("watermark statistical bounds are invalid")
if plan["maximum_population_bytes"] > BOOTSTRAP_POPULATION_MAXIMUM_BYTES:
raise ValueError("signed population bound exceeds the implementation ceiling")
if (
plan["maximum_row_bytes"] > BOOTSTRAP_ROW_MAXIMUM_BYTES
or plan["maximum_row_bytes"] > plan["maximum_population_bytes"]
):
raise ValueError("signed row bound exceeds an implementation ceiling")
def parse_jsonl(
raw: bytes, maximum_row_bytes: int, *, allow_empty: bool = False
) -> list[dict]:
if not raw and allow_empty:
return []
if not raw.endswith(b"\n"):
raise ValueError("JSONL population must end with a newline")
rows: list[dict] = []
seen_ids: set[str] = set()
for line_number, line in enumerate(raw.splitlines(), start=1):
if not line or len(line) > maximum_row_bytes:
raise ValueError(f"JSONL line {line_number} is empty or oversized")
row = strict_json(line)
if not isinstance(row, dict):
raise ValueError(f"JSONL line {line_number} is not an object")
record_id = row.get("synthetic_record_id")
if (
not isinstance(record_id, str)
or not record_id
or len(record_id.encode("utf-8")) > 256
or record_id in seen_ids
):
raise ValueError("synthetic_record_id is absent, oversized, or duplicated")
seen_ids.add(record_id)
rows.append(row)
if not rows and not allow_empty:
raise ValueError("JSONL population is empty")
return rows
def validate_row(row: dict) -> tuple[str, str, int]:
record_id = row["synthetic_record_id"]
zipcode = row.get("zipcode")
month = row.get("month")
if (
not isinstance(zipcode, str)
or len(zipcode) != 5
or not zipcode.isascii()
or not zipcode.isdigit()
or isinstance(month, bool)
or not isinstance(month, int)
or not 1 <= month <= 12
):
raise ValueError(f"record {record_id} has a noncanonical ZIP code or month")
return record_id, zipcode, month
def read_secret(path: Path, *, maximum_bytes: int = 4096) -> bytes:
value = read_stable(
path, maximum_bytes, require_root_controlled=True, require_private=True
)
if len(value) < 32:
raise ValueError(f"secret is shorter than 32 bytes: {path}")
return value
def selected(plan: dict, key: bytes, record_id: str) -> bool:
material = (
plan["watermark_population_id"] + "\x00" + record_id
).encode("utf-8")
digest = hmac.new(key, material, hashlib.sha256).digest()
value = int.from_bytes(digest, "big")
return (
value * plan["selection_denominator"]
< plan["selection_numerator"] * (1 << 256)
)
def write_once(path: Path, payload: bytes) -> None:
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
if path.parent.is_symlink() or not path.parent.is_dir():
raise PermissionError("output parent is not a real directory")
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0),
0o600,
)
try:
view = memoryview(payload)
while view:
written = os.write(descriptor, view)
if written <= 0:
raise OSError("short write")
view = view[written:]
os.fsync(descriptor)
finally:
os.close(descriptor)
def verify_runtime(plan: dict) -> str:
code_sha256 = hashlib.sha256(
read_stable(Path(__file__), POLICY_MAXIMUM_BYTES)
).hexdigest()
if code_sha256 != plan["generator_sha256"]:
raise RuntimeError("control bytes differ from the signed plan")
if scipy.__version__ != plan["scipy_version"]:
raise RuntimeError("SciPy runtime differs from the signed plan")
return code_sha256
def generate(
args: argparse.Namespace, plan: dict, plan_sha256: str, code_sha256: str
) -> None:
source_raw = read_stable(Path(args.source), plan["maximum_population_bytes"])
if hashlib.sha256(source_raw).hexdigest() != plan["source_population_sha256"]:
raise RuntimeError("source population digest differs from the signed plan")
rows = parse_jsonl(source_raw, plan["maximum_row_bytes"])
if len(rows) != plan["expected_source_record_count"]:
raise RuntimeError("source population count differs from the signed plan")
key_id = read_stable(
KEY_ID_PATH, 512, require_root_controlled=True
).decode("utf-8", errors="strict").strip()
if key_id != plan["key_id"]:
raise RuntimeError("mounted watermark key ID differs from the signed plan")
key = read_secret(KEY_PATH)
key_sha256 = hashlib.sha256(key).hexdigest()
if not hmac.compare_digest(key_sha256, plan["key_sha256"]):
raise RuntimeError("mounted HMAC key differs from the signed plan")
output_rows: list[dict] = []
selected_count = 0
for row in rows:
record_id, zipcode, month = validate_row(row)
transformed = dict(row)
if month == plan["eligible_month"] and selected(plan, key, record_id):
transformed["zipcode"] = zipcode[:-1] + plan["target_zip_digit"]
selected_count += 1
output_rows.append(transformed)
output_raw = b"".join(canonical(row) + b"\n" for row in output_rows)
if len(output_raw) > plan["maximum_population_bytes"]:
raise RuntimeError("generated population exceeds the signed byte bound")
output_path = Path(args.output)
manifest_path = Path(args.manifest)
if output_path.exists() or manifest_path.exists():
raise FileExistsError("generated population or private manifest already exists")
write_once(output_path, output_raw)
manifest = {
"schema_version": "aidefend.dv002.watermark-generation.v1",
"plan_version": plan["plan_version"],
"plan_sha256": plan_sha256,
"watermark_population_id": plan["watermark_population_id"],
"authorized_surface": plan["authorized_surface"],
"source_population_sha256": plan["source_population_sha256"],
"output_population_sha256": hashlib.sha256(output_raw).hexdigest(),
"record_count": len(output_rows),
"eligible_record_count": sum(
validate_row(row)[2] == plan["eligible_month"] for row in rows
),
"selected_record_count": selected_count,
"key_id": key_id,
"key_sha256": key_sha256,
"selection_algorithm": plan["selection_algorithm"],
"scipy_version": scipy.__version__,
"generator_sha256": code_sha256,
"generated_at": datetime.now(timezone.utc).isoformat(),
}
write_once(manifest_path, canonical(manifest) + b"\n")
print(json.dumps(manifest, sort_keys=True))
def parse_time(value: object) -> datetime:
if not isinstance(value, str) or not value:
raise ValueError("receipt timestamp is absent")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.utcoffset() is None:
raise ValueError("receipt timestamp lacks a timezone")
return parsed.astimezone(timezone.utc)
def validate_receipt(receipt: dict, plan: dict, plan_sha256: str) -> None:
if set(receipt) != RECEIPT_FIELDS:
raise ValueError("population receipt fields differ")
if receipt["schema_version"] != "aidefend.dv002.watermark-population-receipt.v1":
raise ValueError("unsupported population receipt schema")
for field in ("receipt_version", "authorized_surface", "source_checkpoint"):
validate_identifier(receipt[field], field)
if receipt["authorized_surface"] != plan["authorized_surface"]:
raise RuntimeError("population receipt surface differs from the signed plan")
if receipt["plan_sha256"] != plan_sha256:
raise RuntimeError("population receipt is bound to another plan")
if not isinstance(receipt["collection_complete"], bool):
raise ValueError("collection_complete must be Boolean")
if (
isinstance(receipt["record_count"], bool)
or not isinstance(receipt["record_count"], int)
or receipt["record_count"] < 0
):
raise ValueError("record_count must be a nonnegative integer")
if not isinstance(receipt["population_sha256"], str) or not SHA256.fullmatch(
receipt["population_sha256"]
):
raise ValueError("population receipt digest is invalid")
observed_at = parse_time(receipt["collected_at"])
now = datetime.now(timezone.utc)
age = (now - observed_at).total_seconds()
if age < -plan["maximum_future_skew_seconds"]:
raise ValueError("population receipt is from the future")
if age > plan["maximum_evidence_age_seconds"]:
raise ValueError("population receipt is stale")
def evaluate(args: argparse.Namespace, plan: dict, plan_sha256: str) -> None:
receipt, receipt_sha256 = load_verified_json(
Path(args.receipt), args.expected_receipt_sha256, POLICY_MAXIMUM_BYTES
)
validate_receipt(receipt, plan, plan_sha256)
population_raw = read_stable(
Path(args.population),
plan["maximum_population_bytes"],
allow_empty=True,
)
population_sha256 = hashlib.sha256(population_raw).hexdigest()
rows = parse_jsonl(
population_raw, plan["maximum_row_bytes"], allow_empty=True
)
if (
population_sha256 != receipt["population_sha256"]
or len(rows) != receipt["record_count"]
):
raise RuntimeError("observed population differs from its signed receipt")
eligible = []
for row in rows:
_, zipcode, month = validate_row(row)
if month == plan["eligible_month"]:
eligible.append(zipcode[-1])
if receipt["collection_complete"] is not True:
result = {
"status": "INSUFFICIENT_DATA",
"reason": "signed population receipt reports incomplete collection",
"sample_count": len(eligible),
}
elif len(eligible) < plan["minimum_samples"]:
result = {
"status": "INSUFFICIENT_DATA",
"reason": "eligible sample count is below the signed minimum",
"sample_count": len(eligible),
}
else:
hits = eligible.count(plan["target_zip_digit"])
observed_rate = hits / len(eligible)
adjusted_alpha = float(plan["alpha"]) / plan["hypothesis_count"]
p_value = float(
binomtest(
hits,
len(eligible),
p=float(plan["baseline_rate"]),
alternative="greater",
).pvalue
)
finding = (
p_value <= adjusted_alpha
and observed_rate - float(plan["baseline_rate"])
>= float(plan["minimum_effect"])
)
result = {
"status": "FAIL" if finding else "PASS",
"reason": (
"pre-registered watermark threshold crossed"
if finding
else "pre-registered watermark threshold not crossed"
),
"sample_count": len(eligible),
"hits": hits,
"observed_rate": observed_rate,
"baseline_rate": float(plan["baseline_rate"]),
"p_value": p_value,
"alpha": float(plan["alpha"]),
"adjusted_alpha": adjusted_alpha,
"minimum_effect": float(plan["minimum_effect"]),
"hypothesis_count": plan["hypothesis_count"],
}
evidence = {
"schema_version": "aidefend.dv002.watermark-evaluation.v1",
"plan_version": plan["plan_version"],
"plan_sha256": plan_sha256,
"receipt_sha256": receipt_sha256,
"population_sha256": population_sha256,
"authorized_surface": plan["authorized_surface"],
"watermark_population_id": plan["watermark_population_id"],
"test": "scipy.stats.binomtest.one-sided-greater",
"selection_algorithm": plan["selection_algorithm"],
"generator_sha256": plan["generator_sha256"],
"key_id": plan["key_id"],
"key_sha256": plan["key_sha256"],
"scipy_version": scipy.__version__,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
**result,
}
output = Path(args.result)
if output.exists():
raise FileExistsError("evaluation result already exists")
write_once(output, canonical(evidence) + b"\n")
print(json.dumps(evidence, sort_keys=True))
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--plan", type=Path, required=True)
parser.add_argument("--expected-plan-sha256", required=True)
subparsers = parser.add_subparsers(dest="command", required=True)
generate_parser = subparsers.add_parser("generate")
generate_parser.add_argument("--source", required=True)
generate_parser.add_argument("--output", required=True)
generate_parser.add_argument("--manifest", required=True)
evaluate_parser = subparsers.add_parser("evaluate")
evaluate_parser.add_argument("--population", required=True)
evaluate_parser.add_argument("--receipt", required=True)
evaluate_parser.add_argument("--expected-receipt-sha256", required=True)
evaluate_parser.add_argument("--result", required=True)
args = parser.parse_args()
plan, plan_sha256 = load_verified_json(
args.plan, args.expected_plan_sha256, POLICY_MAXIMUM_BYTES
)
validate_plan(plan)
code_sha256 = verify_runtime(plan)
if args.command == "generate":
generate(args, plan, plan_sha256, code_sha256)
else:
evaluate(args, plan, plan_sha256)
if __name__ == "__main__":
try:
main()
except Exception as exc:
print(
json.dumps(
{"status": "ERROR", "error_type": type(exc).__name__},
sort_keys=True,
),
file=sys.stderr,
)
raise SystemExit(2) from NoneRun the pre-registered test
Verify the signed plan and complete-population receipt, then run
The plan signer and population-collector signer are separate trust identities. Render every plan value from the versioned organizational test policy, sign the canonical plan only after binding the exact source and generator digests, and pin SciPy plus package hashes in the evaluator image. The 15-second Cosign value is an implementation bootstrap ceiling, not a universal framework threshold.
#!/usr/bin/env bash
set -euo pipefail
umask 077
cosign_binary=/usr/local/bin/cosign
plan=/run/policy/dv002-watermark-plan.json
plan_bundle=/run/policy/dv002-watermark-plan.sigstore.json
plan_key=/run/trust/dv002-watermark-plan.pub
receipt=/run/evidence/dv002-suspect-population-receipt.json
receipt_bundle=/run/evidence/dv002-suspect-population-receipt.sigstore.json
receipt_key=/run/trust/dv002-population-collector.pub
for artifact in "$cosign_binary" "$plan" "$plan_bundle" "$plan_key" \
"$receipt" "$receipt_bundle" "$receipt_key"; do
if [[ ! -f "$artifact" || -L "$artifact" ]]; then
echo "required watermark artifact is absent, non-regular, or a symlink: $artifact" >&2
exit 2
fi
done
for artifact in "$cosign_binary" "$plan_key" "$receipt_key"; do
owner="$(stat -Lc '%u' -- "$artifact")"
mode="$(stat -Lc '%a' -- "$artifact")"
if (( 10#$owner != 0 || (8#$mode & 022) != 0 )); then
echo "watermark trust material must be root-owned and not group/world writable: $artifact" >&2
exit 2
fi
done
[[ -x "$cosign_binary" ]] || { echo "fixed Cosign binary is not executable" >&2; exit 2; }
snapshot_root="$(mktemp -d)"
chmod 0700 "$snapshot_root"
trap 'rm -rf -- "$snapshot_root"' EXIT
plan_snapshot="$snapshot_root/plan.json"
plan_bundle_snapshot="$snapshot_root/plan.sigstore.json"
plan_key_snapshot="$snapshot_root/plan.pub"
receipt_snapshot="$snapshot_root/receipt.json"
receipt_bundle_snapshot="$snapshot_root/receipt.sigstore.json"
receipt_key_snapshot="$snapshot_root/receipt.pub"
install -m 0400 -- "$plan" "$plan_snapshot"
install -m 0400 -- "$plan_bundle" "$plan_bundle_snapshot"
install -m 0400 -- "$plan_key" "$plan_key_snapshot"
install -m 0400 -- "$receipt" "$receipt_snapshot"
install -m 0400 -- "$receipt_bundle" "$receipt_bundle_snapshot"
install -m 0400 -- "$receipt_key" "$receipt_key_snapshot"
plan_sha256="$(sha256sum -- "$plan_snapshot" | awk '{print $1}')"
plan_bundle_sha256="$(sha256sum -- "$plan_bundle_snapshot" | awk '{print $1}')"
plan_key_sha256="$(sha256sum -- "$plan_key_snapshot" | awk '{print $1}')"
receipt_sha256="$(sha256sum -- "$receipt_snapshot" | awk '{print $1}')"
receipt_bundle_sha256="$(sha256sum -- "$receipt_bundle_snapshot" | awk '{print $1}')"
receipt_key_sha256="$(sha256sum -- "$receipt_key_snapshot" | awk '{print $1}')"
"$cosign_binary" verify-blob --timeout 15s \
--key "$plan_key_snapshot" --bundle "$plan_bundle_snapshot" "$plan_snapshot" >/dev/null
"$cosign_binary" verify-blob --timeout 15s \
--key "$receipt_key_snapshot" --bundle "$receipt_bundle_snapshot" "$receipt_snapshot" >/dev/null
for binding in \
"$plan_snapshot:$plan_sha256" \
"$plan_bundle_snapshot:$plan_bundle_sha256" \
"$plan_key_snapshot:$plan_key_sha256" \
"$receipt_snapshot:$receipt_sha256" \
"$receipt_bundle_snapshot:$receipt_bundle_sha256" \
"$receipt_key_snapshot:$receipt_key_sha256"; do
path="${binding%%:*}"
expected="${binding##*:}"
[[ "$(sha256sum -- "$path" | awk '{print $1}')" == "$expected" ]] || {
echo "watermark verification snapshot changed: $path" >&2
exit 1
}
done
# Generation consumes the exact signed plan and fails if source bytes, row count,
# generator digest, key ID, or output bounds differ from that plan.
python deception/watermark_control.py \
--plan "$plan_snapshot" \
--expected-plan-sha256 "$plan_sha256" \
generate \
--source fixtures/synthetic-source.jsonl \
--output out/watermarked-population.jsonl \
--manifest security-only/watermark-generation-manifest.json
# Evaluation consumes an independently collected population plus its exact signed
# completeness receipt; an incomplete or undersized population is never PASS.
python deception/watermark_control.py \
--plan "$plan_snapshot" \
--expected-plan-sha256 "$plan_sha256" \
evaluate \
--population evidence/suspect-population.jsonl \
--receipt "$receipt_snapshot" \
--expected-receipt-sha256 "$receipt_sha256" \
--result evidence/watermark-evaluation.jsonIndependent verification
Retain the signed design/policy, secret-key identifier (never the key), generator image/code digest, exact generated population manifest and digest, authorized placement receipt, suspect-population provenance and completeness receipt, test code/environment digest, raw aggregate counts, result, and independent replay. Include negative-control populations and correct for multiple tests when the signed plan evaluates more than one pattern.
Action: Generate, place, and test only the exact signed population; never tune the threshold after observing the suspect data or make attribution from this signal alone.
Quarantine any partial population/manifest publication and never repair it in place. An independent replay re-verifies both signatures, recomputes plan, population, receipt, and code digests, and reproduces the exact one-sided binomial result from the complete measured population.
The signed plan binds the exact control bytes, SciPy version, full-digest selection algorithm, key identifier, and SHA-256 of the mounted high-entropy HMAC key. The implementation rejects linked, non-root-controlled, or group/world-accessible key material and emits a bounded ERROR record with exit code 2 for invalid or unverifiable runtime inputs.
AID-DV-002-G006
Ensure honey elements are isolated and cannot impact production.
Required inputs and boundaries
This method uses the verified signed honey registry produced by the sibling step and verifies that no registered honey asset enters a legitimate production consumer, action, metric, training/retrieval corpus, notification, billing flow, compliance export, or real credential/runtime binding. Verify the upstream registry receipt locally before testing exclusion.
Enforce exclusion at the storage boundary
-- File: deception/production_user_surface.sql
BEGIN;
CREATE SCHEMA IF NOT EXISTS deception;
CREATE TABLE IF NOT EXISTS deception.honey_asset_registry (
honey_id text PRIMARY KEY,
asset_type text NOT NULL CHECK (asset_type IN ('USER','DOCUMENT','MODEL','CREDENTIAL','RAG_CHUNK')),
registry_sha256 text NOT NULL CHECK (registry_sha256 ~ '^[0-9a-f]{64}$'),
owner_team text NOT NULL,
created_at timestamptz NOT NULL
);
REVOKE ALL ON deception.honey_asset_registry FROM PUBLIC;
REVOKE ALL ON public.users FROM PUBLIC;
REVOKE ALL ON public.users FROM production_app;
CREATE OR REPLACE VIEW public.production_users
WITH (security_barrier = true)
AS
SELECT u.user_id, u.email, u.name, u.last_login
FROM public.users AS u
WHERE NOT EXISTS (
SELECT 1 FROM deception.honey_asset_registry AS h
WHERE h.asset_type = 'USER' AND h.honey_id = u.user_id
);
GRANT SELECT ON public.production_users TO production_app;
DO $$
DECLARE
privilege_name text;
BEGIN
IF (SELECT c.relowner = r.oid
FROM pg_class AS c
JOIN pg_namespace AS n ON n.oid = c.relnamespace
JOIN pg_roles AS r ON r.rolname = 'production_app'
WHERE n.nspname = 'public' AND c.relname = 'users') THEN
RAISE EXCEPTION 'production_app must not own public.users';
END IF;
FOREACH privilege_name IN ARRAY ARRAY[
'SELECT','INSERT','UPDATE','DELETE','TRUNCATE','REFERENCES','TRIGGER'
] LOOP
IF has_table_privilege('production_app', 'public.users', privilege_name) THEN
RAISE EXCEPTION 'production_app retains effective % on public.users', privilege_name;
END IF;
END LOOP;
FOREACH privilege_name IN ARRAY ARRAY[
'SELECT','INSERT','UPDATE','REFERENCES'
] LOOP
IF has_any_column_privilege('production_app', 'public.users', privilege_name) THEN
RAISE EXCEPTION 'production_app retains effective column-level % on public.users', privilege_name;
END IF;
END LOOP;
END $$;
COMMIT;Apply the same enforced boundary to vector indices, analytics materializations, training manifests, notification/billing exports, identity providers, secrets, and model registries. Prefer separate namespaces/accounts and deny production identities direct access to raw mixed tables or decoy stores. A code convention or optional NOT IN filter is not an enforcement boundary.
Reconcile every consumer
From independent read identities, enumerate the complete signed honey population and each legitimate consumer's exact output population. Compute set intersections by stable asset ID and require empty sets. Test one registered asset of each applicable type through each consumer, prove denial/exclusion, and retain query plans/policies, role grants, index/export manifests, population digests/counts, and replay results.
Action: Enforce exclusion at authoritative storage and identity boundaries, then continuously replay exact-set reconciliation across every declared consumer.
Run the migration as a separate owner role. Before committing, the guard checks effective privileges after direct, PUBLIC, and inherited grants; separately audit pg_auth_members/pg_has_role(..., 'SET') and remove any path that lets production_app assume the table owner or another raw-table grantee.
AID-DV-002-G007
Integrate honey element alerts into security monitoring.
Runtime policy: Load webhook size/age, identity, secret binding, outbox, and SQLite bounds from the fixed root-controlled, versioned intake profile bound to this canary placement, bind its version/digest to outbox and SIEM receipts, and return ERROR if the profile or any value is absent, invalid, or unverifiable.
Required inputs and boundaries
Normalize provider and self-managed honey events only after verifying the webhook or message-bus producer, raw-body signature, freshness, nonce/event ID, expected canary ID, and registered placement. This step handles durable intake, enrichment, exact-once SIEM delivery, and response routing; the sibling canary step handles placement and trigger behavior.
Preprovision the private SQLite outbox
Create the directory, database, WAL mode, and schema once as a deployment step under the same dedicated UID that runs the intake. Keep every ancestor root- or service-owned and non-group/world-writable. The runtime deliberately uses mode=rw; a missing file, symlink, wrong owner/mode, hard link, changed inode, or uninitialized schema is an operational error rather than permission to create a replacement.
#!/usr/bin/env bash
set -euo pipefail
install -d -o aidefend-dv002 -g aidefend-dv002 -m 0700 /var/lib/aidefend/dv002
outbox=/var/lib/aidefend/dv002/honey-events.sqlite3
if [[ -e "$outbox" || -L "$outbox" ]]; then
echo "refusing to replace an existing outbox" >&2
exit 2
fi
install -o aidefend-dv002 -g aidefend-dv002 -m 0600 /dev/null "$outbox"
runuser -u aidefend-dv002 -- sqlite3 "$outbox" "PRAGMA journal_mode=WAL; CREATE TABLE alerts(event_id TEXT PRIMARY KEY,payload_sha256 TEXT NOT NULL,canonical_json BLOB NOT NULL,runtime_profile_version TEXT NOT NULL,runtime_profile_sha256 TEXT NOT NULL,received_at_ns INTEGER NOT NULL,delivered_at_ns INTEGER);"
chmod 0600 "$outbox"
Verify before acknowledging and commit to the outbox
# File: deception/honey_webhook.py
from __future__ import annotations
import hashlib
import hmac
import json
import math
import os
import re
import sqlite3
import stat
import time
from contextlib import closing
from pathlib import Path
from fastapi import FastAPI, Header, HTTPException, Request
PROFILE_PATH = Path("/run/policy/dv002-honey-intake-profile.json")
SECRET_PATH = Path("/run/secrets/deception/dv002-honey-webhook-hmac.key")
BOOTSTRAP_ARTIFACT_CEILING = 131_072
PROFILE_VERSION_STORAGE_CAPACITY = 128
HMAC_KEY_ID_STORAGE_CAPACITY = 256
OUTBOX_PATH_CAPACITY_BYTES = 4_096
INTEGER_CAPACITY_CEILINGS = {
"maximum_body_bytes": 1_048_576,
"maximum_event_id_bytes": 512,
"maximum_identity_bytes": 2_048,
"maximum_skew_seconds": 3_600,
}
SQLITE_BUSY_WAIT_IMPLEMENTATION_CEILING_SECONDS = 30.0
PROFILE_FIELDS = {
"schema_version", "profile_version", "hmac_key_id", "hmac_key_sha256",
"outbox_path", "maximum_body_bytes", "maximum_event_id_bytes",
"maximum_identity_bytes", "maximum_skew_seconds", "sqlite_busy_timeout_seconds",
}
EVENT_FIELDS = {"schema_version", "canary_id", "placement_id", "triggered_at", "source"}
EVENT_ID_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]*")
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON key: {key}")
result[key] = value
return result
def read_privileged(path: Path, ceiling: int) -> bytes:
if not hasattr(os, "O_NOFOLLOW"):
raise OSError("secure no-follow reads are unavailable")
descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
try:
before = os.fstat(descriptor)
if (not stat.S_ISREG(before.st_mode) or before.st_uid != 0
or before.st_mode & 0o022 or not 1 <= before.st_size <= ceiling):
raise PermissionError(f"untrusted privileged artifact: {path}")
chunks, observed = [], 0
while observed <= ceiling:
chunk = os.read(descriptor, min(65_536, ceiling + 1 - observed))
if not chunk:
break
chunks.append(chunk)
observed += len(chunk)
after = os.fstat(descriptor)
fingerprint = lambda item: (item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns)
if observed != before.st_size or observed > ceiling or fingerprint(before) != fingerprint(after):
raise OSError(f"privileged artifact changed during read: {path}")
return b"".join(chunks)
finally:
os.close(descriptor)
def strict_json(raw: bytes) -> object:
return json.loads(
raw.decode("utf-8", errors="strict"), object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(ValueError("non-finite JSON value: " + value)),
)
def load_contract() -> tuple[dict, str, bytes]:
raw = read_privileged(PROFILE_PATH, BOOTSTRAP_ARTIFACT_CEILING)
profile = strict_json(raw)
if not isinstance(profile, dict) or set(profile) != PROFILE_FIELDS:
raise RuntimeError("honey-intake profile schema differs")
if profile["schema_version"] != "aidefend.dv002.honey-intake-profile.v1":
raise RuntimeError("unsupported honey-intake profile")
text_capacities = {
"profile_version": PROFILE_VERSION_STORAGE_CAPACITY,
"hmac_key_id": HMAC_KEY_ID_STORAGE_CAPACITY,
}
for field, ceiling in text_capacities.items():
value = profile[field]
if (not isinstance(value, str) or not value.strip() or value != value.strip()
or len(value.encode("utf-8")) > ceiling
or any(ord(character) < 32 or 127 <= ord(character) <= 159 for character in value)):
raise RuntimeError(f"authenticated profile identity is invalid: {field}")
for field, ceiling in INTEGER_CAPACITY_CEILINGS.items():
if type(profile[field]) is not int or not 1 <= profile[field] <= ceiling:
raise RuntimeError(f"invalid authenticated integer: {field}")
busy = profile["sqlite_busy_timeout_seconds"]
if (type(busy) not in (int, float) or isinstance(busy, bool)
or not math.isfinite(float(busy))
or not 0 < float(busy) <= SQLITE_BUSY_WAIT_IMPLEMENTATION_CEILING_SECONDS):
raise RuntimeError("invalid authenticated SQLite timeout")
outbox = profile["outbox_path"]
if (not isinstance(outbox, str) or not outbox or len(outbox.encode("utf-8")) > OUTBOX_PATH_CAPACITY_BYTES
or any(ord(character) < 32 or 127 <= ord(character) <= 159 for character in outbox)
or not Path(outbox).is_absolute() or Path(outbox) != Path(os.path.normpath(outbox))):
raise RuntimeError("outbox_path must be a bounded normalized authenticated absolute path")
key_digest = profile["hmac_key_sha256"]
if not isinstance(key_digest, str) or len(key_digest) != 64 or set(key_digest) - set("0123456789abcdef"):
raise RuntimeError("authenticated HMAC-key digest is invalid")
secret = read_privileged(SECRET_PATH, BOOTSTRAP_ARTIFACT_CEILING)
if not 32 <= len(secret) <= 64 or not hmac.compare_digest(hashlib.sha256(secret).hexdigest(), key_digest):
raise RuntimeError("fixed HMAC secret is weak or differs from the authenticated profile")
return profile, hashlib.sha256(raw).hexdigest(), secret
PROFILE, PROFILE_SHA256, SECRET = load_contract()
OUTBOX = Path(PROFILE["outbox_path"])
app = FastAPI()
def open_verified_outbox() -> tuple[int, os.stat_result]:
if not hasattr(os, "O_NOFOLLOW"):
raise OSError("secure no-follow SQLite opens are unavailable")
if (not OUTBOX.is_absolute() or OUTBOX.name in {"", ".", ".."}
or OUTBOX != Path(os.path.normpath(str(OUTBOX)))):
raise PermissionError("outbox path is not a normalized absolute file path")
for ancestor in reversed(OUTBOX.parents):
metadata = os.lstat(ancestor)
if (not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode)
or metadata.st_uid not in {0, os.geteuid()} or metadata.st_mode & 0o022):
raise PermissionError(f"untrusted SQLite path ancestor: {ancestor}")
descriptor = os.open(OUTBOX, os.O_RDWR | os.O_CLOEXEC | os.O_NOFOLLOW)
try:
metadata = os.fstat(descriptor)
linked = os.stat(OUTBOX, follow_symlinks=False)
identity = lambda value: (value.st_dev, value.st_ino)
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.geteuid()
or stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_nlink != 1
or identity(metadata) != identity(linked)):
raise PermissionError("SQLite outbox is not the preprovisioned private regular file")
return descriptor, metadata
except Exception:
os.close(descriptor)
raise
def assert_outbox_identity(descriptor: int, expected: os.stat_result) -> None:
current = os.fstat(descriptor)
linked = os.stat(OUTBOX, follow_symlinks=False)
identity = lambda value: (value.st_dev, value.st_ino)
if (identity(current) != identity(expected) or identity(linked) != identity(expected)
or not stat.S_ISREG(linked.st_mode) or stat.S_IMODE(linked.st_mode) != 0o600
or linked.st_uid != os.geteuid() or linked.st_nlink != 1):
raise OSError("SQLite outbox identity changed during the transaction")
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
def bounded_identity(value: object, field: str) -> str:
if (not isinstance(value, str) or not value.strip()
or len(value.encode("utf-8")) > PROFILE["maximum_identity_bytes"]
or any(ord(character) < 32 for character in value)):
raise HTTPException(400, f"invalid event identity: {field}")
return value
async def bounded_body(request: Request) -> bytes:
ceiling = PROFILE["maximum_body_bytes"]
claimed = request.headers.get("content-length")
if claimed is not None:
if not claimed.isascii() or not claimed.isdigit() or int(claimed, 10) > ceiling:
raise HTTPException(413, "invalid or oversized content-length")
chunks, observed = [], 0
async for chunk in request.stream():
observed += len(chunk)
if observed > ceiling:
raise HTTPException(413, "request body exceeds authenticated bound")
chunks.append(bytes(chunk))
return b"".join(chunks)
def commit(event_id: str, record: dict) -> tuple[str, bool]:
payload = canonical(record)
payload_sha256 = hashlib.sha256(payload).hexdigest()
descriptor = -1
try:
descriptor, expected_metadata = open_verified_outbox()
uri = OUTBOX.as_uri() + "?mode=rw"
with closing(sqlite3.connect(
uri, uri=True, timeout=float(PROFILE["sqlite_busy_timeout_seconds"]),
)) as database:
with database:
assert_outbox_identity(descriptor, expected_metadata)
if database.execute("PRAGMA journal_mode").fetchone()[0].lower() != "wal":
raise sqlite3.DatabaseError("preprovisioned outbox is not in WAL mode")
database.execute("PRAGMA synchronous=FULL")
database.execute("BEGIN IMMEDIATE")
existing = database.execute(
"SELECT payload_sha256,canonical_json FROM alerts WHERE event_id=?", (event_id,),
).fetchone()
if existing is not None:
if existing[0] != payload_sha256 or bytes(existing[1]) != payload:
raise HTTPException(409, "event_id replay payload conflict")
assert_outbox_identity(descriptor, expected_metadata)
return payload_sha256, True
database.execute(
"INSERT INTO alerts(event_id,payload_sha256,canonical_json,runtime_profile_version,"
"runtime_profile_sha256,received_at_ns) VALUES(?,?,?,?,?,?)",
(event_id, payload_sha256, payload, PROFILE["profile_version"], PROFILE_SHA256, time.time_ns()),
)
assert_outbox_identity(descriptor, expected_metadata)
return payload_sha256, False
except HTTPException:
raise
except (OSError, PermissionError, sqlite3.Error) as exc:
raise HTTPException(503, "durable honey-event intake unavailable") from exc
finally:
if descriptor != -1:
os.close(descriptor)
@app.post("/v1/honey-events")
async def receive(
request: Request,
x_event_id: str = Header(),
x_timestamp: str = Header(),
x_signature: str = Header(),
):
raw = await bounded_body(request) # Bound the stream before authentication or JSON allocation.
if (not isinstance(x_event_id, str) or not x_event_id
or len(x_event_id.encode("utf-8")) > PROFILE["maximum_event_id_bytes"]
or EVENT_ID_PATTERN.fullmatch(x_event_id) is None):
raise HTTPException(401, "invalid event identifier")
if not x_timestamp.isascii() or not x_timestamp.isdigit() or len(x_timestamp) > 20:
raise HTTPException(401, "invalid authentication timestamp")
if len(x_signature) != 64 or any(character not in "0123456789abcdef" for character in x_signature):
raise HTTPException(401, "invalid authentication signature")
timestamp = int(x_timestamp, 10)
now = int(time.time())
if abs(now - timestamp) > PROFILE["maximum_skew_seconds"]:
raise HTTPException(401, "stale or future-dated event")
expected = hmac.new(
SECRET, x_timestamp.encode("ascii") + b"." + x_event_id.encode("utf-8") + b"." + raw,
hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(expected, x_signature):
raise HTTPException(401, "signature mismatch")
try:
event = strict_json(raw)
except (ValueError, UnicodeError, json.JSONDecodeError) as exc:
raise HTTPException(400, "event is not strict JSON") from exc
if not isinstance(event, dict) or set(event) != EVENT_FIELDS:
raise HTTPException(400, "event schema differs")
if event["schema_version"] != "aidefend.honey-event.v1":
raise HTTPException(400, "unsupported honey-event schema")
if type(event["triggered_at"]) is not int or event["triggered_at"] != timestamp:
raise HTTPException(400, "event timestamp is not bound to the authenticated envelope")
for field in ("canary_id", "placement_id", "source"):
bounded_identity(event[field], field)
record = {
"schema_version": "aidefend.dv002.authenticated-honey-intake.v1",
"event_id": x_event_id,
"runtime_profile_version": PROFILE["profile_version"],
"runtime_profile_sha256": PROFILE_SHA256,
"hmac_key_id": PROFILE["hmac_key_id"],
"raw_body_sha256": hashlib.sha256(raw).hexdigest(),
"event": event,
}
payload_sha256, duplicate = commit(x_event_id, record)
return {
"accepted_event_id": x_event_id,
"duplicate": duplicate,
"outbox_payload_sha256": payload_sha256,
"runtime_profile_version": PROFILE["profile_version"],
"runtime_profile_sha256": PROFILE_SHA256,
}Use the provider's official signature scheme instead of this HMAC envelope when one exists. A least-privilege forwarder verifies the canary/placement against the signed registry, enriches only with trusted inventory data, writes the incident event to the SIEM using event_id as idempotency key, verifies the acknowledgement, then independently reads it back before marking the outbox row delivered. Unknown canaries and invalid signatures become integrity alerts and are never treated as clean.
Verify the result
Replay valid, tampered, stale, duplicate, unknown-canary, SIEM-outage, and recovery fixtures. Retain raw-body digest, signature key/provider identity, event/canary/placement IDs, registry digest, outbox row, transport acknowledgement, SIEM event ID, readback, timing, and incident-routing receipt.
Action: Acknowledge only after durable authenticated intake, deliver idempotently, and retain independent SIEM/incident readback for the complete event population.
The intake profile and HMAC secret are fixed privileged artifacts, not paired environment values. Duplicate event IDs are idempotent only when the canonical authenticated record is byte-identical; a different payload returns a conflict and is never acknowledged as a clean duplicate. The text, body, identity, clock-skew, path, and SQLite-timeout ceilings in this example are implementation capacity guards, not organizational policy defaults; a deployment that genuinely needs larger inputs must deliberately revise and load-test the implementation instead of accepting an unbounded signed profile.
Open-source tools
- Canarytokens
- Faker
- Mimesis
Source-available / open-weight tools
- SDV (Business Source License 1.1; source-available)
Commercial / hosted tools
- Thinkst Canary
- Acalvio ShadowPlex
- Proofpoint Identity Threat Defense
Threat mappings
MITRE ATLAS
- AML.T0002.002 Acquire Public AI Artifacts: AI Agent Configuration (canary config artifacts detect acquisition)
- AML.T0025 Exfiltration via Cyber Means (honey data/canaries trigger alerts on exfiltration)
- AML.T0057 LLM Data Leakage (honey records and trackable URLs reveal disclosure from connected data stores)
- AML.T0035 AI Artifact Collection (decoy artifacts detect unauthorized artifact collection)
- AML.T0005 Create Proxy AI Model (decoy data and model artifacts reveal gathering for proxy training)
- AML.T0005.000 Create Proxy AI Model: Train Proxy via Gathered AI Artifacts (canaries in artifact stores detect gathering for proxy training)
- AML.T0064 Gather RAG-Indexed Targets
- AML.T0095.000 Search Open Websites/Domains: Code Repositories (repository canaries detect public-code reconnaissance)
- AML.T0055 Unsecured Credentials (seeded non-authoritative credentials alert on discovery or attempted use)
- AML.T0083 Credentials from AI Agent Configuration (canary credentials in synthetic agent configurations expose harvesting)
- AML.T0098 AI Agent Tool Credential Harvesting (canary credentials in agent-accessible stores expose harvesting attempts)
- AML.T0075 Enterprise Resource Discovery (decoy artifacts and canary records alert when repositories or file stores are browsed)
- AML.T0126 Automated Collection (canary records inside datasets and repositories are swept up by automated collection and alert)
MAESTRO
- Data Exfiltration (L2) (honey data triggers alert on exfiltration)
- Compromised Agents (L7) (honey tokens detect compromised agent data access)
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure (honey records trigger attributable evidence when sensitive-looking data is exported or used)
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI03:2026 Identity and Privilege Abuse (non-authoritative canary credentials expose agent credential harvesting or attempted reuse)
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- AITech-14.1 Unauthorized Access (detects use of a seeded canary cloud credential)
- AITech-8.2 Data Exfiltration / Exposure (exposure signals for seeded honey data and AI artifacts)
- AITech-8.3 Information Disclosure (exposure signals for seeded honey data and AI artifacts)
Google Secure AI Framework 2.0 - Risks
- MXF: Model Exfiltration (canary tokens detect model and data exfiltration)
- SDD: Sensitive Data Disclosure (honey data mimicking sensitive information triggers an attributable disclosure alert)
Databricks AI Security Framework 3.0
- Model 7.2: Model assets leak (decoy artifacts detect model asset exfiltration)
- Agents - Tools MCP Server 13.19: Credential and Token Exposure (canary API keys detect credential harvesting)
- Agents - Tools MCP Server 13.23: Data Exfiltration (canary tokens detect data exfiltration)