AID-H-026.003: Taxonomy-Stratified Coverage Gap Analysis
Measure detector performance by attack-technique category rather than relying solely on aggregate metrics. Break down results by a defined prompt-attack taxonomy such as role play, objective manipulation, obfuscation, indirect injection, encoding, multi-turn Crescendo-style escalation, many-shot pressure, policy-puppetry role or policy shifts, context saturation, or other organization-specific classes. The purpose is to expose category-level blind spots, prioritize the next hard-mining or fuzzing cycle, and supply independently measured coverage eligibility or a veto to the final detector-promotion authority. Category eligibility alone does not authorize a detector release.
Pillars: model, app
Lifecycle: validation, improvement
Scope boundary
Owns independently sealed taxonomy evaluation, complete raw category observations, coverage eligibility or veto findings, and hardening backlog priorities. It neither retrains the detector nor makes, signs, or executes the final release promotion decision.
AID-H-026.001: AID-H-026.003 authenticates the sealed evaluation population and measures category coverage; AID-H-026.001 produces candidates and owns final promotion, utility gates and deployment readback. ELIGIBLE states only that the measured coverage facet meets its profile; HOLD supplies a veto. Neither is a release authorization or duplicate deployment assurance.
AID-H-026.002: AID-H-026.003 measures taxonomy-stratified coverage; AID-H-026.002 generates adversarial variants for the categories that need expansion. The measurement and corpus-generation stages remain separately evidenced.
Implementation guidance
- AID-H-026.003-G001: Compute taxonomy-stratified metrics and convert weak categories into the next retraining backlog.
- AID-H-026.003-G002: Authenticate and evaluate the immutable out-of-loop population, then publish independently signed ELIGIBLE or HOLD coverage evidence for the final promotion gate.
- AID-H-026.003-G003: Use attribution methods to debug representative detector misses and guide the next hardening cycle.
AID-H-026.003-G001
Compute taxonomy-stratified metrics and convert weak categories into the next retraining backlog.
Purpose and metric boundary
Measure malicious detection recall separately for every required attack category and false-positive rate for the governed benign category. Do not report per-attack precision or F1 from an all-positive category slice: without negative examples in that slice those values are not meaningful. Promotion consumes raw counts, denominators, and category completeness rather than aggregate accuracy.
Step 1: Validate the complete taxonomy population and compute count-based metrics
# File: detector_hardening/category_report.py
from __future__ import annotations
import math
import numbers
from dataclasses import dataclass
import pandas as pd
EVAL_COLUMNS = {"case_id", "label", "pred", "category", "source"}
REPORT_COLUMNS = {
"category", "sample_count", "positive_count", "negative_count",
"true_positive_count", "false_positive_count", "detection_recall",
"false_positive_rate", "detector_version", "evaluation_set_id",
}
@dataclass(frozen=True)
class CategoryGatePolicy:
required_categories: tuple[str, ...]
benign_category: str
min_attack_recall: float
max_benign_fpr: float
max_recall_regression: float
def probability(value: object, name: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(name + " must be numeric")
result = float(value)
if not math.isfinite(result) or not 0.0 <= result <= 1.0:
raise ValueError(name + " must be finite and in [0,1]")
return result
def validate_gate_policy(policy: CategoryGatePolicy) -> None:
if (len(policy.required_categories) < 2
or len(set(policy.required_categories)) != len(policy.required_categories)
or any(not isinstance(value, str) or not value for value in policy.required_categories)
or policy.benign_category not in policy.required_categories):
raise ValueError("required category policy differs")
probability(policy.min_attack_recall, "min_attack_recall")
probability(policy.max_benign_fpr, "max_benign_fpr")
probability(policy.max_recall_regression, "max_recall_regression")
def stratified_report(
eval_df: pd.DataFrame,
detector_version: str,
evaluation_set_id: str,
policy: CategoryGatePolicy,
) -> pd.DataFrame:
validate_gate_policy(policy)
if not detector_version or not evaluation_set_id or set(eval_df.columns) != EVAL_COLUMNS:
raise ValueError("evaluation identity or schema differs")
if eval_df.empty or eval_df["case_id"].isna().any() or eval_df["case_id"].duplicated().any():
raise ValueError("evaluation population is empty or case IDs are invalid")
binary_values = eval_df["label"].tolist() + eval_df["pred"].tolist()
if any(isinstance(value, bool) or not isinstance(value, numbers.Integral)
or int(value) not in (0, 1) for value in binary_values):
raise ValueError("label or prediction must be a strict binary integer")
category_values = eval_df["category"].tolist()
if (any(not isinstance(value, str) or not value for value in category_values)
or set(category_values) != set(policy.required_categories)):
raise ValueError("evaluation population does not exactly cover required taxonomy")
rows: list[dict] = []
for category in policy.required_categories:
category_df = eval_df[eval_df["category"] == category]
if category_df.empty:
raise RuntimeError("required category population is missing")
labels = category_df["label"]
predictions = category_df["pred"]
if category == policy.benign_category:
if not (labels == 0).all():
raise ValueError("benign category contains malicious labels")
false_positives = int((predictions == 1).sum())
recall, fpr = None, false_positives / len(category_df)
true_positives = 0
else:
if not (labels == 1).all():
raise ValueError("attack category contains benign labels")
true_positives = int((predictions == 1).sum())
false_positives = 0
recall, fpr = true_positives / len(category_df), None
rows.append({
"category": category, "sample_count": len(category_df),
"positive_count": int((labels == 1).sum()),
"negative_count": int((labels == 0).sum()),
"true_positive_count": true_positives,
"false_positive_count": false_positives,
"detection_recall": recall, "false_positive_rate": fpr,
"detector_version": detector_version,
"evaluation_set_id": evaluation_set_id,
})
# Object dtype preserves deliberately undefined metrics as None (JSON null).
# Do not replace arbitrary NaN/Inf: canonical evidence must still reject them.
return pd.DataFrame(rows, columns=sorted(REPORT_COLUMNS), dtype=object)
Step 2: Compare identical category populations and emit a bounded hardening backlog
# File: detector_hardening/weak_category_gate.py
import pandas as pd
from detector_hardening.category_report import (
CategoryGatePolicy,
REPORT_COLUMNS,
probability,
validate_gate_policy,
)
def identify_weak_categories(
current_report: pd.DataFrame,
previous_report: pd.DataFrame,
policy: CategoryGatePolicy,
) -> list[dict]:
validate_gate_policy(policy)
for name, report in (("current", current_report),) + ((("previous", previous_report),) if previous_report is not None else ()):
if set(report.columns) != REPORT_COLUMNS or report["category"].duplicated().any():
raise ValueError(name + " report schema or category identity differs")
if set(report["category"]) != set(policy.required_categories):
raise ValueError(name + " report does not cover the signed taxonomy")
previous_by_category = None if previous_report is None else previous_report.set_index("category")
weak: list[dict] = []
for row in current_report.to_dict(orient="records"):
category = row["category"]
prior = None if previous_by_category is None else previous_by_category.loc[category]
if category == policy.benign_category:
value = probability(row["false_positive_rate"], "false_positive_rate")
if value > policy.max_benign_fpr:
weak.append({"category": category, "issue": "high_fpr",
"value": value, "threshold": policy.max_benign_fpr})
continue
current_recall = probability(row["detection_recall"], "detection_recall")
if current_recall < policy.min_attack_recall:
weak.append({"category": category, "issue": "low_recall",
"value": current_recall,
"threshold": policy.min_attack_recall})
previous_recall = None if prior is None else probability(prior["detection_recall"], "previous_recall")
regression = None if previous_recall is None else previous_recall - current_recall
if regression is not None and regression > policy.max_recall_regression:
weak.append({"category": category, "issue": "recall_regression",
"current": current_recall, "previous": previous_recall,
"delta": regression,
"threshold": policy.max_recall_regression})
return weak
Keep the action safely bounded
Reject incomplete or mixed-label category populations, calculate transparent count-based recall and benign FPR for every required category, compare the exact prior baseline, and emit a coverage veto when a taxonomy policy gate or category-completeness check fails. AID-H-026.001 owns the final release decision and deployment enforcement.
Verify complete coverage
Materialize the complete population of every required taxonomy category and every eligible evaluation case, raw prediction, count, denominator, prior-category comparison, weak-category finding, and backlog disposition before evaluation; require stable unique IDs and reject duplicates or undeclared members.
Verify safely
Write the receipt and exact replay inputs to an append-only or WORM sink under an identity that the detector and training job cannot alter, for the policy-derived retention period. A separately credentialed verifier rejoins case IDs to raw labels and predictions, recomputes every count and ratio, runs missing/extra/mixed-category negative fixtures, and confirms backlog and promotion consumers received the same report bytes; signature-only self-attestation is insufficient.
AID-H-026.003-G002
Authenticate and evaluate the immutable out-of-loop population, then publish independently signed ELIGIBLE or HOLD coverage evidence for the final promotion gate.
Sealed evaluation authority
Keep benchmark cases outside training, mining, fuzzing, tuning, attribution and developer access. Verify the independent holdout signature and its exact digest from the signed evaluation scope. This control produces measured taxonomy eligibility or a veto, not a release decision. AID-H-026.001 remains the final promotion and deployment owner.
Verify the exact sealed holdout and role-specific signed inputs
# File: detector_hardening/sealed_holdout.py
from __future__ import annotations
import hashlib
import json
import re
from pathlib import Path
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
SHA256 = re.compile(r"^[a-f0-9]{64}$")
MANIFEST_FIELDS = {"schema_version", "holdout_id", "version", "created_at", "samples"}
SAMPLE_FIELDS = {"case_id", "text", "label_int", "category", "source_ref"}
def canonical(value: object) -> bytes:
return (json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False) + "\n").encode("utf-8")
def strict_json(payload: bytes) -> object:
def unique(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(payload, object_pairs_hook=unique,
parse_constant=lambda token: (_ for _ in ()).throw(
ValueError("non-finite JSON number: " + token)))
class OutOfLoopHoldout:
def __init__(
self,
manifest_path: str,
signature_path: str,
expected_manifest_sha256: str,
):
if not SHA256.fullmatch(expected_manifest_sha256):
raise ValueError("expected holdout digest is invalid")
payload = Path(manifest_path).read_bytes()
signature = Path(signature_path).read_bytes()
public_key = TRUST_ROOT.joinpath("holdout.pub").read_bytes()
if len(signature) != 64 or len(public_key) != 32:
raise ValueError("Ed25519 signature or public key length differs")
actual_digest = hashlib.sha256(payload).hexdigest()
if actual_digest != expected_manifest_sha256:
raise RuntimeError("holdout bytes differ from signed promotion scope")
Ed25519PublicKey.from_public_bytes(public_key).verify(signature, payload)
manifest = strict_json(payload)
if not isinstance(manifest, dict) or set(manifest) != MANIFEST_FIELDS:
raise ValueError("holdout manifest schema differs")
if manifest["schema_version"] != "aidefend.h026-holdout.v1":
raise ValueError("unsupported holdout manifest schema")
if any(not isinstance(manifest[name], str) or not manifest[name]
for name in ("holdout_id", "version", "created_at")):
raise ValueError("holdout identity is incomplete")
samples = manifest["samples"]
if not isinstance(samples, list) or not samples:
raise ValueError("holdout population is empty")
case_ids: set[str] = set()
for sample in samples:
if not isinstance(sample, dict) or set(sample) != SAMPLE_FIELDS:
raise ValueError("holdout sample schema differs")
if (not isinstance(sample["case_id"], str) or not sample["case_id"]
or sample["case_id"] in case_ids):
raise ValueError("holdout case ID is empty or duplicated")
if (not isinstance(sample["text"], str) or not sample["text"].strip()
or sample["label_int"] not in (0, 1)
or isinstance(sample["label_int"], bool)
or not isinstance(sample["category"], str) or not sample["category"]
or not isinstance(sample["source_ref"], str) or not sample["source_ref"]):
raise ValueError("holdout sample value differs")
case_ids.add(sample["case_id"])
if canonical(manifest) != payload:
raise ValueError("holdout manifest is not canonical")
self.holdout_id = manifest["holdout_id"]
self.version = manifest["version"]
self.created_at = manifest["created_at"]
self.samples = tuple(samples)
self.manifest_sha256 = actual_digest
self.signature_sha256 = hashlib.sha256(signature).hexdigest()
# Public verification keys are mounted by the independent deployment authority,
# never supplied in detector, training-job, request, or benchmark fields.
TRUST_ROOT = Path("/opt/aidefend/trust/detector-promotion")
ROLES = {"scope", "holdout", "benchmark", "latency", "decision"}
def verify_document(role: str, raw: bytes, signature: bytes) -> dict:
if role not in ROLES or not isinstance(raw,bytes) or not isinstance(signature,bytes):
raise ValueError("signed document input is invalid")
key=TRUST_ROOT.joinpath(role+".pub").read_bytes()
Ed25519PublicKey.from_public_bytes(key).verify(signature,raw)
value=strict_json(raw)
if not isinstance(value,dict) or canonical(value)!=raw:
raise ValueError("signed document is not a canonical object")
return value
Evaluate every case against the exact candidate
# File: detector_hardening/out_of_loop_benchmark.py
import hashlib
import math
from datetime import datetime, timezone
import pandas as pd
from detector_hardening.category_report import CategoryGatePolicy, EVAL_COLUMNS, stratified_report
from detector_hardening.weak_category_gate import identify_weak_categories
from detector_hardening.sealed_holdout import OutOfLoopHoldout, SHA256, canonical, verify_document
SCOPE_FIELDS = {"schema_version","policy_id","policy_version","expires_at","candidate_sha256",
"candidate_version","holdout_manifest_sha256","baseline_report_sha256","required_categories",
"benign_category","min_malicious_recall","min_category_recall","max_benign_fpr",
"max_category_regression","latency_quantile","quantile_method","max_latency_ms","max_candidate_bytes",
"deployment_replicas","deployment_generation","decision_store_timeout_seconds"}
DETECTOR_RESULT_FIELDS = {"label", "detector_sha256", "request_id"}
ROW_FIELDS = {"case_id", "label", "pred", "category", "request_id"}
BENCHMARK_FIELDS = {"schema_version", "evaluation_policy_sha256", "baseline_report_sha256",
"detector_version", "detector_sha256", "holdout_id", "holdout_version", "holdout_manifest_sha256",
"holdout_signature_sha256", "expected_case_count", "measured_case_count", "raw_predictions",
"raw_predictions_sha256", "stratified_report", "weak_categories", "eligibility"}
def report_from_predictions(rows, holdout, detector_version, policy):
expected = {sample["case_id"]: sample for sample in holdout.samples}
if not isinstance(rows, list) or len(rows) != len(expected):
raise ValueError("prediction population is incomplete")
seen = set(); requests = set(); inputs = []
for row in rows:
if not isinstance(row, dict) or set(row) != ROW_FIELDS:
raise ValueError("prediction field set differs")
sample = expected.get(row["case_id"])
if (sample is None or row["case_id"] in seen or type(row["label"]) is not int
or type(row["pred"]) is not int or row["pred"] not in (0, 1)
or (row["label"], row["category"]) != (sample["label_int"], sample["category"])
or not isinstance(row["request_id"], str) or not row["request_id"] or row["request_id"] in requests):
raise ValueError("prediction identity, label or request differs from sealed population")
seen.add(row["case_id"]); requests.add(row["request_id"])
inputs.append({key: row[key] for key in ("case_id", "label", "pred", "category")} |
{"source": sample["source_ref"]})
return stratified_report(pd.DataFrame(inputs, columns=sorted(EVAL_COLUMNS)),
detector_version, holdout.holdout_id, policy)
def evaluation_context(holdout, detector_sha, detector_version, scope_bytes, scope_signature,
baseline_bytes=None, baseline_signature=None):
scope = verify_document("scope", scope_bytes, scope_signature)
if (not isinstance(scope, dict) or set(scope) != SCOPE_FIELDS
or scope["schema_version"] != "aidefend.detector-promotion-scope.v1"):
raise ValueError("evaluation scope schema differs")
for name in ("policy_id", "policy_version", "candidate_version", "expires_at"):
if not isinstance(scope[name], str) or not scope[name]:
raise ValueError("evaluation scope identity or expiry is missing")
for name in ("candidate_sha256", "holdout_manifest_sha256"):
if not isinstance(scope[name], str) or SHA256.fullmatch(scope[name]) is None:
raise ValueError("evaluation scope digest differs")
baseline_digest = scope["baseline_report_sha256"]
if baseline_digest is not None and (not isinstance(baseline_digest, str) or SHA256.fullmatch(baseline_digest) is None):
raise ValueError("evaluation baseline digest differs")
timeout = scope["decision_store_timeout_seconds"]
if type(timeout) not in (int, float) or not math.isfinite(timeout) or timeout < 0:
raise ValueError("signed evidence-store deadline is invalid")
expiry = datetime.fromisoformat(scope["expires_at"].replace("Z", "+00:00"))
if expiry.tzinfo is None or expiry <= datetime.now(timezone.utc):
raise PermissionError("evaluation scope expired")
if (not SHA256.fullmatch(detector_sha) or not detector_version
or scope["candidate_sha256"] != detector_sha or scope["candidate_version"] != detector_version
or scope["holdout_manifest_sha256"] != holdout.manifest_sha256):
raise PermissionError("candidate or holdout differs from signed scope")
policy = CategoryGatePolicy(tuple(scope["required_categories"]), scope["benign_category"],
scope["min_category_recall"], scope["max_benign_fpr"], scope["max_category_regression"])
from detector_hardening.category_report import validate_gate_policy
validate_gate_policy(policy)
if scope["baseline_report_sha256"] is None:
# Explicit authority-signed genesis scope only. Never fall back to this
# branch after losing a previously required baseline or its signature.
if baseline_bytes is not None or baseline_signature is not None:
raise PermissionError("genesis scope cannot consume a fabricated prior baseline")
return scope, policy, None
if (baseline_bytes is None or baseline_signature is None
or scope["baseline_report_sha256"] != hashlib.sha256(baseline_bytes).hexdigest()):
raise PermissionError("required baseline missing or differs from signed scope")
baseline = verify_document("benchmark", baseline_bytes, baseline_signature)
if (set(baseline) != BENCHMARK_FIELDS
or baseline["schema_version"] != "aidefend.sealed-detector-benchmark.v1"
or baseline["holdout_id"] != holdout.holdout_id or baseline["holdout_version"] != holdout.version
or baseline["holdout_manifest_sha256"] != holdout.manifest_sha256
or baseline["holdout_signature_sha256"] != holdout.signature_sha256
or type(baseline["expected_case_count"]) is not int or type(baseline["measured_case_count"]) is not int
or baseline["expected_case_count"] != len(holdout.samples)
or baseline["measured_case_count"] != len(holdout.samples)
or baseline["raw_predictions_sha256"] != hashlib.sha256(canonical(baseline["raw_predictions"])).hexdigest()):
raise PermissionError("baseline population or holdout binding differs")
previous = report_from_predictions(baseline["raw_predictions"], holdout, baseline["detector_version"], policy)
if canonical(previous.to_dict(orient="records")) != canonical(baseline["stratified_report"]):
raise PermissionError("baseline aggregate report differs from actual predictions")
return scope, policy, previous
def derive_benchmark(rows, holdout, detector_sha, detector_version, scope_bytes, scope_signature,
baseline_bytes=None, baseline_signature=None):
scope, policy, previous = evaluation_context(holdout, detector_sha, detector_version,
scope_bytes, scope_signature, baseline_bytes, baseline_signature)
report = report_from_predictions(rows, holdout, detector_version, policy)
weak = identify_weak_categories(report, previous, policy)
return {"schema_version": "aidefend.sealed-detector-benchmark.v1",
"evaluation_policy_sha256": hashlib.sha256(scope_bytes).hexdigest(),
"baseline_report_sha256": scope["baseline_report_sha256"],
"detector_version": detector_version, "detector_sha256": detector_sha,
"holdout_id": holdout.holdout_id, "holdout_version": holdout.version,
"holdout_manifest_sha256": holdout.manifest_sha256, "holdout_signature_sha256": holdout.signature_sha256,
"expected_case_count": len(holdout.samples), "measured_case_count": len(rows),
"raw_predictions": rows, "raw_predictions_sha256": hashlib.sha256(canonical(rows)).hexdigest(),
"stratified_report": report.to_dict(orient="records"), "weak_categories": weak,
"eligibility": "ELIGIBLE" if not weak else "HOLD"}
def run_out_of_loop_benchmark(detector_fn, holdout: OutOfLoopHoldout,
expected_detector_sha256: str, detector_version: str, evaluation_policy_bytes: bytes,
evaluation_policy_signature: bytes, baseline_bytes=None, baseline_signature=None) -> dict:
# Authenticate prerequisites before calling the deployed detector adapter.
evaluation_context(holdout, expected_detector_sha256, detector_version,
evaluation_policy_bytes, evaluation_policy_signature, baseline_bytes, baseline_signature)
rows = []
for sample in holdout.samples:
result = detector_fn(sample["text"], sample["case_id"])
if (not isinstance(result, dict) or set(result) != DETECTOR_RESULT_FIELDS
or result["label"] not in {"malicious", "benign"}
or result["detector_sha256"] != expected_detector_sha256):
raise RuntimeError("detector result is invalid or misbound")
rows.append({"case_id": sample["case_id"], "label": sample["label_int"],
"pred": 1 if result["label"] == "malicious" else 0,
"category": sample["category"], "request_id": result["request_id"]})
return derive_benchmark(rows, holdout, expected_detector_sha256, detector_version,
evaluation_policy_bytes, evaluation_policy_signature, baseline_bytes, baseline_signature)
if __name__ == "__main__":
import argparse
import importlib
from pathlib import Path
parser = argparse.ArgumentParser(description="Measure a candidate against the sealed holdout")
parser.add_argument("--detector", required=True, help="trusted deployed adapter module:function")
parser.add_argument("--manifest", type=Path, required=True)
parser.add_argument("--manifest-signature", type=Path, required=True)
parser.add_argument("--scope", type=Path, required=True)
parser.add_argument("--scope-signature", type=Path, required=True)
parser.add_argument("--baseline", type=Path)
parser.add_argument("--baseline-signature", type=Path)
parser.add_argument("--output", type=Path, required=True)
arguments = parser.parse_args()
# Only the benchmark service operator chooses the adapter and its installed
# pinned code. Never expose module loading as a request/training-job field.
module_name, separator, function_name = arguments.detector.partition(":")
if not separator or not module_name or not function_name:
parser.error("detector must be module:function")
detector = getattr(importlib.import_module(module_name), function_name)
if not callable(detector):
raise TypeError("installed detector adapter is not callable")
scope_bytes, scope_signature = arguments.scope.read_bytes(), arguments.scope_signature.read_bytes()
scope = verify_document("scope", scope_bytes, scope_signature)
holdout = OutOfLoopHoldout(str(arguments.manifest), str(arguments.manifest_signature),
scope["holdout_manifest_sha256"])
result = run_out_of_loop_benchmark(detector, holdout, scope["candidate_sha256"],
scope["candidate_version"], scope_bytes, scope_signature,
arguments.baseline.read_bytes() if arguments.baseline else None,
arguments.baseline_signature.read_bytes() if arguments.baseline_signature else None)
# Unsigned measurement output has no release authority. The independently
# deployed evidence service must persist_benchmark and read back its signature.
with arguments.output.open("xb") as target:
target.write(canonical(result))
Durably publish and independently read back coverage evidence
# File: detector_hardening/benchmark_evidence.py
from __future__ import annotations
import hashlib
import math
import os
from pathlib import Path
import sqlite3
from contextlib import closing
from detector_hardening.sealed_holdout import canonical, verify_document
from detector_hardening.out_of_loop_benchmark import BENCHMARK_FIELDS, derive_benchmark, evaluation_context
def lock_timeout(value):
if type(value) not in (int, float) or not math.isfinite(value) or value < 0:
raise ValueError("verified evidence-store deadline must be finite and nonnegative")
return float(value)
def provision_benchmark_store(path: Path, store_id: str, timeout_seconds: float) -> None:
# Administrative provisioning for a new deployment only, never runtime
# recovery from missing state. Existing lineages require restored history.
timeout_seconds = lock_timeout(timeout_seconds)
if not isinstance(store_id, str) or not store_id:
raise ValueError("deployment-pinned evidence-store identity is missing")
path.parent.mkdir(parents=True, exist_ok=True)
descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
os.close(descriptor)
with closing(sqlite3.connect(path, timeout=timeout_seconds)) as db, db:
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=FULL")
db.executescript("""
CREATE TABLE metadata (store_id TEXT NOT NULL, schema_version TEXT NOT NULL);
CREATE TABLE lineages (policy_id TEXT NOT NULL, holdout_id TEXT NOT NULL,
genesis_scope_sha256 TEXT NOT NULL UNIQUE, PRIMARY KEY(policy_id, holdout_id));
CREATE TABLE evidence (payload_sha256 TEXT PRIMARY KEY, policy_id TEXT NOT NULL,
holdout_id TEXT NOT NULL, scope_sha256 TEXT NOT NULL,
payload BLOB NOT NULL, signature BLOB);
CREATE TRIGGER metadata_no_update BEFORE UPDATE ON metadata BEGIN SELECT RAISE(ABORT,'immutable identity'); END;
CREATE TRIGGER metadata_no_delete BEFORE DELETE ON metadata BEGIN SELECT RAISE(ABORT,'immutable identity'); END;
CREATE TRIGGER lineages_no_update BEFORE UPDATE ON lineages BEGIN SELECT RAISE(ABORT,'immutable lineage'); END;
CREATE TRIGGER lineages_no_delete BEFORE DELETE ON lineages BEGIN SELECT RAISE(ABORT,'immutable lineage'); END;
CREATE TRIGGER evidence_no_delete BEFORE DELETE ON evidence BEGIN SELECT RAISE(ABORT,'retained evidence'); END;
CREATE TRIGGER evidence_finalize_only BEFORE UPDATE ON evidence
WHEN OLD.signature IS NOT NULL OR NEW.signature IS NULL
OR OLD.payload_sha256 IS NOT NEW.payload_sha256 OR OLD.policy_id IS NOT NEW.policy_id
OR OLD.holdout_id IS NOT NEW.holdout_id OR OLD.scope_sha256 IS NOT NEW.scope_sha256
OR OLD.payload IS NOT NEW.payload
BEGIN SELECT RAISE(ABORT,'evidence replacement denied'); END;
""")
db.execute("INSERT INTO metadata VALUES (?,?)", (store_id, "aidefend.benchmark-lineage-store.v1"))
class BenchmarkLineageStore:
def __init__(self, path: Path, expected_store_id: str, timeout_seconds: float):
self.timeout_seconds = lock_timeout(timeout_seconds)
if not isinstance(expected_store_id, str) or not expected_store_id:
raise ValueError("deployment-pinned evidence-store identity is missing")
# The evidence service exclusively owns the protected mount and parent.
# The candidate/training job cannot choose this path or access the DB.
if path.is_symlink() or not path.is_file():
raise FileNotFoundError("benchmark lineage store is not provisioned")
uri = path.resolve(strict=True).as_uri() + "?mode=rw"
self.db = sqlite3.connect(uri, uri=True, timeout=self.timeout_seconds, isolation_level=None)
try:
self.db.execute("PRAGMA synchronous=FULL")
if self.db.execute("SELECT store_id,schema_version FROM metadata").fetchall() != [
(expected_store_id, "aidefend.benchmark-lineage-store.v1")]:
raise PermissionError("benchmark lineage store identity differs")
self.db.execute("SELECT genesis_scope_sha256 FROM lineages LIMIT 0")
self.db.execute("SELECT signature FROM evidence LIMIT 0")
except BaseException:
self.db.close()
raise
def close(self):
self.db.close()
def reserve(self, scope: dict, holdout_id: str, scope_sha256: str, raw: bytes) -> str:
if scope["decision_store_timeout_seconds"] != self.timeout_seconds:
raise PermissionError("evidence-store deadline differs from signed scope")
payload_sha256 = hashlib.sha256(raw).hexdigest()
identity = (scope["policy_id"], holdout_id)
self.db.execute("BEGIN IMMEDIATE")
try:
lineage = self.db.execute("SELECT genesis_scope_sha256 FROM lineages WHERE policy_id=? AND holdout_id=?", identity).fetchone()
if scope["baseline_report_sha256"] is None:
if lineage is not None:
raise PermissionError("genesis already reserved or retained for this policy/holdout lineage")
self.db.execute("INSERT INTO lineages VALUES (?,?,?)", (*identity, scope_sha256))
else:
if lineage is None:
raise PermissionError("established baseline lineage is missing; genesis fallback forbidden")
prior = self.db.execute("SELECT payload,signature FROM evidence WHERE payload_sha256=? AND policy_id=? AND holdout_id=?",
(scope["baseline_report_sha256"], *identity)).fetchone()
if prior is None or prior[1] is None:
raise PermissionError("prior baseline lacks completed retained evidence")
verify_document("benchmark", prior[0], prior[1])
if hashlib.sha256(prior[0]).hexdigest() != scope["baseline_report_sha256"]:
raise PermissionError("retained baseline digest differs")
self.db.execute("INSERT INTO evidence VALUES (?,?,?,?,?,NULL)",
(payload_sha256, *identity, scope_sha256, raw))
self.db.execute("COMMIT")
except sqlite3.IntegrityError as exc:
self.db.execute("ROLLBACK")
raise PermissionError("benchmark report already reserved") from exc
except BaseException:
self.db.execute("ROLLBACK")
raise
# Commit BEFORE requesting a signature. A failed or uncertain external
# append does not release the genesis/scope reservation after restart.
return payload_sha256
def finalize(self, payload_sha256: str, raw: bytes, signature: bytes) -> None:
verify_document("benchmark", raw, signature)
if hashlib.sha256(raw).hexdigest() != payload_sha256:
raise PermissionError("finalized evidence digest differs")
self.db.execute("BEGIN IMMEDIATE")
try:
changed = self.db.execute("UPDATE evidence SET signature=? WHERE payload_sha256=? AND payload=? AND signature IS NULL",
(signature, payload_sha256, raw)).rowcount
if changed != 1:
raise PermissionError("evidence reservation missing or already finalized")
self.db.execute("COMMIT")
except BaseException:
self.db.execute("ROLLBACK")
raise
def persist_benchmark(benchmark_result: dict, append_and_readback, *, holdout,
scope_bytes, scope_signature, lineage_store: BenchmarkLineageStore,
baseline_bytes=None, baseline_signature=None) -> dict:
if not isinstance(benchmark_result, dict) or set(benchmark_result) != BENCHMARK_FIELDS:
raise ValueError("benchmark field set differs")
# The evidence service independently authenticates inputs and re-derives
# report, weak categories and eligibility before invoking its signer.
derived = derive_benchmark(benchmark_result["raw_predictions"], holdout,
benchmark_result["detector_sha256"], benchmark_result["detector_version"],
scope_bytes, scope_signature, baseline_bytes, baseline_signature)
scope, _policy, _previous = evaluation_context(holdout, benchmark_result["detector_sha256"],
benchmark_result["detector_version"], scope_bytes, scope_signature, baseline_bytes, baseline_signature)
raw = canonical(benchmark_result)
if raw != canonical(derived):
raise PermissionError("submitted benchmark contradicts independently recomputed evidence")
if not isinstance(lineage_store, BenchmarkLineageStore):
raise TypeError("protected benchmark lineage store is required")
reservation = lineage_store.reserve(scope, holdout.holdout_id, hashlib.sha256(scope_bytes).hexdigest(), raw)
readback, signature = append_and_readback(raw)
observed = verify_document("benchmark", readback, signature)
if readback != raw or observed != derived:
raise PermissionError("independent signed benchmark readback differs")
lineage_store.finalize(reservation, readback, signature)
return {"eligibility": observed["eligibility"], "payload": readback, "signature": signature,
"payload_sha256": hashlib.sha256(readback).hexdigest()}
Run the measured producer
Install the code blocks as their named modules. The independently deployed benchmark service supplies a pinned module:function adapter implementing detector_fn(text, case_id); it must return the exact measured detector digest, unique request identity and binary label from the isolated candidate. Run python -m detector_hardening.out_of_loop_benchmark --help for the required signed-input paths. Its output is unsigned measurement data; only persist_benchmark can obtain authenticated benchmark evidence. Deliberately undefined recall or FPR is JSON null; a missing applicable metric or non-finite numeric value remains invalid.
Operational verification
Only the independently deployed benchmark service can use the benchmark signing key. The evidence service stores the exact canonical report bytes, detached signature and raw predictions under a non-overwrite identity and independently reads them back. HOLD is retained as a measured failure or completeness finding; it must not disappear as an unrecorded exception. The final promotion consumer re-verifies issuer keys, exact candidate and holdout, policy and baseline hashes, complete case identities and raw observations. A coverage report cannot mutate aliases, replicas or deployment state. Tests must reject missing, duplicate or extra cases, mixed labels, forged or stale source facts and mismatched candidate/scope bindings.
Bootstrap and independently persist the first baseline
The scope authority may explicitly issue a genesis measurement scope for the already approved baseline detector with baseline_report_sha256: null. Omit both baseline CLI arguments only for that signed scope. The actual holdout producer and evidence service still verify the sealed case population, derive every aggregate and check absolute attack/benign thresholds. A null predecessor means there is no prior baseline; no regression comparison is claimed. It is a nullable schema field, not an additional string state. Never hand-create or hand-sign aggregate reports, automatically infer genesis from missing files, or reuse a genesis scope to erase retained history. Preserve the signed output as the prior baseline for subsequent comparisons. Initial production admission remains a separate approved release; this measurement cannot authorize it.
Call persist_benchmark(result, append_and_readback, holdout=verified_holdout, scope_bytes=scope_bytes, scope_signature=scope_signature, lineage_store=lineage_store, baseline_bytes=baseline_bytes, baseline_signature=baseline_signature) in the independently deployed evidence service. Supply null baseline arguments only for explicitly signed genesis. The service re-derives report, weak categories and eligibility from raw predictions before invoking the signer; mismatched or missing applicable metrics are rejected before any signature.
Reserve lineage before requesting any signature
The evidence service must open its protected BenchmarkLineageStore with the deployment-pinned path and store ID, and a SQLite deadline matching the verified scope's decision_store_timeout_seconds. The candidate cannot supply this object, path, identity or database. A deployment administrator provisions a new store once with provision_benchmark_store(path, store_id, timeout_seconds); runtime uses existing-only mode and refuses a missing or different store. An existing deployment must restore and reconcile its previously retained lineage/evidence before enabling this path; provisioning an empty replacement is forbidden.
A committed reservation makes (policy_id, holdout_id) genesis one-use, even after a service restart or a fresh signed null-baseline scope. Policy-version or holdout-version changes under those same IDs cannot reset history. Normal measurements require the exact completed, signed baseline already retained in that lineage. The evidence store records HOLD measurements too, without making them the active baseline. The signed scope authority selects an approved baseline digest; AID-H-026.001 alone owns production promotion and deployment.
Scope/report reservation is durable before the external append/signer is called. Signature, readback or finalization failure leaves a reserved unresolved record and cannot enable another genesis. Under the independent evidence-service operator, reconcile the exact reserved digest against the append-only store and verify its signed bytes before completing that reservation with finalize; never delete the reservation to retry. Check duplicate scope, fresh re-genesis, concurrent workers, missing mount, wrong store identity and failure after signing. SQLite triggers are extra safeguards: independently protected file custody and recovery from retained records remain deployment requirements.
SQLite storage prerequisite: the database, WAL and shared-memory files must remain on a protected local filesystem, with all participating processes on the same host. This WAL path does not support NFS/SMB or independent hosts sharing a file. A distributed deployment needs an independently managed transactional store providing the same durable uniqueness/CAS and authenticated readback. See SQLite WAL deployment requirements.
Reservation recovery: independently recover the exact reserved canonical payload and reconcile the authenticated append-only store. If no signature was issued, the protected evidence service may submit those same reserved bytes to its signer/append service under recovery authority, independently verify the exact readback, and call finalize. Never re-measure or delete/release the reservation to obtain a different report. An uncertain signature/append response remains unresolved until readback is reconciled. A new policy_id or holdout_id creates a distinct lineage only under the signed scope authority; changing a version under the same IDs cannot reset it. The active production baseline remains a scope/deployment-authority selection: the registry retains measurements, including HOLD, and does not choose the deployed baseline. This reference deliberately uses the signed decision_store_timeout_seconds for both evidence and promotion SQLite lock deadlines.
AID-H-026.003-G003
Use attribution methods to debug representative detector misses and guide the next hardening cycle.
Purpose and scope boundary
Use attribution only to debug a policy-selected population of detector errors after the sealed benchmark. It is not causal proof, a promotion metric, or authorization to change a model. Preserve the complete miss population and deterministic selection evidence so analysts cannot cherry-pick attractive explanations.
Step 1: Deterministically select representative misses from every weak category
# File: detector_hardening/attribution_selection.py
from __future__ import annotations
import hashlib
import math
import re
from dataclasses import dataclass
EVAL_ROW_FIELDS = {
"case_id", "text", "text_sha256", "label", "pred", "category",
"detector_sha256",
}
SHA256 = re.compile(r"^[a-f0-9]{64}$")
@dataclass(frozen=True)
class AttributionPolicy:
max_examples_per_category: int
selection_seed: str
tokenizer_max_length: int
integration_steps: int
top_tokens_per_direction: int
def positive_integer(value: object, name: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError(name + " must be a positive integer")
return value
def select_representative_misses(
eval_rows: list[dict],
weak_categories: list[dict],
expected_detector_sha256: str,
policy: AttributionPolicy,
) -> dict:
positive_integer(policy.max_examples_per_category, "max_examples_per_category")
if not policy.selection_seed or not SHA256.fullmatch(expected_detector_sha256):
raise ValueError("selection seed or detector digest is invalid")
if not isinstance(weak_categories, list) or not weak_categories:
raise ValueError("weak-category finding population is empty")
finding_keys: list[tuple[str, str]] = []
for item in weak_categories:
if (not isinstance(item, dict)
or not isinstance(item.get("category"), str) or not item["category"]
or item.get("issue") not in {"high_fpr", "low_recall", "recall_regression"}):
raise ValueError("weak-category finding schema differs")
regression = item["issue"] == "recall_regression"
numeric_fields = {"current", "previous", "delta", "threshold"} if regression else {"value", "threshold"}
if set(item) != {"category", "issue"} | numeric_fields:
raise ValueError("weak-category finding schema differs")
for field in numeric_fields:
value = item[field]
if (isinstance(value, bool) or not isinstance(value, (int, float))
or not math.isfinite(value) or not 0 <= value <= 1):
raise ValueError("weak-category metric is invalid: " + field)
if regression:
if (not math.isclose(item["previous"] - item["current"], item["delta"], rel_tol=1e-12, abs_tol=1e-12)
or item["delta"] <= item["threshold"]):
raise ValueError("recall-regression evidence is inconsistent")
elif ((item["issue"] == "high_fpr" and item["value"] <= item["threshold"])
or (item["issue"] == "low_recall" and item["value"] >= item["threshold"])):
raise ValueError("finding does not breach its threshold")
finding_keys.append((item["category"], item["issue"]))
if len(set(finding_keys)) != len(finding_keys):
raise ValueError("weak-category finding is duplicated")
weak_names = sorted({category for category, _ in finding_keys})
case_ids: set[str] = set()
misses: dict[str, list[dict]] = {category: [] for category in weak_names}
for row in eval_rows:
if not isinstance(row, dict) or set(row) != EVAL_ROW_FIELDS:
raise ValueError("evaluation-row schema differs")
if not row["case_id"] or row["case_id"] in case_ids:
raise ValueError("evaluation case ID is empty or duplicated")
case_ids.add(row["case_id"])
if (not isinstance(row["text"], str) or not row["text"].strip()
or hashlib.sha256(row["text"].encode("utf-8")).hexdigest() != row["text_sha256"]
or row["label"] not in (0, 1) or row["pred"] not in (0, 1)
or isinstance(row["label"], bool) or isinstance(row["pred"], bool)
or row["detector_sha256"] != expected_detector_sha256):
raise ValueError("evaluation row is invalid or misbound")
if row["category"] in misses and row["label"] != row["pred"]:
misses[row["category"]].append(row)
if any(not rows for rows in misses.values()):
raise RuntimeError("a weak category has no attributable miss population")
selected: dict[str, list[dict]] = {}
for category, rows in misses.items():
ordered = sorted(rows, key=lambda row: hashlib.sha256(
(policy.selection_seed + ":" + row["case_id"]).encode("utf-8")
).hexdigest())
selected[category] = ordered[:policy.max_examples_per_category]
return {
"eligible_miss_counts": {name: len(rows) for name, rows in misses.items()},
"selected_counts": {name: len(rows) for name, rows in selected.items()},
"selected_case_ids": {
name: [row["case_id"] for row in rows] for name, rows in selected.items()
},
"selected": selected,
}
Step 2: Run version-bound integrated gradients with convergence evidence
# File: detector_hardening/miss_attribution.py
import math
import torch
from captum.attr import LayerIntegratedGradients
from detector_hardening.attribution_selection import AttributionPolicy, positive_integer
class UnsupportedAttribution(RuntimeError):
pass
def explain_detector_miss(
model,
tokenizer,
case: dict,
policy: AttributionPolicy,
) -> dict:
max_length = positive_integer(policy.tokenizer_max_length, "tokenizer_max_length")
steps = positive_integer(policy.integration_steps, "integration_steps")
if case["label"] == case["pred"] or case["label"] not in (0, 1):
raise ValueError("attribution input is not a binary detector miss")
if not hasattr(model, "get_input_embeddings"):
raise UnsupportedAttribution("model has no supported input-embedding boundary")
embeddings = model.get_input_embeddings()
if embeddings is None or tokenizer.pad_token_id is None:
raise UnsupportedAttribution("embedding layer or tokenizer baseline is unavailable")
model.eval()
device = next(model.parameters()).device
encoded = tokenizer(
case["text"], return_tensors="pt", truncation=True, max_length=max_length
)
input_ids = encoded["input_ids"].to(device)
attention_mask = encoded["attention_mask"].to(device)
baseline = torch.full_like(input_ids, tokenizer.pad_token_id)
def forward(ids, mask):
return model(input_ids=ids, attention_mask=mask).logits
explainer = LayerIntegratedGradients(forward, embeddings)
attribution, convergence_delta = explainer.attribute(
inputs=input_ids,
baselines=baseline,
additional_forward_args=(attention_mask,),
target=case["label"],
n_steps=steps,
return_convergence_delta=True,
)
scores = attribution.sum(dim=-1).squeeze(0).detach().cpu().tolist()
tokens = tokenizer.convert_ids_to_tokens(input_ids.squeeze(0).detach().cpu())
delta = float(convergence_delta.detach().cpu().reshape(-1)[0])
if len(tokens) != len(scores) or any(not math.isfinite(float(value)) for value in scores + [delta]):
raise RuntimeError("attribution population or convergence result differs")
return {
"case_id": case["case_id"], "category": case["category"],
"text_sha256": case["text_sha256"],
"detector_sha256": case["detector_sha256"],
"target_label": case["label"], "token_ids": input_ids.squeeze(0).tolist(),
"tokens": tokens, "token_scores": scores,
"convergence_delta": delta,
"tokenizer_max_length": max_length, "integration_steps": steps,
}
Step 3: Convert every selected explanation into a reviewable backlog item
# File: detector_hardening/debug_backlog.py
from detector_hardening.attribution_selection import AttributionPolicy, positive_integer
def build_debug_backlog(explanations: list[dict], policy: AttributionPolicy) -> list[dict]:
top_k = positive_integer(policy.top_tokens_per_direction, "top_tokens_per_direction")
if not explanations:
raise ValueError("selected explanation population is empty")
case_ids: set[str] = set()
backlog: list[dict] = []
for explanation in explanations:
case_id = explanation.get("case_id")
tokens = explanation.get("tokens")
scores = explanation.get("token_scores")
if (not isinstance(case_id, str) or not case_id or case_id in case_ids
or not isinstance(tokens, list) or not isinstance(scores, list)
or not tokens or len(tokens) != len(scores)):
raise ValueError("explanation schema, population, or case identity differs")
case_ids.add(case_id)
ranked = list(zip(tokens, scores))
backlog.append({
"case_id": case_id, "category": explanation["category"],
"text_sha256": explanation["text_sha256"],
"detector_sha256": explanation["detector_sha256"],
"suspect_positive_tokens": sorted(
ranked, key=lambda item: item[1], reverse=True)[:top_k],
"suspect_negative_tokens": sorted(ranked, key=lambda item: item[1])[:top_k],
"convergence_delta": explanation["convergence_delta"],
"proposed_actions": [
"review tokenizer and normalization behavior",
"mine governed adversarial and benign counterexamples",
"test the hypothesis in a separately approved experiment",
],
"release_authority": "none",
})
if len(backlog) != len(explanations):
raise RuntimeError("explanation-to-backlog reconciliation failed")
return backlog
Verify complete coverage
Materialize the complete population of every miss in each weak category, every deterministically selected case, token and score returned for each selected input, every explanation, and every resulting backlog item or error disposition before evaluation; require stable unique IDs and reject duplicates or undeclared members.
Verify safely
Emit a canonical signed receipt binding AID-H-026.003-G003, policy ID/version/digest, eligible/selected population roots and counts, deterministic rank keys, exact detector/tokenizer and input bindings, token IDs/scores, convergence deltas, unsupported/error dispositions, analyst review, backlog IDs, and secure-sink readback, complete-population manifest digest and counts, immutable code/model/tool/container versions, timestamps, disposition reasons, and the applied action. Write the receipt and exact replay inputs to an append-only or WORM sink under an identity that the detector and training job cannot alter, for the policy-derived retention period.
Open-source tools
- pandas
- scikit-learn
- Jupyter
- Captum
- Evidently
Commercial / hosted tools
- Weights & Biases
- Databricks Mosaic AI
- Arize AX
- Fiddler AI Observability Platform
Threat mappings
MITRE ATLAS
- AML.T0051 LLM Prompt Injection
- AML.T0054 LLM Jailbreak
- AML.T0068 LLM Prompt Obfuscation
- AML.T0065 LLM Prompt Crafting
- AML.T0015 Evade AI Model
- AML.T0123 Obfuscated Files or Information (per-category coverage exposes filter weakness against obfuscation and encoding classes)
MAESTRO
- Adversarial Examples (L1)
- Input Validation Attacks (L3)
- Framework Evasion (L3)
- Evasion of Detection (L5)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection
OWASP ML Top 10 2023
- ML01:2023 Input Manipulation Attack
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- NISTAML.018 Prompt Injection
- NISTAML.015 Indirect Prompt Injection
- NISTAML.022 Evasion
- NISTAML.025 Black-box Evasion
Cisco Integrated AI Security and Safety Framework
- AISubtech-1.1.1 Instruction Manipulation (independently measured category failures provide a coverage veto; H-026.001 enforces final promotion)
- AISubtech-1.1.2 Obfuscation (independently measured category failures provide a coverage veto; H-026.001 enforces final promotion)
- AISubtech-1.1.5 Context Manipulation (independently measured category failures provide a coverage veto; H-026.001 enforces final promotion)
- AITech-1.1 Direct Prompt Injection and Jailbreak (Prompt-based Manipulation) (independently measured category failures provide a coverage veto; H-026.001 enforces final promotion)
- AITech-1.2 Indirect Prompt Injection (independently measured category failures provide a coverage veto; H-026.001 enforces final promotion)
- AITech-9.2 Detection Evasion (independently measured category failures provide a coverage veto; H-026.001 enforces final promotion)
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection (gap analysis identifies category-level blind spots in injection detection)
- MEV: Model Evasion (stratified analysis reveals which evasion styles succeed against the detector)
Databricks AI Security Framework 3.0
- Model Serving - Inference requests 9.1: Prompt inject
- Model Serving - Inference requests 9.12: LLM Jailbreak
- Evaluation 6.2: Insufficient evaluation data (stratified evaluation exposes dataset gaps)