Scope boundary

Owns detection-rule asset versioning, event-contract dependencies, fixture compilation and replay, signed bundle promotion, deployed readback, routing, and per-rule evaluation or delivery health. It does not own telemetry collection or the underlying specialized threat semantics.

AID-D-005.001: AID-D-005.002 owns rule engineering and deployment; AID-D-005.001 owns generation, normalization, transport, and retention of the input telemetry. A rule cannot establish coverage when its required event population is absent or malformed.

AID-D-005.010: AID-D-005.002 monitors execution and delivery health for each deployed rule; AID-D-005.010 monitors the shared sensor, collector, schema, clock, sampling, queue, and exporter path. A healthy rule can still be blind because its upstream sensor failed, while a healthy pipeline does not prove each rule is deployed correctly.

AID-D-018: AID-D-005.002 proves rule packaging, deployment, routing, and technical execution health; AID-D-018 proves whether production detections actually identify the required security scenarios with acceptable efficacy and coverage. A rule that runs without errors may still miss the attack it is intended to detect.

Implementation guidance

• AID-D-005.002-G001: Maintain each AI security detection rule as a versioned control asset with an explicit event-contract dependency, owner, target platforms, routes, and fixture manifest.
• AID-D-005.002-G002: Compile and replay every target-platform rule in CI against its complete positive, negative, boundary, malformed-event, and dependency-loss fixture population.
• AID-D-005.002-G003: Promote a signed rule bundle, read its deployed identity and routes back from the target platform API, and monitor evaluation, input, error, and delivery health for the complete deployed population.
• AID-D-005.002-G004: Forward high-confidence AI security alerts into SOAR or case-management systems with stable correlation fields and response recommendations.

AID-D-005.002-G001

Rule asset contract

This sub-technique manages detector rules; it does not define the specialized detector claim. Keep each rule and its metadata in version control, bind it to the event contract owned by the relevant sensor or detector, and require a separately reviewed fixture manifest before CI accepts it.

from __future__ import annotations

# File: detection_rules/validate_manifest.py

import hashlib
import json
from pathlib import Path

REQUIRED = {
    "rule_id", "rule_version", "control_id", "owner",
    "event_contract", "event_contract_version", "platform_targets",
    "route_ids", "rule_path", "rule_sha256",
    "fixtures_manifest_sha256",
}


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


def validate_bundle(root: Path, manifest_path: Path) -> dict:
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if manifest.get("schema_version") != "aidefend.detection-rule-bundle.v1":
        raise ValueError("unsupported rule-bundle schema")
    rules = manifest.get("rules")
    if not isinstance(rules, list) or not rules:
        raise ValueError("rule population is empty")
    identifiers = []
    for item in rules:
        if set(item) != REQUIRED:
            raise ValueError("rule metadata schema mismatch")
        if any(not isinstance(item[field], str) or not item[field] for field in REQUIRED - {"platform_targets", "route_ids"}):
            raise ValueError("rule metadata text field missing")
        if any(not isinstance(item[field], list) or not item[field] for field in ("platform_targets", "route_ids")):
            raise ValueError("rule target or route population missing")
        rule_path = (root / item["rule_path"]).resolve(strict=True)
        if root.resolve(strict=True) not in rule_path.parents or sha256_file(rule_path) != item["rule_sha256"]:
            raise ValueError("rule path escaped bundle or digest mismatched")
        identifiers.append((item["rule_id"], item["rule_version"]))
    if len(identifiers) != len(set(identifiers)):
        raise ValueError("rule identity duplicated")
    return {"rule_count": len(rules), "rule_identities": sorted(identifiers), "bundle_manifest_sha256": sha256_file(manifest_path)}

Verification: reject missing owners, unknown event-contract versions, duplicate rule identities, unapproved routes, path escape, digest drift, and absent positive, negative, error, and dependency-loss fixture classes.

AID-D-005.002-G002

Platform-adapter CI gate

