AID-D-003.002: Sensitive Information & Data Leakage Detection
Detect confidential data such as PII, secrets, source code, internal project names, and private tickets in model output and emit versioned findings before delivery or broad logging.
Pillars: data, app
Lifecycle: operation
Scope boundary
Owns findings for explicit sensitive information, secrets, memorized data, and direct confidential-data leakage in model output. It does not redact, block, or release output.
AID-D-003.007: AID-D-003.002 detects explicit sensitive values or memorized confidential content in output; AID-D-003.007 detects protected attributes inferred from combinations of otherwise non-sensitive facts. Direct leakage can be found by entity, pattern, or memorization evidence, while inferential privacy harm may contain no sensitive token from the source data.
AID-H-006: AID-D-003.002 emits sensitive-output findings before delivery or broad logging; AID-H-006 consumes output findings and owns deterministic sanitization, denial, schema enforcement, and release behavior. Detection identifies the risk, while Harden controls whether any bytes may leave the enforcement point.
Implementation guidance
- AID-D-003.002-G001: Detect common sensitive formats (PII, secrets, credentials) using pattern matching.
- AID-D-003.002-G002: Use Named Entity Recognition (NER) and Presidio to emit structured unstructured-PII findings.
- AID-D-003.002-G003: Detect direct memorization / regurgitation of training data.
- AID-D-003.002-G004: Detect organization-specific secrets and internal identifiers.
- AID-D-003.002-G005: Inspect streaming model output with bounded overlap and emit findings before the separate sink gate decides whether bytes may be released.
AID-D-003.002-G001
Detect common sensitive formats (PII, secrets, credentials) using pattern matching.
Pin the deterministic scanner contract
Deploy this reusable module with a code digest and RULESET_REVISION named by the signed output-DLP policy. Regex findings are defense-in-depth candidates: payment-card matches require Luhn validation, and redaction or denial remains owned by AID-H-006. Use keyed references for matched secrets; a bare hash of an SSN or card number is reversible by enumeration.
Scan bounded output and emit a bound result
from __future__ import annotations
# File: output_filters/deterministic_pii.py
import hashlib
import hmac
import re
from dataclasses import asdict, dataclass
RULESET_REVISION = "aidefend.output-pii-regex.2026-07"
PATTERNS = {
"CREDIT_CARD": re.compile(r"\b(?:[0-9][ -]?){12,18}[0-9]\b"),
"US_SSN": re.compile(r"\b(?!000|666|9[0-9]{2})[0-9]{3}-(?!00)[0-9]{2}-(?!0000)[0-9]{4}\b"),
"AWS_ACCESS_KEY": re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"),
}
@dataclass(frozen=True)
class DlpFinding:
rule_id: str
match_hmac_sha256: str
start: int
end: int
def _luhn_valid(value: str) -> bool:
digits = [int(character) for character in value if "0" <= character <= "9"]
if not 13 <= len(digits) <= 19:
return False
checksum = 0
parity = len(digits) % 2
for index, digit in enumerate(digits):
if index % 2 == parity:
digit *= 2
digit = digit - 9 if digit > 9 else digit
checksum += digit
return checksum % 10 == 0
def _secret_ref(key: bytes, value: str) -> str:
return hmac.new(key, value.encode("utf-8"), hashlib.sha256).hexdigest()
def scan_output(
*,
response_id: str,
text: str,
evidence_hmac_key: bytes,
policy_version: str,
maximum_characters: int,
) -> dict:
if not response_id or not policy_version:
raise ValueError("response or policy binding is absent")
if not isinstance(text, str) or not text:
raise ValueError("output text is absent")
if isinstance(maximum_characters, bool) or not isinstance(maximum_characters, int) or maximum_characters < 1:
raise ValueError("signed maximum_characters is invalid")
if len(text) > maximum_characters:
raise ValueError("output exceeds the signed scanner bound")
if not isinstance(evidence_hmac_key, bytes) or len(evidence_hmac_key) < 32:
raise ValueError("evidence HMAC key must contain at least 256 bits")
findings: list[DlpFinding] = []
for rule_id, pattern in PATTERNS.items():
for match in pattern.finditer(text):
if rule_id == "CREDIT_CARD" and not _luhn_valid(match.group(0)):
continue
findings.append(DlpFinding(
rule_id=rule_id,
match_hmac_sha256=_secret_ref(evidence_hmac_key, match.group(0)),
start=match.start(),
end=match.end(),
))
findings.sort(key=lambda item: (item.start, item.end, item.rule_id))
return {
"schema_version": "aidefend.output_dlp_result.v2",
"control_status": "PASS",
"finding_status": "FINDING" if findings else "NO_FINDING",
"response_id": response_id,
"content_hmac_sha256": _secret_ref(evidence_hmac_key, text),
"policy_version": policy_version,
"ruleset_revision": RULESET_REVISION,
"evaluated_character_count": len(text),
"findings": [asdict(item) for item in findings],
}
def safe_scan(**kwargs: object) -> dict:
try:
return scan_output(**kwargs)
except Exception as error:
return {
"schema_version": "aidefend.output_dlp_result.v2",
"control_status": "ERROR",
"finding_status": "UNKNOWN",
"response_id": str(kwargs.get("response_id", "")),
"policy_version": str(kwargs.get("policy_version", "")),
"ruleset_revision": RULESET_REVISION,
"error_class": type(error).__name__,
}Verification and boundary
The pre-delivery adapter invokes safe_scan exactly once for every applicable response and treats ERROR as unavailable detection, not NO_FINDING. Replay format-valid and invalid cards, SSNs, access keys, Unicode digit lookalikes, overlaps, maximum-size, and scanner-error fixtures. Independently recompute the response HMAC and match spans from retained authorized fixtures. Send the bound finding to AID-H-006; this Detect module does not redact or release content.
AID-D-003.002-G002
Use Named Entity Recognition (NER) and Presidio to emit structured unstructured-PII findings.
Configure a pinned Presidio profile
Presidio supports configurable recognizers and NLP engines; do not rely on whichever default model happens to be installed. Build one signed deployment profile per supported language that names the recognizer set, NLP model/container revision, entity list, score threshold, and maximum text size. The profile-verification adapter returns the typed profile and analyzer only after signature, schema, recognizer-registry, NLP-model, and configuration-digest verification.
Emit typed findings and preserve failures
from __future__ import annotations
# File: output_detectors/presidio_finding.py
import hashlib
import hmac
import math
import re
from dataclasses import dataclass
from presidio_analyzer import AnalyzerEngine
@dataclass(frozen=True)
class VerifiedPresidioProfile:
policy_version: str
detector_revision: str
analyzer_configuration_sha256: str
language: str
entities: tuple[str, ...]
score_threshold: float
maximum_characters: int
@dataclass(frozen=True)
class VerifiedAnalyzer:
engine: AnalyzerEngine
detector_revision: str
configuration_sha256: str
def _hmac_ref(key: bytes, value: str) -> str:
return hmac.new(key, value.encode("utf-8"), hashlib.sha256).hexdigest()
def _is_sha256(value: object) -> bool:
return (
isinstance(value, str)
and re.fullmatch(r"[0-9a-f]{64}", value) is not None
)
def analyze_output(
*,
analyzer: VerifiedAnalyzer,
profile: VerifiedPresidioProfile,
request_language: str,
response_id: str,
text: str,
evidence_hmac_key: bytes,
) -> dict:
if not response_id or not profile.language or request_language != profile.language:
raise ValueError("response identity or signed language route is invalid")
if (
not profile.policy_version
or not profile.detector_revision
or analyzer.detector_revision != profile.detector_revision
or not _is_sha256(profile.analyzer_configuration_sha256)
or not _is_sha256(analyzer.configuration_sha256)
or analyzer.configuration_sha256 != profile.analyzer_configuration_sha256
):
raise ValueError("Presidio analyzer differs from the verified profile")
if (
not profile.entities
or len(profile.entities) != len(set(profile.entities))
or any(not isinstance(item, str) or not item for item in profile.entities)
):
raise ValueError("signed entity set is empty or malformed")
if (
isinstance(profile.score_threshold, bool)
or not isinstance(profile.score_threshold, (int, float))
):
raise ValueError("signed score threshold must be numeric")
threshold = float(profile.score_threshold)
if not math.isfinite(threshold) or not 0.0 <= threshold <= 1.0:
raise ValueError("signed score threshold is invalid")
if (
isinstance(profile.maximum_characters, bool)
or not isinstance(profile.maximum_characters, int)
or profile.maximum_characters < 1
or not isinstance(text, str)
or not text
or len(text) > profile.maximum_characters
):
raise ValueError("output is absent or outside the signed size bound")
if not isinstance(evidence_hmac_key, bytes) or len(evidence_hmac_key) < 32:
raise ValueError("evidence HMAC key must contain at least 256 bits")
raw = analyzer.engine.analyze(
text=text,
language=profile.language,
entities=list(profile.entities),
score_threshold=threshold,
)
findings = []
for result in raw:
score = float(result.score)
start, end = int(result.start), int(result.end)
if (
not math.isfinite(score)
or not 0.0 <= score <= 1.0
or not 0 <= start < end <= len(text)
or result.entity_type not in profile.entities
):
raise ValueError("Presidio returned an invalid or unconfigured finding")
findings.append({
"entity_type": result.entity_type,
"start": start,
"end": end,
"score": round(score, 6),
})
findings.sort(key=lambda item: (item["start"], item["end"], item["entity_type"]))
return {
"schema_version": "aidefend.output_pii_finding.v2",
"control_status": "PASS",
"finding_status": "FINDING" if findings else "NO_FINDING",
"response_id": response_id,
"content_hmac_sha256": _hmac_ref(evidence_hmac_key, text),
"policy_version": profile.policy_version,
"detector_revision": profile.detector_revision,
"analyzer_configuration_sha256": analyzer.configuration_sha256,
"language": profile.language,
"evaluated_character_count": len(text),
"findings": findings,
}
def safe_analyze(**kwargs: object) -> dict:
try:
return analyze_output(**kwargs)
except Exception as error:
profile = kwargs.get("profile")
return {
"schema_version": "aidefend.output_pii_finding.v2",
"control_status": "ERROR",
"finding_status": "UNKNOWN",
"response_id": str(kwargs.get("response_id", "")),
"policy_version": getattr(profile, "policy_version", ""),
"detector_revision": getattr(profile, "detector_revision", ""),
"error_class": type(error).__name__,
}Verify each supported language
Pin the Presidio package, NLP engine, model, and recognizer registry in the deployment manifest, and construct VerifiedAnalyzer from that verified configuration; never pass an independently created engine whose configuration digest cannot be compared to the profile. Route only explicitly supported languages; an absent or unsupported language produces ERROR/UNKNOWN. Replay signed positive, negative, domain-specific, multilingual, threshold-boundary, malformed-span, model-outage, and oversized fixtures, then compare spans and the keyed content binding. Redaction and delivery decisions remain in AID-H-006.
AID-D-003.002-G003
Detect direct memorization / regurgitation of training data.
Build the index from the verified source
Run the builder under a dedicated deployment identity. The source path, Sigstore bundle, verification key, timeout, and output path must come from a root-owned service definition, not request data. The builder captures the source and bundle once, verifies those private bytes, validates the complete JSONL population, and publishes a canonical index with an atomic rename.
from __future__ import annotations
# File: monitoring/leakage_index.py
import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path
BUILDER_VERSION = "d003002-leakage-index-v1"
def write_private(path: Path, payload: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.chmod(path, 0o400)
def strict_json(raw: bytes) -> object:
def unique_object(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key: " + key)
value[key] = item
return value
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=lambda token: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + token)
),
)
def canonical_json_bytes(value: object) -> bytes:
return (
json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+ "\n"
).encode("utf-8")
def capture_verified_source(
source: Path,
bundle: Path,
key_reference: str,
timeout_seconds: float,
) -> bytes:
if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
raise ValueError("signature-verification timeout is invalid")
source = Path(source)
bundle = Path(bundle)
if source == bundle:
raise ValueError("source and bundle paths collide")
source_raw = source.read_bytes()
bundle_raw = bundle.read_bytes()
remote_key = bool(re.match(
r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
)) or key_reference.startswith("pkcs11:")
key_raw = None if remote_key else (
key_reference.encode("utf-8")
if key_reference.startswith("-----BEGIN ")
else Path(key_reference).read_bytes()
)
with tempfile.TemporaryDirectory(prefix="aidefend-d003002-source-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
source_snapshot = root / "training-sentences.jsonl"
bundle_snapshot = root / "training-sentences.bundle"
write_private(source_snapshot, source_raw)
write_private(bundle_snapshot, bundle_raw)
captured = [(source_snapshot, source_raw), (bundle_snapshot, bundle_raw)]
key_argument = key_reference
if key_raw is not None:
key_snapshot = root / "verify-key.pub"
write_private(key_snapshot, key_raw)
captured.append((key_snapshot, key_raw))
key_argument = str(key_snapshot)
try:
subprocess.run(
[
"cosign", "verify-blob", "--key", key_argument,
"--bundle", str(bundle_snapshot), str(source_snapshot),
],
check=True,
capture_output=True,
text=True,
timeout=timeout_seconds,
)
if any(
not hmac.compare_digest(candidate.read_bytes(), expected)
for candidate, expected in captured
):
raise RuntimeError("private verification snapshot changed")
finally:
for candidate, _ in captured:
os.chmod(candidate, 0o600)
return source_raw
def validate_records(source_raw: bytes) -> list[dict]:
lines = source_raw.decode("utf-8", errors="strict").splitlines()
if not lines or any(not line.strip() for line in lines):
raise ValueError("verified training-sentence population is empty or malformed")
records = [strict_json(line.encode("utf-8")) for line in lines]
if any(
not isinstance(record, dict)
or set(record) != {"sentence_id", "text"}
or not isinstance(record["sentence_id"], str)
or not record["sentence_id"]
or not isinstance(record["text"], str)
or not record["text"].strip()
for record in records
):
raise ValueError("training-sentence schema differs")
sentence_ids = [record["sentence_id"] for record in records]
if len(sentence_ids) != len(set(sentence_ids)):
raise ValueError("duplicate training-sentence ID")
return records
def build_index_document(source_raw: bytes) -> dict:
records = validate_records(source_raw)
phrases = sorted({
record["text"] for record in records
if len(record["text"].split()) > 10
})
if not phrases:
raise ValueError("verified source contains no eligible long phrase")
return {
"schema_version": "aidefend.leakage_index.v1",
"builder_version": BUILDER_VERSION,
"source_sha256": hashlib.sha256(source_raw).hexdigest(),
"source_record_count": len(records),
"indexed_phrase_count": len(phrases),
"phrases": {phrase: [phrase] for phrase in phrases},
}
def atomic_publish(path: Path, payload: bytes) -> None:
requested = Path(path)
parent = requested.parent.resolve(strict=True)
if not parent.is_dir() or parent.stat().st_mode & 0o022:
raise PermissionError("index directory must be an existing non-group-writable directory")
target = parent / requested.name
if target.exists() and (target.is_symlink() or not target.is_file()):
raise ValueError("existing index target is not a regular file")
descriptor, temporary_name = tempfile.mkstemp(
prefix="." + target.name + ".", dir=parent
)
temporary = Path(temporary_name)
try:
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.chmod(temporary, 0o400)
os.replace(temporary, target)
directory_fd = os.open(parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
if not hmac.compare_digest(target.read_bytes(), payload):
raise RuntimeError("published index readback differs")
finally:
temporary.unlink(missing_ok=True)
def main() -> None:
source = Path(os.environ["TRAINING_SENTENCES_JSONL"])
bundle = Path(os.environ["TRAINING_SENTENCES_SIGSTORE_BUNDLE"])
key = os.environ["TRAINING_SENTENCES_VERIFY_KEY"]
timeout = float(os.environ["D003002_VERIFY_TIMEOUT_SECONDS"])
output = Path(os.environ["LEAKAGE_INDEX_PATH"])
source_raw = capture_verified_source(source, bundle, key, timeout)
atomic_publish(output, canonical_json_bytes(build_index_document(source_raw)))
if __name__ == "__main__":
main()Load only an index reproducible from the signed source
At detector startup, independently verify and capture the source again, rebuild the expected canonical bytes, and compare them with the deployed index. This makes a stale, partially written, or substituted unsigned derivative unusable even if an attacker can edit its path.
from __future__ import annotations
# File: output_filters/leakage_detector.py
import hashlib
import hmac
import json
import math
import os
from pathlib import Path
from typing import Literal
from flashtext import KeywordProcessor
from monitoring.leakage_index import (
build_index_document,
canonical_json_bytes,
capture_verified_source,
strict_json,
)
ControlStatus = Literal["PASS", "ERROR"]
FindingStatus = Literal["FINDING", "NO_FINDING", "UNKNOWN"]
SOURCE = Path(os.environ["TRAINING_SENTENCES_JSONL"])
SOURCE_BUNDLE = Path(os.environ["TRAINING_SENTENCES_SIGSTORE_BUNDLE"])
SOURCE_KEY = os.environ["TRAINING_SENTENCES_VERIFY_KEY"]
INDEX_PATH = Path(os.environ["LEAKAGE_INDEX_PATH"])
VERIFY_TIMEOUT_SECONDS = float(os.environ["D003002_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(VERIFY_TIMEOUT_SECONDS) or VERIFY_TIMEOUT_SECONDS <= 0:
raise RuntimeError("signature-verification timeout is invalid")
source_raw = capture_verified_source(
SOURCE, SOURCE_BUNDLE, SOURCE_KEY, VERIFY_TIMEOUT_SECONDS
)
expected_document = build_index_document(source_raw)
expected_index_raw = canonical_json_bytes(expected_document)
index_raw = INDEX_PATH.read_bytes()
if not hmac.compare_digest(index_raw, expected_index_raw):
raise RuntimeError("deployed index is not the canonical derivative of signed source")
index_document = strict_json(index_raw)
if index_document != expected_document:
raise RuntimeError("deployed leakage-index schema or binding differs")
INDEX_SHA256 = hashlib.sha256(index_raw).hexdigest()
leakage_detector = KeywordProcessor()
leakage_detector.add_keywords_from_dict(index_document["phrases"])
def detect_training_data_leakage(response_id: str, text: str) -> dict:
if not isinstance(response_id, str) or not response_id:
raise ValueError("response_id is required")
if not isinstance(text, str):
raise TypeError("text must be a string")
response_sha256 = hashlib.sha256(text.encode("utf-8")).hexdigest()
matches = sorted(set(leakage_detector.extract_keywords(text)))
match_sha256 = sorted({
hashlib.sha256(value.encode("utf-8")).hexdigest() for value in matches
})
return {
"schema_version": "aidefend.training_leakage_finding.v1",
"control_status": "PASS",
"finding_status": "FINDING" if matches else "NO_FINDING",
"response_id": response_id,
"response_sha256": response_sha256,
"source_sha256": index_document["source_sha256"],
"index_sha256": INDEX_SHA256,
"builder_version": index_document["builder_version"],
"matched_phrase_count": len(matches),
"matched_phrase_sha256": match_sha256,
}
def safe_detect_training_data_leakage(response_id: str, text: str) -> dict:
try:
return detect_training_data_leakage(response_id, text)
except Exception as error:
value = text if isinstance(text, str) else ""
return {
"schema_version": "aidefend.training_leakage_finding.v1",
"control_status": "ERROR",
"finding_status": "UNKNOWN",
"response_id": response_id,
"response_sha256": hashlib.sha256(value.encode("utf-8")).hexdigest(),
"source_sha256": index_document["source_sha256"],
"index_sha256": INDEX_SHA256,
"error_class": type(error).__name__,
}Exercise and operate the detector
Before promotion, replay an exact long memorized passage, benign overlap, a short phrase below the indexing rule, Unicode text, empty output, corrupt JSONL, duplicate IDs, an empty eligible population, a substituted source or bundle, a changed index byte, and an interrupted builder write. Recompute the canonical index and finding digests independently. Run the detector over every applicable model response and retain its result in the authenticated event path; a startup verification error or missing response result is ERROR/unknown, never a clean result. Send findings to AID-H-006 for release enforcement; this detector does not block or redact.
AID-D-003.002-G004
Detect organization-specific secrets and internal identifiers.
Versioned tenant rule pack
Keep organization-specific literals and regular expressions in a tenant-scoped policy that is signed outside the model-serving identity. The numeric bound below is part of this example policy version; production values must come from the organization's signed policy, not from application defaults.
config/proprietary-patterns.v7.json
{
"schema_version": "aidefend.proprietary-pattern-policy.v1",
"tenant_id": "tenant-example",
"policy_version": "2026-07-16.7",
"normalizer": "NFKC_CASEFOLD",
"maximum_pattern_length": 256,
"literal_rules": [
{"rule_id": "internal-project-chimera", "value": "Project Chimera"},
{"rule_id": "internal-forecast-file", "value": "Q3-financial-forecast.xlsx"}
],
"regex_rules": [
{"rule_id": "internal-ticket-id", "pattern": "JIRA-[A-Z]+-[0-9]+"},
{"rule_id": "internal-prod-host", "pattern": "[a-z]{3}-[a-z]+-prod-[0-9]{2}"}
]
}Fail-closed detector with signed policy and signed evidence
output_filters/proprietary_filter.py
from __future__ import annotations
import base64
import hashlib
import json
import unicodedata
from dataclasses import dataclass
from typing import Any
import re2
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey,
Ed25519PublicKey,
)
@dataclass(frozen=True)
class RuleMatch:
rule_id: str
kind: str
normalized_start: int
normalized_end: int
def _canonical(value: dict[str, Any]) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
).encode("utf-8")
def _b64decode(value: str) -> bytes:
return base64.b64decode(value.encode("ascii"), validate=True)
class ProprietaryFilter:
REQUIRED_POLICY_KEYS = {
"schema_version", "tenant_id", "policy_version", "normalizer",
"maximum_pattern_length", "literal_rules", "regex_rules",
}
def __init__(
self,
*,
signed_policy_envelope: bytes,
expected_policy_key_id: str,
policy_verify_key_raw: bytes,
evidence_key_id: str,
evidence_signing_key: Ed25519PrivateKey,
) -> None:
envelope = json.loads(signed_policy_envelope.decode("utf-8", errors="strict"))
if set(envelope) != {"key_id", "payload_b64", "signature_b64"}:
raise ValueError("policy envelope schema mismatch")
if envelope["key_id"] != expected_policy_key_id:
raise ValueError("unexpected policy signing key")
payload = _b64decode(envelope["payload_b64"])
signature = _b64decode(envelope["signature_b64"])
try:
Ed25519PublicKey.from_public_bytes(policy_verify_key_raw).verify(
signature, payload
)
except InvalidSignature as exc:
raise ValueError("policy signature verification failed") from exc
policy = json.loads(payload.decode("utf-8", errors="strict"))
if set(policy) != self.REQUIRED_POLICY_KEYS:
raise ValueError("policy schema mismatch")
if (
policy["schema_version"] != "aidefend.proprietary-pattern-policy.v1"
or policy["normalizer"] != "NFKC_CASEFOLD"
or not isinstance(policy["tenant_id"], str)
or not policy["tenant_id"]
or not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
):
raise ValueError("invalid policy binding")
maximum = int(policy["maximum_pattern_length"])
if maximum < 1:
raise ValueError("maximum_pattern_length must be positive")
self.tenant_id = policy["tenant_id"]
self.policy_version = policy["policy_version"]
self.policy_sha256 = hashlib.sha256(payload).hexdigest()
self.evidence_key_id = evidence_key_id
self.evidence_signing_key = evidence_signing_key
self.literals: list[tuple[str, str]] = []
self.regexes: list[tuple[str, Any]] = []
seen: set[str] = set()
for item in policy["literal_rules"]:
if not isinstance(item, dict) or set(item) != {"rule_id", "value"}:
raise ValueError("literal rule schema mismatch")
rule_id, value = str(item["rule_id"]), str(item["value"])
if not rule_id or not value or rule_id in seen or len(value) > maximum:
raise ValueError("invalid or duplicate literal rule")
seen.add(rule_id)
self.literals.append((rule_id, self._normalize(value)))
for item in policy["regex_rules"]:
if not isinstance(item, dict) or set(item) != {"rule_id", "pattern"}:
raise ValueError("regex rule schema mismatch")
rule_id, pattern = str(item["rule_id"]), str(item["pattern"])
if not rule_id or not pattern or rule_id in seen or len(pattern) > maximum:
raise ValueError("invalid or duplicate regex rule")
seen.add(rule_id)
self.regexes.append((rule_id, re2.compile(pattern)))
@staticmethod
def _normalize(value: str) -> str:
return unicodedata.normalize("NFKC", value).casefold()
def _sign(self, result: dict[str, Any]) -> dict[str, Any]:
payload = _canonical(result)
return {
"payload_b64": base64.b64encode(payload).decode("ascii"),
"signature_b64": base64.b64encode(
self.evidence_signing_key.sign(payload)
).decode("ascii"),
"key_id": self.evidence_key_id,
}
def scan(self, *, response_id: str, tenant_id: str, text: str) -> dict[str, Any]:
response_sha256 = hashlib.sha256(text.encode("utf-8")).hexdigest()
if tenant_id != self.tenant_id:
result = {
"schema_version": "aidefend.proprietary-pattern-result.v1",
"control_status": "ERROR",
"finding_status": "UNKNOWN",
"reason": "tenant_policy_binding_mismatch",
"response_id": response_id,
"response_sha256": response_sha256,
"policy_sha256": self.policy_sha256,
}
return {"result": result, "signed_evidence": self._sign(result)}
normalized = self._normalize(text)
matches: list[RuleMatch] = []
for rule_id, literal in self.literals:
start = 0
while True:
offset = normalized.find(literal, start)
if offset < 0:
break
matches.append(
RuleMatch(rule_id, "literal", offset, offset + len(literal))
)
start = offset + max(1, len(literal))
for rule_id, pattern in self.regexes:
matches.extend(
RuleMatch(rule_id, "regex", match.start(), match.end())
for match in pattern.finditer(normalized)
)
ordered = sorted(
matches,
key=lambda item: (
item.normalized_start, item.normalized_end, item.rule_id
),
)
result = {
"schema_version": "aidefend.proprietary-pattern-result.v1",
"control_status": "PASS",
"finding_status": "FINDING" if ordered else "NO_FINDING",
"response_id": response_id,
"response_sha256": response_sha256,
"tenant_id": tenant_id,
"policy_version": self.policy_version,
"policy_sha256": self.policy_sha256,
"matches": [match.__dict__ for match in ordered],
}
return {"result": result, "signed_evidence": self._sign(result)}Verification: Build and sign the policy in an identity isolated from serving. In CI and staging, verify bad signatures, wrong tenants, duplicate rule IDs, oversized rules, malformed UTF-8, Unicode normalization, regex boundaries, cross-chunk reassembly, benign near-matches, rotated policies, evidence-signature readback, and both finding states. Run the independent verifier with a different identity and public key.
Action: Instantiate this detector only from a verified tenant policy; scan every applicable canonical output field before release; write the signed receipt to append-only evidence storage; and pass the structured finding to AID-H-006. A policy, tenant, parser, regex, or evidence failure is ERROR, never NO_FINDING.
AID-D-003.002-G005
Inspect streaming model output with bounded overlap and emit findings before the separate sink gate decides whether bytes may be released.
Derive overlap from the pinned scanner
A streaming detector must know the maximum match length of its active signed rule set. Use at least maximum_match_characters - 1 characters of overlap; a generic constant can miss a valid 19-digit card with separators. Bind chunks to one response, require a gap-free sequence, convert window-relative spans to global response offsets, de-duplicate overlap findings, and emit a terminal receipt.
Reusable ordered-stream detector
from __future__ import annotations
# File: output_detectors/streaming_dlp_detector.py
import hashlib
import hmac
import re
from dataclasses import asdict
from typing import Protocol
from output_filters.deterministic_pii import DlpFinding
SHA256 = re.compile(r"^[0-9a-f]{64}$")
def _is_sha256(value: object) -> bool:
return isinstance(value, str) and SHA256.fullmatch(value) is not None
class Scanner(Protocol):
revision: str
maximum_match_characters: int
def __call__(self, text: str) -> list[DlpFinding]: ...
class StreamingDlpDetector:
def __init__(
self,
*,
response_id: str,
policy_version: str,
scanner: Scanner,
evidence_hmac_key: bytes,
maximum_response_characters: int,
) -> None:
if (
not response_id
or not policy_version
or not isinstance(scanner.revision, str)
or not scanner.revision
):
raise ValueError("response, policy, or scanner binding is absent")
if (
isinstance(scanner.maximum_match_characters, bool)
or not isinstance(scanner.maximum_match_characters, int)
or scanner.maximum_match_characters < 1
):
raise ValueError("scanner maximum match span is invalid")
if (
isinstance(maximum_response_characters, bool)
or not isinstance(maximum_response_characters, int)
or maximum_response_characters < scanner.maximum_match_characters
):
raise ValueError("signed response bound is invalid")
if not isinstance(evidence_hmac_key, bytes) or len(evidence_hmac_key) < 32:
raise ValueError("evidence HMAC key must contain at least 256 bits")
self.response_id = response_id
self.policy_version = policy_version
self.scanner = scanner
self.key = evidence_hmac_key
self.maximum = maximum_response_characters
self.overlap = scanner.maximum_match_characters - 1
self.tail = ""
self.total_characters = 0
self.next_chunk_index = 0
self.seen: set[tuple[str, int, int, str]] = set()
self.content_hmac = hmac.new(self.key, digestmod=hashlib.sha256)
self.failed = False
self.finished = False
def inspect_chunk(self, *, chunk_index: int, chunk: str) -> dict:
if self.failed or self.finished:
raise RuntimeError("stream is already terminal")
if chunk_index != self.next_chunk_index or not isinstance(chunk, str):
self.failed = True
raise ValueError("chunk is non-text, duplicated, reordered, or missing")
if self.total_characters + len(chunk) > self.maximum:
self.failed = True
raise ValueError("stream exceeds the signed size bound")
window = self.tail + chunk
window_start = self.total_characters - len(self.tail)
try:
raw_findings = self.scanner(window)
except Exception:
self.failed = True
raise
emitted = []
for item in raw_findings:
start = window_start + int(item.start)
end = window_start + int(item.end)
key = (item.rule_id, start, end, item.match_hmac_sha256)
if not 0 <= start < end <= self.total_characters + len(chunk):
self.failed = True
raise ValueError("scanner returned an invalid span")
if key not in self.seen:
self.seen.add(key)
emitted.append({**asdict(item), "start": start, "end": end})
self.content_hmac.update(chunk.encode("utf-8"))
self.total_characters += len(chunk)
self.next_chunk_index += 1
self.tail = window[-self.overlap:] if self.overlap else ""
return {
"schema_version": "aidefend.streaming_dlp_chunk.v2",
"control_status": "PASS",
"finding_status": "FINDING" if emitted else "NO_FINDING",
"response_id": self.response_id,
"chunk_index": chunk_index,
"policy_version": self.policy_version,
"scanner_revision": self.scanner.revision,
"findings": emitted,
}
def finish(self, *, expected_chunk_count: int, expected_content_hmac_sha256: str) -> dict:
if self.failed or self.finished:
raise RuntimeError("stream is failed or already terminal")
if (
isinstance(expected_chunk_count, bool)
or not isinstance(expected_chunk_count, int)
or expected_chunk_count < 0
or expected_chunk_count != self.next_chunk_index
or not _is_sha256(expected_content_hmac_sha256)
):
self.failed = True
raise ValueError("response manifest count or content binding is invalid")
observed_hmac = self.content_hmac.hexdigest()
if not hmac.compare_digest(observed_hmac, expected_content_hmac_sha256):
self.failed = True
raise ValueError("stream content differs from the authenticated response manifest")
self.finished = True
return {
"schema_version": "aidefend.streaming_dlp_terminal.v2",
"control_status": "PASS",
"finding_status": "FINDING" if self.seen else "NO_FINDING",
"response_id": self.response_id,
"policy_version": self.policy_version,
"scanner_revision": self.scanner.revision,
"chunk_count": self.next_chunk_index,
"evaluated_character_count": self.total_characters,
"content_hmac_sha256": observed_hmac,
"finding_count": len(self.seen),
}Integration and verification
The response-stream adapter obtains expected chunk count and keyed content digest from the authenticated generator manifest, catches any constructor, chunk, scanner, or finalization exception, and emits one terminal ERROR/UNKNOWN receipt; cancellation without a terminal receipt is a coverage defect. Replay every split position for the longest configured rule, Unicode text, duplicates, reorder, truncation, cancellation, scanner error, and maximum-size overflow. AID-H-006 owns holdback, redaction, denial, and release.
Open-source tools
- Microsoft Presidio
- spaCy
- NLTK
- Hugging Face Transformers
- FlashText
- TruffleHog
- Gitleaks
- detect-secrets
Commercial / hosted tools
- Nightfall Firewall for AI
- Google Cloud Sensitive Data Protection
- Microsoft Purview Data Loss Prevention
- Tonic Textual
- Broadcom Symantec Data Loss Prevention
- Forcepoint DLP
- HiddenLayer AI Runtime Security
- Palo Alto Networks Prisma AIRS
Threat mappings
MITRE ATLAS
- AML.T0024 Exfiltration via AI Inference API
- AML.T0057 LLM Data Leakage
- AML.T0048.003 External Harms: User Harm
- AML.T0056 Extract LLM System Prompt (findings identify system-prompt content in outputs)
- AML.T0085 Data from AI Services (findings identify sensitive data returned through AI services)
- AML.T0085.000 Data from AI Services: RAG Databases (findings identify sensitive RAG content in responses)
- AML.T0085.001 Data from AI Services: AI Agent Tools (findings identify sensitive data returned by agent tools)
- AML.T0086 Exfiltration via AI Agent Tool Invocation (findings identify sensitive data in tool parameters or results)
- AML.T0082 RAG Credential Harvesting (findings identify credential content in outputs)
- AML.T0098 AI Agent Tool Credential Harvesting (findings identify credential exposure through tools)
- AML.T0069.002 Discover LLM System Information: System Prompt (findings identify leaked system-prompt content)
- AML.T0084.000 Discover AI Agent Configuration: Embedded Knowledge (findings identify leaked knowledge-source details)
MAESTRO
- Data Exfiltration (L2)
- Data Leakage through Observability (L5)
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI02:2026 Tool Misuse and Exploitation (DLP detects sensitive data exfiltrated through tool outputs)
- ASI03:2026 Identity and Privilege Abuse (data leakage detection catches credential/identity exposure)
NIST Adversarial Machine Learning 2025
- NISTAML.036 Leaking information from user interactions
- NISTAML.038 Data Extraction
- NISTAML.035 Prompt Extraction (findings identify system-prompt content in outputs)
Cisco Integrated AI Security and Safety Framework
- AISubtech-10.1.3 Sensitive Data Reconstruction
- AISubtech-14.1.1 Credential Theft (credential disclosure detected in generated output)
- AISubtech-15.1.24 Intellectual Property Compromise: Confidential Data
- AISubtech-8.2.1 Training Data Exposure
- AISubtech-8.2.2 LLM Data Leakage
- AISubtech-8.3.2 System Information Exposure (internal production hostnames recognized by the supplied patterns)
- AISubtech-8.3.3 Personally Identifiable Information Exposure
- AISubtech-8.3.4 Payment Card Industry Data Exposure
- AITech-15.1 Harmful Content
- AITech-8.2 Data Exfiltration / Exposure
- AITech-8.3 Information Disclosure
Google Secure AI Framework 2.0 - Risks
- SDD: Sensitive Data Disclosure
- EDH: Excessive Data Handling (DLP detects output of data beyond policy limits)
- ISD: Inferred Sensitive Data (detects disclosure of inferred sensitive information)
Databricks AI Security Framework 3.0
- Model Serving - Inference response 10.6: Sensitive data output from a model
- Agents - Tools MCP Server 13.23: Data Exfiltration