AID-D-008: Secondary AI Defender Fusion & Assurance

Version 1.20261015 · Actionable technique

Build and operate a versioned secondary AI defender that consumes normalized findings from existing AIDEFEND detection families and learns cross-detector combinations, coordinated campaigns, or evasive patterns that isolated detectors may miss. The control owns the fusion model's feature contract, calibration, promotion, runtime scoring, evidence, and controlled retraining.

Pillars: model, app

Lifecycle: building, validation, operation, improvement

Scope boundary

Owns the feature contract, calibration, promotion, runtime scoring, evidence, retraining, and rollback of a secondary fusion model that combines normalized findings from canonical Detect owners. It emits a cross-detector finding without re-owning source-domain detection, SIEM routing, or containment.

AID-D-005.003: AID-D-008 serves a governed learned fusion detector over normalized source findings; AID-D-005.003 performs human hypothesis-driven threat hunting over telemetry. Fusion discovers calibrated combinations within its feature contract, while hunting can investigate novel hypotheses outside that contract.

AID-D-005.008: AID-D-008 produces a cross-detector security finding; AID-D-005.008 performs read-only evidence gathering and recommendation for a queued alert. Meta-detection and case triage have different outputs, privileges, and validation requirements.

AID-D-018: AID-D-008 owns operation of its fusion model; AID-D-018 validates whether production detections, including fusion where applicable, identify required scenarios with acceptable efficacy and coverage. A calibrated fusion artifact still requires independent scenario-level production efficacy validation.

Warning

Medium on Detector Cost, Latency & False-Positive Coupling

A fusion model adds compute and can amplify correlated errors from its source detectors. Resolve latency budgets, minimum source coverage, confidence thresholds, and safe fallback behavior from a signed deployment policy. Run shadow evaluation before promotion and never interpret missing source findings as benign evidence.

Implementation guidance

AID-D-008-G001

Normalize complete findings from canonical Detect owners into a signed-policy, coverage-aware feature contract for the D-008 fusion model.
Concept

D-008 consumes findings, not raw domain telemetry. Normalize findings from the canonical Detect owners into a complete, versioned feature row while preserving missing/error states. A source detector remains responsible for its own semantics; D-008 learns only cross-detector combinations.

Build a signed-policy feature contract
from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
FUSION_NORMALIZATION_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D008_NORMALIZATION_COMMAND_TIMEOUT_SECONDS")

# File: d008/build_fusion_features.py

import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from collections import Counter
from pathlib import Path


POLICY_PATH = Path(os.environ["D008_SOURCE_POLICY_PATH"])
POLICY_SIGNATURE = Path(os.environ["D008_SOURCE_POLICY_SIGNATURE"])
POLICY_VERIFY_KEY = os.environ["D008_SOURCE_POLICY_VERIFY_KEY"]
REQUIRED_FINDING_FIELDS = {
    "finding_id", "source_control_id", "asset_id", "session_id",
    "status", "severity", "confidence", "observed_at",
    "evidence_sha256", "detector_version",
}
VALID_STATUS = {"FINDING", "NO_THRESHOLD_BREACH", "INSUFFICIENT_DATA", "ERROR"}
VALID_SEVERITY = {"none", "low", "medium", "high", "critical"}


_AIDEFEND_CAPTURED: dict[Path, bytes] = {}


def _aidefend_bytes(path: Path) -> bytes:
    normalized = Path(path)
    if normalized not in _AIDEFEND_CAPTURED:
        _AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
    return _AIDEFEND_CAPTURED[normalized]