Use a real adapter for each declared target platform. The adapter must compile the native rule, replay timestamped immutable fixtures through the same field mapping used in production, and return the matched rule IDs plus compile or evaluation errors. No target may be marked supported from schema validation alone.

from __future__ import annotations

# File: detection_rules/ci_gate.py

from collections.abc import Callable

CompileRule = Callable[[dict, str], dict]
EvaluateRule = Callable[[dict, list[dict]], dict]

REQUIRED_CLASSES = {"positive", "negative", "boundary", "malformed_event", "dependency_loss"}


def test_rule(rule_meta: dict, rule_source: str, fixtures: list[dict], *, compile_rule: CompileRule, evaluate_rule: EvaluateRule) -> dict:
    classes = {case.get("fixture_class") for case in fixtures}
    if classes != REQUIRED_CLASSES:
        raise ValueError("fixture-class population incomplete")
    fixture_ids = [case.get("fixture_id") for case in fixtures]
    if any(not value for value in fixture_ids) or len(fixture_ids) != len(set(fixture_ids)):
        raise ValueError("fixture identity missing or duplicated")
    compiled = compile_rule(rule_meta, rule_source)
    if compiled.get("status") != "PASS" or compiled.get("rule_id") != rule_meta["rule_id"]:
        return {"rule_id": rule_meta["rule_id"], "outcome": "ERROR", "reason": "native_compile_failed"}
    observed = evaluate_rule(compiled, fixtures)
    rows = observed.get("results")
    if not isinstance(rows, list) or {row.get("fixture_id") for row in rows} != set(fixture_ids):
        return {"rule_id": rule_meta["rule_id"], "outcome": "INSUFFICIENT_DATA", "reason": "fixture_population_not_reconciled"}
    failures = []
    expected = {case["fixture_id"]: case["expected_state"] for case in fixtures}
    for row in rows:
        if row.get("observed_state") != expected[row["fixture_id"]]:
            failures.append(row["fixture_id"])
    return {"rule_id": rule_meta["rule_id"], "outcome": "PASS" if not failures else "FAIL", "failed_fixture_ids": sorted(failures), "measured_fixture_count": len(rows)}

Promotion gate: require all declared platform adapters to compile and replay successfully. Preserve native compiled artifacts, adapter versions, field-map digest, fixtures and expected states, complete results, and CI provenance. A timeout, unsupported operator, missing field, or unexecuted fixture is not a pass.

AID-D-005.002-G003

Deployment and health contract

The promotion service signs the exact bundle digest and target matrix. A separately credentialed verifier reads each platform's effective rule ID, native revision, enabled state, route bindings, and deployed digest back after deployment. The health monitor reconciles that deployment population with per-rule evaluation timestamps, eligible input counts, evaluation errors, and downstream delivery acknowledgements.

from __future__ import annotations

# File: detection_rules/health.py

from datetime import datetime, timezone


def evaluate_health(expected: list[dict], readback: list[dict], telemetry: list[dict], policy: dict) -> dict:
    expected_keys = {(item["platform"], item["rule_id"]) for item in expected}
    readback_by_key = {(item["platform"], item["rule_id"]): item for item in readback}
    telemetry_by_key = {(item["platform"], item["rule_id"]): item for item in telemetry}
    if set(readback_by_key) != expected_keys or set(telemetry_by_key) != expected_keys:
        return {"outcome": "INSUFFICIENT_DATA", "reason": "deployed_rule_population_not_reconciled"}
    findings = []
    errors = []
    now = datetime.now(timezone.utc).timestamp()
    for item in expected:
        key = (item["platform"], item["rule_id"])
        actual, health = readback_by_key[key], telemetry_by_key[key]
        if actual.get("enabled") is not True or actual.get("deployed_sha256") != item["bundle_rule_sha256"] or sorted(actual.get("route_ids", [])) != sorted(item["route_ids"]):
            findings.append({"key": key, "reason": "effective_rule_or_route_drift"})
        if health.get("evaluation_error_count", 0) > 0:
            errors.append({"key": key, "reason": "rule_evaluation_error"})
        if now - float(health["last_evaluated_at_unix"]) > float(policy["maximum_rule_idle_seconds"]):
            findings.append({"key": key, "reason": "rule_not_evaluating"})
        if health.get("eligible_input_count", 0) > 0 and health.get("delivery_ack_count", 0) < health.get("alert_emission_count", 0):
            findings.append({"key": key, "reason": "alert_delivery_gap"})
    outcome = "ERROR" if errors else ("FAIL" if findings else "PASS")
    return {"outcome": outcome, "expected_rule_count": len(expected_keys), "measured_rule_count": len(telemetry_by_key), "findings": findings, "errors": errors}

Independent verification: compare the signed bundle, native readback, route acknowledgements, and immutable health evidence from an identity that cannot deploy or edit rules. Thresholds and freshness windows come from signed organizational policy; a quiet environment is not automatically an unhealthy rule, but a rule that is not evaluating eligible input is.

AID-D-005.002-G004

Verify the runtime profile before constructing the adapter

The SOAR product remains an external dependency. This module owns only a bounded mTLS handoff contract: it verifies a signed local profile from an image-pinned trust key, validates the endpoint and deadlines, records an idempotent attempt before network I/O, and distinguishes HTTP acceptance from independent SOAR readback.

from __future__ import annotations

# File: detection/soar_handoff.py

import base64
import hashlib
import json
import math
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol
from urllib.parse import urlsplit

import requests
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


@dataclass(frozen=True)
class VerifiedRuntimeProfile:
    version: str
    sha256: str
    soar_url: str
    connect_timeout_seconds: float
    read_timeout_seconds: float
    client_cert_path: str
    client_key_path: str
    ca_bundle_path: str
    maximum_payload_bytes: int


class AttemptStore(Protocol):
    def create_pending(self, *, alert_id: str, payload_sha256: str, profile_sha256: str) -> str: ...
    def mark_transport(self, attempt_id: str, *, state: str, http_status: int | None, error_class: str | None) -> None: ...


Readback = Callable[[str], dict]
RESERVED_ALERT_FIELDS = {
    "schema_version", "runtime_profile_version", "runtime_profile_sha256",
}


def _strict_object(pairs: list[tuple[str, object]]) -> dict:
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError("duplicate runtime-profile key: " + key)
        value[key] = item
    return value


def load_verified_profile(
    *,
    profile_path: Path,
    signature_path: Path,
    trusted_public_key: Ed25519PublicKey,
) -> VerifiedRuntimeProfile:
    payload = profile_path.read_bytes()
    signature = base64.b64decode(signature_path.read_bytes(), validate=True)
    trusted_public_key.verify(signature, payload)
    value = json.loads(
        payload.decode("utf-8", errors="strict"),
        object_pairs_hook=_strict_object,
        parse_constant=lambda item: (_ for _ in ()).throw(ValueError("non-finite profile value")),
    )
    required = {
        "version", "soar_url", "connect_timeout_seconds", "read_timeout_seconds",
        "client_cert_path", "client_key_path", "ca_bundle_path", "maximum_payload_bytes",
    }
    if not isinstance(value, dict) or set(value) != required:
        raise ValueError("runtime-profile schema mismatch")
    digest = hashlib.sha256(payload).hexdigest()
    profile = VerifiedRuntimeProfile(sha256=digest, **value)
    parsed = urlsplit(profile.soar_url)
    deadlines = (float(profile.connect_timeout_seconds), float(profile.read_timeout_seconds))
    if (
        not profile.version
        or parsed.scheme != "https"
        or not parsed.hostname
        or parsed.username
        or parsed.password
        or parsed.query
        or parsed.fragment
        or any(not math.isfinite(item) or item <= 0 for item in deadlines)
        or isinstance(profile.maximum_payload_bytes, bool)
        or profile.maximum_payload_bytes < 1
    ):
        raise ValueError("verified runtime profile contains an unsafe endpoint or bound")
    return profile