def _aidefend_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 _aidefend_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 _aidefend_strict_jsonl(raw: bytes) -> list[object]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("JSONL input is empty or contains a blank record")
    return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str | Path,
    timeout_seconds: float,
) -> bytes:
    payload_path = Path(payload_path)
    proof_path = Path(proof_path)
    if payload_path == proof_path:
        raise RuntimeError("signed payload and signature paths collide")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    key_text = str(key_reference)
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
    )) or key_text.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_text.encode("utf-8")
        if key_text.startswith("-----BEGIN ")
        else Path(key_text).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "payload.sig"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_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(proof_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    _AIDEFEND_CAPTURED[payload_path] = payload_raw
    return payload_raw


def load_policy() -> dict:
    raw = _aidefend_capture_signed_bytes(
        POLICY_PATH, POLICY_SIGNATURE, POLICY_VERIFY_KEY,
        FUSION_NORMALIZATION_COMMAND_TIMEOUT_SECONDS,
    )
    policy = _aidefend_strict_json(raw)
    required = {
        "schema_version", "policy_version", "source_control_ids",
        "minimum_source_coverage",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise ValueError("D-008 source policy schema mismatch")
    sources = policy["source_control_ids"]
    if (
        policy["schema_version"] != "aidefend.d008-source-policy.v1"
        or not isinstance(sources, list) or len(sources) < 2
        or sources != sorted(set(sources))
    ):
        raise ValueError("D-008 source policy is invalid")
    minimum = policy["minimum_source_coverage"]
    if isinstance(minimum, bool) or not isinstance(minimum, int) or not 2 <= minimum <= len(sources):
        raise ValueError("minimum source coverage is invalid")
    return policy


def build_feature_row(findings: list[dict], asset_id: str, session_id: str) -> dict:
    policy = load_policy()
    allowed = set(policy["source_control_ids"])
    seen_ids: set[str] = set()
    normalized: list[dict] = []
    for finding in findings:
        if not isinstance(finding, dict) or set(finding) != REQUIRED_FINDING_FIELDS:
            raise ValueError("finding schema mismatch")
        if finding["finding_id"] in seen_ids:
            raise ValueError("duplicate finding_id")
        if finding["asset_id"] != asset_id or finding["session_id"] != session_id:
            raise ValueError("finding scope mismatch")
        if finding["source_control_id"] not in allowed:
            raise ValueError("unapproved source control")
        if finding["status"] not in VALID_STATUS or finding["severity"] not in VALID_SEVERITY:
            raise ValueError("finding status or severity is invalid")
        for field in ("finding_id", "asset_id", "session_id", "detector_version"):
            if not isinstance(finding[field], str) or not finding[field].strip():
                raise ValueError(f"finding {field} is invalid")
        if not re.fullmatch(r"AID-D-\d{3}(?:\.\d{3})?", finding["source_control_id"]):
            raise ValueError("source control ID is invalid")
        if not re.fullmatch(r"[0-9a-f]{64}", finding["evidence_sha256"]):
            raise ValueError("finding evidence digest is invalid")
        if not re.fullmatch(
            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z",
            finding["observed_at"],
        ):
            raise ValueError("finding timestamp is not canonical UTC")
        confidence = finding["confidence"]
        if (
            isinstance(confidence, bool) or not isinstance(confidence, (int, float))
            or not math.isfinite(float(confidence)) or not 0 <= confidence <= 1
        ):
            raise ValueError("finding confidence is invalid")
        seen_ids.add(finding["finding_id"])
        normalized.append(finding)

    by_source = Counter(item["source_control_id"] for item in normalized)
    present = sorted(by_source)
    measured = sorted({
        item["source_control_id"]
        for item in normalized
        if item["status"] in {"FINDING", "NO_THRESHOLD_BREACH"}
    })
    insufficient = sorted({
        item["source_control_id"]
        for item in normalized
        if item["status"] == "INSUFFICIENT_DATA"
    })
    errors = sorted({
        item["source_control_id"]
        for item in normalized
        if item["status"] == "ERROR"
    })
    missing = sorted(allowed - set(present))
    status = (
        "MEASURED"
        if len(measured) >= policy["minimum_source_coverage"] and not errors
        else "INSUFFICIENT_DATA"
    )
    scores = [
        float(item["confidence"])
        for item in normalized
        if item["status"] in {"FINDING", "NO_THRESHOLD_BREACH"}
    ]
    row = {
        "schema_version": "aidefend.d008-fusion-features.v1",
        "policy_version": policy["policy_version"],
        "asset_id": asset_id,
        "session_id": session_id,
        "status": status,
        "source_controls_present": present,
        "source_controls_covered": measured,
        "source_controls_missing": missing,
        "source_controls_insufficient": insufficient,
        "source_controls_error": errors,
        "finding_count": len(normalized),
        "high_or_critical_count": sum(
            item["severity"] in {"high", "critical"} for item in normalized
        ),
        "error_count": sum(item["status"] == "ERROR" for item in normalized),
        "max_confidence": max(scores) if scores else None,
        "mean_confidence": (sum(scores) / len(scores)) if scores else None,
    }
    for source in policy["source_control_ids"]:
        row[f"count::{source}"] = int(by_source[source])
    return row

Evidence: Retain the verified source-policy bytes/signature, complete input finding IDs and evidence digests, covered/missing source sets, feature-builder digest, and canonical feature row. Missing required source coverage is INSUFFICIENT_DATA, not a benign sample.

Action: Build one canonical feature row per measured asset/session and retain the complete signed source-finding population for replay.

Before you begin

The D-008 fusion-platform owner applies this detector stage only when a signed fusion policy names at least two canonical Detect producers whose findings are semantically compatible for the tenant/workflow and detector release.

AID-D-008-G002

Train and calibrate the D-008 fusion model, then promote it only after a separate evaluator checks the integrity-verified holdout against the signed policy.
Concept

Train and calibrate only the D-008 fusion model. Domain labels remain owned by their source controls. Prompt-injection detector retraining is excluded and belongs to AID-H-026; the holdout and evaluation evidence consumed here must satisfy AID-H-007.004.

Train, calibrate, and gate the candidate
from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
FUSION_TRAINING_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D008_TRAINING_COMMAND_TIMEOUT_SECONDS")

# File: d008/train_fusion_model.py

import hashlib
import hmac
import io
import json
import os
import re
import subprocess
import tempfile
from pathlib import Path

import skops.io as sio
import numpy as np
import pandas as pd
from sklearn.calibration import CalibratedClassifierCV
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import recall_score
from sklearn.model_selection import GroupKFold, GroupShuffleSplit


DATASET = Path(os.environ["D008_TRAINING_TABLE"])
POLICY = Path(os.environ["D008_PROMOTION_POLICY"])
POLICY_SIGNATURE = Path(os.environ["D008_PROMOTION_POLICY_SIGNATURE"])
EVALUATION_MANIFEST = Path(os.environ["D008_EVALUATION_MANIFEST"])
EVALUATION_SIGNATURE = Path(os.environ["D008_EVALUATION_MANIFEST_SIGNATURE"])
OUT = Path("artifacts/d008")
FEATURES = [
    "finding_count", "high_or_critical_count", "error_count",
    "max_confidence", "mean_confidence",
]


_AIDEFEND_CAPTURED: dict[Path, bytes] = {}


def _aidefend_bytes(path: Path) -> bytes:
    normalized = Path(path)
    if normalized not in _AIDEFEND_CAPTURED:
        _AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
    return _AIDEFEND_CAPTURED[normalized]


def _aidefend_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 _aidefend_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 _aidefend_strict_jsonl(raw: bytes) -> list[object]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("JSONL input is empty or contains a blank record")
    return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str | Path,
    timeout_seconds: float,
) -> bytes:
    payload_path = Path(payload_path)
    proof_path = Path(proof_path)
    if payload_path == proof_path:
        raise RuntimeError("signed payload and signature paths collide")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    key_text = str(key_reference)
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
    )) or key_text.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_text.encode("utf-8")
        if key_text.startswith("-----BEGIN ")
        else Path(key_text).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "payload.sig"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_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(proof_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    _AIDEFEND_CAPTURED[payload_path] = payload_raw
    return payload_raw


def sha256_file(path: Path) -> str:
    return hashlib.sha256(_aidefend_bytes(Path(path))).hexdigest()


policy_raw = _aidefend_capture_signed_bytes(
    POLICY, POLICY_SIGNATURE, os.environ["D008_POLICY_VERIFY_KEY"],
    FUSION_TRAINING_COMMAND_TIMEOUT_SECONDS,
)
evaluation_raw = _aidefend_capture_signed_bytes(
    EVALUATION_MANIFEST, EVALUATION_SIGNATURE,
    os.environ["D008_EVALUATION_VERIFY_KEY"],
    FUSION_TRAINING_COMMAND_TIMEOUT_SECONDS,
)
policy = _aidefend_strict_json(policy_raw)
evaluation = _aidefend_strict_json(evaluation_raw)
required_evaluation = {
    "schema_version", "status", "control", "evaluation_id",
    "dataset_sha256", "holdout_population_sha256",
}
if not isinstance(evaluation, dict) or set(evaluation) != required_evaluation:
    raise SystemExit("ERROR: evaluation manifest schema mismatch")
if (
    evaluation["schema_version"] != "aidefend.h007004-evaluation.v1"
    or evaluation["status"] != "PASS"
    or evaluation["control"] != "AID-D-008"
    or evaluation["dataset_sha256"] != sha256_file(DATASET)
    or any(
        not isinstance(evaluation[name], str) or not evaluation[name].strip()
        for name in ("evaluation_id",)
    )
    or any(
        not isinstance(evaluation[name], str)
        or not re.fullmatch(r"[0-9a-f]{64}", evaluation[name])
        for name in ("dataset_sha256", "holdout_population_sha256")
    )
):
    raise SystemExit("ERROR: evaluation manifest is not a bound PASS for this dataset")
required_policy = {
    "schema_version", "policy_version", "evaluation_manifest_sha256",
    "minimum_rows", "decision_threshold", "maximum_false_positive_rate",
    "minimum_malicious_recall", "holdout_fraction", "random_seed",
    "minimum_training_groups", "calibration_folds", "forest_n_estimators",
    "forest_min_samples_leaf", "forest_n_jobs", "calibration_method",
}
if not isinstance(policy, dict) or set(policy) != required_policy:
    raise SystemExit("ERROR: promotion policy schema mismatch")
if policy["schema_version"] != "aidefend.d008-promotion-policy.v1":
    raise SystemExit("ERROR: unsupported promotion policy")
if policy["evaluation_manifest_sha256"] != sha256_file(EVALUATION_MANIFEST):
    raise SystemExit("ERROR: evaluation manifest digest differs from signed policy")
positive_integer_fields = (
    "minimum_rows", "minimum_training_groups", "calibration_folds",
    "forest_n_estimators", "forest_min_samples_leaf",
)
if any(
    isinstance(policy[name], bool)
    or not isinstance(policy[name], int)
    or policy[name] < 1
    for name in positive_integer_fields
):
    raise SystemExit("ERROR: signed training-policy integer bound is invalid")
if policy["minimum_training_groups"] < policy["calibration_folds"]:
    raise SystemExit("ERROR: minimum training groups cannot be below calibration folds")
for name in (
    "decision_threshold", "maximum_false_positive_rate",
    "minimum_malicious_recall",
):
    value = policy[name]
    if (
        isinstance(value, bool)
        or not isinstance(value, (int, float))
        or not _aidefend_math.isfinite(float(value))
        or not 0.0 <= float(value) <= 1.0
    ):
        raise SystemExit(f"ERROR: signed policy fraction is invalid: {name}")
if not 0.0 < float(policy["decision_threshold"]) < 1.0:
    raise SystemExit("ERROR: decision threshold must be inside (0, 1)")
if (
    isinstance(policy["holdout_fraction"], bool)
    or not isinstance(policy["holdout_fraction"], (int, float))
    or not 0 < float(policy["holdout_fraction"]) < 1
):
    raise SystemExit("ERROR: signed holdout fraction is invalid")
if (
    isinstance(policy["random_seed"], bool)
    or not isinstance(policy["random_seed"], int)
    or not 0 <= policy["random_seed"] <= 2**32 - 1
):
    raise SystemExit("ERROR: signed random seed is invalid")
if (
    isinstance(policy["forest_n_jobs"], bool)
    or not isinstance(policy["forest_n_jobs"], int)
    or policy["forest_n_jobs"] == 0
):
    raise SystemExit("ERROR: signed forest worker count is invalid")
if policy["calibration_method"] not in {"sigmoid", "isotonic"}:
    raise SystemExit("ERROR: signed calibration method is invalid")
dataset_raw = _aidefend_bytes(DATASET)
frame = pd.read_parquet(io.BytesIO(dataset_raw))
required_columns = {"event_id", "principal_or_session_group", "label", "status", *FEATURES}
if required_columns - set(frame.columns):
    raise SystemExit("ERROR: training table columns are incomplete")
if len(frame) < int(policy["minimum_rows"]):
    raise SystemExit("INSUFFICIENT_DATA: training population below signed policy")
if frame["event_id"].duplicated().any() or (frame["status"] != "MEASURED").any():
    raise SystemExit("INSUFFICIENT_DATA: duplicate or unmeasured training rows")
if frame[FEATURES].isnull().any().any() or not np.isfinite(frame[FEATURES].to_numpy()).all():
    raise SystemExit("ERROR: feature matrix contains missing or non-finite values")

labels = frame["label"].astype(int)
if set(labels.unique()) != {0, 1}:
    raise SystemExit("INSUFFICIENT_DATA: both benign and malicious labels are required")
groups = frame["principal_or_session_group"].astype(str)
split = GroupShuffleSplit(
    n_splits=1, test_size=float(policy["holdout_fraction"]),
    random_state=policy["random_seed"],
)
train_index, test_index = next(split.split(frame[FEATURES], labels, groups))
x_train, x_test = frame.iloc[train_index][FEATURES], frame.iloc[test_index][FEATURES]
y_train, y_test = labels.iloc[train_index], labels.iloc[test_index]
g_train = groups.iloc[train_index]
if g_train.nunique() < policy["minimum_training_groups"]:
    raise SystemExit("INSUFFICIENT_DATA: independent training groups are below signed policy")
folds = list(GroupKFold(n_splits=policy["calibration_folds"]).split(x_train, y_train, g_train))
if any(y_train.iloc[fit].nunique() != 2 or y_train.iloc[cal].nunique() != 2 for fit, cal in folds):
    raise SystemExit("INSUFFICIENT_DATA: calibration fold lacks one class")

base = RandomForestClassifier(
    n_estimators=policy["forest_n_estimators"],
    min_samples_leaf=policy["forest_min_samples_leaf"],
    class_weight="balanced_subsample", random_state=policy["random_seed"],
    n_jobs=policy["forest_n_jobs"],
)
model = CalibratedClassifierCV(estimator=base, method=policy["calibration_method"], cv=folds)
model.fit(x_train, y_train)
scores = model.predict_proba(x_test)[:, 1]
threshold = float(policy["decision_threshold"])
predicted = scores >= threshold
benign = y_test.to_numpy() == 0
malicious = y_test.to_numpy() == 1
if not benign.any() or not malicious.any():
    raise SystemExit("INSUFFICIENT_DATA: holdout lacks one class")
metrics = {
    "false_positive_rate": float(np.mean(predicted[benign])),
    "malicious_recall": float(recall_score(y_test, predicted)),
}
status = (
    "PASS"
    if metrics["false_positive_rate"] <= float(policy["maximum_false_positive_rate"])
    and metrics["malicious_recall"] >= float(policy["minimum_malicious_recall"])
    else "FAIL"
)

OUT.mkdir(parents=True, exist_ok=True)
artifact = OUT / "fusion_model.skops"
sio.dump({
    "model": model, "features": FEATURES, "threshold": threshold,
    "schema_version": "aidefend.d008-fusion-model.v1",
}, artifact)
trusted_type_names = sorted(sio.get_untrusted_types(file=artifact))
report = {
    "schema_version": "aidefend.d008-promotion-report.v1",
    "status": status,
    "artifact_format": "skops",
    "trusted_type_names": trusted_type_names,
    "feature_names": FEATURES,
    "policy_version": policy["policy_version"],
    "dataset_sha256": sha256_file(DATASET),
    "policy_sha256": sha256_file(POLICY),
    "evaluation_manifest_sha256": policy["evaluation_manifest_sha256"],
    "model_sha256": sha256_file(artifact),
    "train_rows": int(len(train_index)),
    "holdout_rows": int(len(test_index)),
    "threshold": threshold,
    **metrics,
}
(OUT / "promotion_report.json").write_text(
    json.dumps(report, sort_keys=True, indent=2) + "\n", encoding="utf-8"
)
if status != "PASS":
    raise SystemExit("FAIL: D-008 candidate did not meet promotion policy")

Independent verification: A training-independent identity verifies the H-007.004 evaluation manifest, replays the group split and metrics from immutable dataset bytes, reviews the serialized candidate type allowlist, hashes the serialized candidate, and signs the promotion report. Thresholds come from signed deployment policy; no values shown in code are AIDEFEND defaults.

Action: Promote only a PASS candidate whose signed policy, evaluation manifest, immutable dataset, and replayed metrics all verify.

AID-D-008-G003

Serve the verified D-008 fusion artifact and emit a structured cross-detector finding without routing or enforcing a response.
Concept

At runtime, verify the signed promotion manifest, artifact digest, and exact serialized-type allowlist before deserialization, require a measured feature row, and emit a D-008 finding only. D-005 owns publication, SIEM/SOAR routing, case creation, and hunting.

Serve the verified fusion model
from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
FUSION_SERVING_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D008_SERVING_COMMAND_TIMEOUT_SECONDS")

# File: d008/serve_fusion_model.py

import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path

import skops.io as sio
import pandas as pd


MODEL = Path(os.environ["D008_MODEL_PATH"])
MANIFEST = Path(os.environ["D008_MODEL_MANIFEST"])
SIGNATURE = Path(os.environ["D008_MODEL_MANIFEST_SIGNATURE"])


_AIDEFEND_CAPTURED: dict[Path, bytes] = {}


def _aidefend_bytes(path: Path) -> bytes:
    normalized = Path(path)
    if normalized not in _AIDEFEND_CAPTURED:
        _AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
    return _AIDEFEND_CAPTURED[normalized]


def _aidefend_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 _aidefend_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 _aidefend_strict_jsonl(raw: bytes) -> list[object]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("JSONL input is empty or contains a blank record")
    return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str | Path,
    timeout_seconds: float,
) -> bytes:
    payload_path = Path(payload_path)
    proof_path = Path(proof_path)
    if payload_path == proof_path:
        raise RuntimeError("signed payload and signature paths collide")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    key_text = str(key_reference)
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
    )) or key_text.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_text.encode("utf-8")
        if key_text.startswith("-----BEGIN ")
        else Path(key_text).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "payload.sig"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_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(proof_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    _AIDEFEND_CAPTURED[payload_path] = payload_raw
    return payload_raw


def sha256_file(path: Path) -> str:
    return hashlib.sha256(_aidefend_bytes(Path(path))).hexdigest()


def load_verified_model() -> tuple[dict, dict]:
    manifest_raw = _aidefend_capture_signed_bytes(
        MANIFEST, SIGNATURE, os.environ["D008_MODEL_VERIFY_KEY"],
        FUSION_SERVING_COMMAND_TIMEOUT_SECONDS,
    )
    manifest = _aidefend_strict_json(manifest_raw)
    required_manifest = {
        "schema_version", "status", "artifact_format", "trusted_type_names",
        "feature_names", "policy_version", "dataset_sha256", "policy_sha256",
        "evaluation_manifest_sha256", "model_sha256", "train_rows",
        "holdout_rows", "threshold", "false_positive_rate", "malicious_recall",
    }
    if not isinstance(manifest, dict) or set(manifest) != required_manifest:
        raise ValueError("D-008 promotion manifest schema mismatch")
    if (
        manifest["schema_version"] != "aidefend.d008-promotion-report.v1"
        or manifest["status"] != "PASS"
        or manifest["artifact_format"] != "skops"
        or not isinstance(manifest["policy_version"], str)
        or not manifest["policy_version"].strip()
    ):
        raise ValueError("D-008 artifact was not approved for serving")
    feature_names = manifest["feature_names"]
    if (
        not isinstance(feature_names, list)
        or not feature_names
        or any(not isinstance(name, str) or not name for name in feature_names)
        or feature_names != list(dict.fromkeys(feature_names))
    ):
        raise ValueError("D-008 signed feature contract is invalid")
    trusted_types = manifest["trusted_type_names"]
    if (
        not isinstance(trusted_types, list)
        or any(not isinstance(name, str) or not name for name in trusted_types)
        or trusted_types != sorted(set(trusted_types))
    ):
        raise ValueError("D-008 serialized-type allowlist is invalid")
    if manifest["model_sha256"] != sha256_file(MODEL):
        raise ValueError("D-008 model digest differs from signed manifest")
    model_raw = _aidefend_bytes(MODEL)
    observed_types = sorted(sio.get_untrusted_types(data=model_raw))
    if observed_types != trusted_types:
        raise ValueError("D-008 serialized types differ from signed allowlist")
    bundle = sio.loads(model_raw, trusted=trusted_types)
    if (
        not isinstance(bundle, dict)
        or set(bundle) != {"model", "features", "threshold", "schema_version"}
        or bundle["schema_version"] != "aidefend.d008-fusion-model.v1"
        or not isinstance(bundle["features"], list)
        or not bundle["features"]
        or any(not isinstance(name, str) or not name for name in bundle["features"])
        or bundle["features"] != list(dict.fromkeys(bundle["features"]))
        or bundle["features"] != manifest["feature_names"]
        or isinstance(bundle["threshold"], bool)
        or not isinstance(bundle["threshold"], (int, float))
        or not math.isfinite(float(bundle["threshold"]))
        or not 0.0 <= float(bundle["threshold"]) <= 1.0
        or float(bundle["threshold"]) != float(manifest["threshold"])
    ):
        raise ValueError("D-008 model schema mismatch")
    return bundle, manifest


def score(feature_row: dict) -> dict:
    if not isinstance(feature_row, dict):
        return {"status": "ERROR", "reason": "feature row is not an object"}
    if feature_row.get("status") != "MEASURED":
        return {"status": "INSUFFICIENT_DATA", "reason": "source coverage is incomplete"}
    bundle, manifest = load_verified_model()
    required = {
        "schema_version", "policy_version", "asset_id", "session_id",
        "source_controls_covered", "source_controls_missing",
        "source_controls_insufficient", "source_controls_error",
    }
    if (
        required - set(feature_row)
        or feature_row["schema_version"] != "aidefend.d008-fusion-features.v1"
        or feature_row["policy_version"] != manifest["policy_version"]
        or any(
            not isinstance(feature_row[name], list)
            or feature_row[name] != sorted(set(feature_row[name]))
            for name in (
                "source_controls_covered", "source_controls_missing",
                "source_controls_insufficient", "source_controls_error",
            )
        )
        or len(feature_row["source_controls_covered"]) < 2
        or feature_row["source_controls_error"]
    ):
        return {"status": "ERROR", "reason": "feature-row contract is incompatible"}
    values = [feature_row.get(name) for name in bundle["features"]]
    if any(
        isinstance(value, bool) or not isinstance(value, (int, float))
        or not math.isfinite(float(value))
        for value in values
    ):
        return {"status": "ERROR", "reason": "runtime feature vector is invalid"}
    probability = float(bundle["model"].predict_proba(
        pd.DataFrame([values], columns=bundle["features"])
    )[0, 1])
    if not math.isfinite(probability) or not 0.0 <= probability <= 1.0:
        return {"status": "ERROR", "reason": "model probability is invalid"}
    threshold = float(bundle["threshold"])
    return {
        "schema_version": "aidefend.d008-finding.v1",
        "status": "FINDING" if probability >= threshold else "NO_THRESHOLD_BREACH",
        "asset_id": feature_row["asset_id"],
        "session_id": feature_row["session_id"],
        "probability": probability,
        "threshold": threshold,
        "model_sha256": manifest["model_sha256"],
        "policy_version": manifest["policy_version"],
        "source_controls_covered": feature_row["source_controls_covered"],
        "source_controls_missing": feature_row["source_controls_missing"],
    }

Evidence: Retain the signed model manifest, artifact digest, signed serialized-type allowlist, exact feature row, source finding/evidence digests, score, threshold, result, and exception class. Replay the same row against the immutable artifact. The result is probabilistic evidence and never directly authorizes containment or another side effect.

Action: Load only a signed PASS artifact whose serialized types exactly match the reviewed allowlist, emit the D-008 finding, and leave publication or response routing to D-005.

AID-D-008-G004

Retrain and roll back only the D-008 fusion model from reviewed, evidence-bound D-008 outcomes.
Concept

Retrain only the D-008 fusion model from reviewed D-008 outcomes and source-finding evidence. Do not write labels back into D-002, D-003, D-011, or H-027 corpora from this workflow.

Admit reviewed feedback
from __future__ import annotations

# File: d008/admit_feedback.py

import hashlib
import json
from pathlib import Path


REQUIRED = {
    "d008_finding_id", "feature_row_sha256", "source_evidence_sha256s",
    "analyst_disposition", "reviewer_id", "reviewed_at",
}


def admit(record: dict, out_dir: Path) -> Path:
    if not isinstance(record, dict) or set(record) != REQUIRED:
        raise ValueError("feedback schema mismatch")
    if record["analyst_disposition"] not in {"confirmed_malicious", "confirmed_benign"}:
        raise ValueError("feedback lacks a final reviewed disposition")
    evidence = record["source_evidence_sha256s"]
    if not isinstance(evidence, list) or not evidence or evidence != sorted(set(evidence)):
        raise ValueError("source evidence digests are incomplete or duplicated")
    payload = (json.dumps(
        record, sort_keys=True, separators=(",", ":"), allow_nan=False
    ) + "\n").encode("utf-8")
    digest = hashlib.sha256(payload).hexdigest()
    out_dir.mkdir(parents=True, exist_ok=True)
    target = out_dir / f"{digest}.json"
    with target.open("xb") as handle:
        handle.write(payload)
    return target
Candidate cycle and rollback

On the cadence or evidence trigger defined by signed organizational policy, build a candidate table from immutable admitted records, run the same training and independent promotion workflow above, and deploy by immutable digest. Keep the previous model and manifest available for rollback. A candidate that improves aggregate accuracy but breaches malicious-recall, benign false-positive, source-coverage, latency, or group-specific regression limits is FAIL. Missing reviewed labels or evaluation coverage is INSUFFICIENT_DATA.

Action: Record the prior/candidate digests, admitted feedback population, H-007.004 evaluation manifest, promotion report, independent signature, rollout observation, and rollback result for every D-008 model change.

Before you begin

The fusion-model lifecycle owner applies this detector stage only when reviewed D-008 outcome evidence and a signed retraining policy authorize a candidate update; raw alerts or unreviewed feedback are never automatic labels.

Open-source tools

Source-available / open-weight tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-D-008: Secondary AI Defender Fusion & Assurance | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service