def publish_to_soar(
    *,
    alert: dict,
    profile: VerifiedRuntimeProfile,
    attempts: AttemptStore,
    readback: Readback,
) -> dict:
    alert_id = alert.get("alert_id")
    if not isinstance(alert_id, str) or not alert_id:
        raise ValueError("normalized alert_id is absent")
    collisions = RESERVED_ALERT_FIELDS & set(alert)
    if collisions:
        raise ValueError(f"normalized alert overrides reserved fields: {sorted(collisions)}")
    payload = {
        **alert,
        "schema_version": "aidefend.soar_handoff.v2",
        "runtime_profile_version": profile.version,
        "runtime_profile_sha256": profile.sha256,
    }
    body = json.dumps(payload, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
    if len(body) > profile.maximum_payload_bytes:
        raise ValueError("SOAR payload exceeds the verified profile bound")
    payload_sha256 = hashlib.sha256(body).hexdigest()
    attempt_id = attempts.create_pending(
        alert_id=alert_id,
        payload_sha256=payload_sha256,
        profile_sha256=profile.sha256,
    )
    idempotency_key = hashlib.sha256(
        (alert_id + ":" + payload_sha256).encode("utf-8")
    ).hexdigest()
    try:
        response = requests.post(
            profile.soar_url,
            data=body,
            headers={"Content-Type": "application/json", "Idempotency-Key": idempotency_key},
            cert=(profile.client_cert_path, profile.client_key_path),
            verify=profile.ca_bundle_path,
            timeout=(profile.connect_timeout_seconds, profile.read_timeout_seconds),
            allow_redirects=False,
        )
        if 300 <= response.status_code < 400:
            raise RuntimeError("SOAR redirect is forbidden")
        response.raise_for_status()
        attempts.mark_transport(
            attempt_id, state="HTTP_ACCEPTED",
            http_status=response.status_code, error_class=None,
        )
    except Exception as error:
        attempts.mark_transport(
            attempt_id, state="TRANSPORT_ERROR",
            http_status=None, error_class=type(error).__name__,
        )
        raise

    observed = readback(alert_id)
    if (
        observed.get("alert_id") != alert_id
        or observed.get("payload_sha256") != payload_sha256
        or not observed.get("acknowledgement_id")
    ):
        raise RuntimeError("SOAR readback is absent or bound to another payload")
    return {
        "attempt_id": attempt_id,
        "alert_id": alert_id,
        "payload_sha256": payload_sha256,
        "acknowledgement_id": observed["acknowledgement_id"],
        "delivery_state": "READ_BACK",
    }

Integration and failure tests

Load the Ed25519 public key from a read-only trust bundle mounted by the deployment, not from the alert request. The attempt store must durably bind each alert_id to exactly one payload digest: an exact retry returns the existing attempt, while the same alert ID with different bytes is a conflict; the readback adapter uses a separate read-only SOAR identity. Test invalid profile signature, duplicate JSON keys, non-HTTPS URL, TLS failure, timeout, redirect, duplicate alert ID, HTTP acceptance without readback, wrong payload digest, and successful retry. Containment and remediation remain in Isolate/Evict controls.

Open-source tools

• OpenSearch
• Elasticsearch 8.16+ AGPLv3 source build
• Grafana Loki
• Grafana Alloy
• Wazuh
• Sigma CLI
• pySigma
• ElastAlert2

Commercial / hosted tools

• Splunk Enterprise Security
• Microsoft Sentinel
• Google Security Operations
• IBM QRadar SIEM
• Datadog Cloud SIEM
• Exabeam New-Scale SIEM
• LogRhythm SIEM

Threat mappings

MITRE ATLAS

• AML.T0107 Exploitation for Defense Evasion

MAESTRO

• Evasion of Detection (L5)
• Compromised Observability Tools (L5)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• N/A

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Platform 12.3: Lack of incident response