AID-D-004.003: Runtime Configuration & Policy Drift Detection and Monitoring

Version 1.20261015 · Actionable sub-technique

Continuously detect unauthorized or out-of-process changes to AI-serving configurations such as model-serving YAMLs, feature-store ACLs, RAG index schemas, inference-time policy files, and the deployed model-interface tuple of tokenizer, special tokens, chat template, generation configuration, and tool or reasoning parser settings. Compare effective production state with the formally approved state and emit evidence-rich drift findings.

Pillars: infra, app

Lifecycle: operation

Scope boundary

Owns observe-first declared-versus-live configuration, model-interface composition, and policy drift findings. It does not create the approved release tuple, harden repositories, block admission, or automatically reconcile the runtime.

AID-H-021.002: AID-D-004.003 detects that effective runtime configuration or policy drifted; AID-H-021.002 verifies signed runtime configuration and fails closed on unauthorized state. Observation and alerting remain separate from the enforcement point that rejects or halts the configuration.

AID-D-004.006: AID-D-004.003 detects generic configuration and policy drift; AID-D-004.006 reconciles the route actually selected after alias, default, retry, provider translation, and failover resolution. A route can drift despite unchanged configuration, and configuration drift may affect controls unrelated to routing.

AID-M-001.002: AID-M-001.002 defines and reads back the approved release-bound model-interface tuple; AID-D-004.003 continuously detects later drift in that effective tuple. Release conformance and ongoing observation use the same binding but produce different outcomes and evidence windows.

Implementation guidance

AID-D-004.003-G001

Monitor configuration-repository push events and alert on unverified or suspicious changes.
Concept:

Signed-commit requirements are preventive, but Detect still needs visibility into bypass attempts, force-pushes, and direct branch updates. Receive repository webhook events, validate the webhook signature, and forward suspicious pushes into SIEM/SOAR.

Step 1: Validate the webhook request and inspect commit verification
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")
GITHUB_REQUEST_TIMEOUT_SECONDS = _aidefend_positive_float("D004003_GITHUB_REQUEST_TIMEOUT_SECONDS")
GITHUB_MAX_RESPONSE_BYTES = _aidefend_positive_int("D004003_GITHUB_MAX_RESPONSE_BYTES")
GITHUB_RESPONSE_CHUNK_BYTES = _aidefend_positive_int("D004003_GITHUB_RESPONSE_CHUNK_BYTES")
if GITHUB_MAX_RESPONSE_BYTES < 2 or GITHUB_RESPONSE_CHUNK_BYTES > GITHUB_MAX_RESPONSE_BYTES:
    raise RuntimeError("signed GitHub response bounds are invalid")

# File: detection/config_repo_webhook.py

import hashlib
import hmac
import json
import math
import logging
import os
import re

import requests
from flask import Flask, abort, request

app = Flask(__name__)
logger = logging.getLogger("config_repo_monitor")
WEBHOOK_SECRET = os.environ["CONFIG_REPO_WEBHOOK_SECRET"].encode("utf-8")
GITHUB_TOKEN = os.environ["CONFIG_REPO_GITHUB_TOKEN"]


def verify_signature(raw_body: bytes, provided_signature: str) -> bool:
    expected = "sha256=" + hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, provided_signature)


@app.post("/webhooks/config-repo")
def handle_push():
    signature = request.headers.get("X-Hub-Signature-256", "")
    if not verify_signature(request.data, signature):
        abort(401)

    payload = request.get_json(force=True)
    repository = payload["repository"]["full_name"]
    for commit in payload.get("commits", []):
        if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository):
            abort(400)
        commit_id = commit.get("id")
        if not isinstance(commit_id, str) or not re.fullmatch(r"[0-9a-fA-F]{40}", commit_id):
            abort(400)
        with requests.get(
            f"https://api.github.com/repos/{repository}/commits/{commit_id}",
            headers={
                "Authorization": f"Bearer {GITHUB_TOKEN}",
                "Accept": "application/vnd.github+json",
                "Accept-Encoding": "identity",
                "X-GitHub-Api-Version": "2022-11-28",
            },
            timeout=GITHUB_REQUEST_TIMEOUT_SECONDS,
            allow_redirects=False,
            stream=True,
        ) as response:
            if 300 <= response.status_code < 400:
                raise RuntimeError("GitHub redirect is forbidden")
            response.raise_for_status()
            chunks, observed = [], 0
            for chunk in response.iter_content(chunk_size=GITHUB_RESPONSE_CHUNK_BYTES):
                observed += len(chunk)
                if observed > GITHUB_MAX_RESPONSE_BYTES:
                    raise RuntimeError("GitHub response exceeds signed bound")
                chunks.append(chunk)
        def reject_duplicate_keys(pairs):
            value = {}
            for key, item in pairs:
                if key in value:
                    raise ValueError(f"duplicate GitHub JSON key: {key}")
                value[key] = item
            return value
        document = json.loads(b"".join(chunks).decode("utf-8", errors="strict"), object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)))
        if not isinstance(document, dict):
            raise RuntimeError("GitHub response is not an object")
        verification = document.get("commit", {}).get("verification", {})
        if verification.get("verified") is not True:
            logger.warning(
                json.dumps(
                    {
                        "event_type": "config_repo_unverified_commit",
                        "repo": payload["repository"]["full_name"],
                        "commit_id": commit["id"],
                        "author": commit.get("author", {}).get("username"),
                    }
                )
            )
    return "", 204

Action: Alert on unsigned commits, unusual force-push activity, or direct writes to protected branches. Load the webhook secret from your secret manager or runtime environment and fail closed if it is missing.

Verify safely

An independent verifier must clone or fetch the exact refs with a separate read-only identity, verify commit ancestry and signatures, recompute changed-path sets, and compare required approvals and event coverage, read evidence back, verify signatures, digests, and immutability, and compare the canonical result digest.

AID-D-004.003-G002

Detect live AI-serving configuration drift against declared IaC.
Observe drift without mutating production

Register an Argo CD application in manual-sync mode from a signed target that includes a window ID and a minimum expected resource population. Both registration and each detector run invoke the same loader, which verifies and captures the target into a private invocation directory; neither stage reopens the original path after verification. Rollback, self-heal, and pruning remain outside this detector.

Verify and materialize one target generation
from __future__ import annotations

# File: detection/load_drift_target.py

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


FIELDS = {
    "schema_version", "policy_version", "window_id", "app_name", "project",
    "repo_url", "repo_path", "target_revision", "cluster_server", "namespace",
    "minimum_resource_count",
}
REVISION = re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$")


def strict_json(raw: bytes) -> object:
    def unique_object(pairs):
        result = {}
        for key, value in pairs:
            if key in result:
                raise ValueError("duplicate JSON key: " + key)
            result[key] = value
        return result

    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 write_private(path: Path, raw: 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(raw)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--target", type=Path, required=True)
    parser.add_argument("--bundle", type=Path, required=True)
    parser.add_argument("--key", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    timeout = float(os.environ["D004003_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError("signature-verification timeout is invalid")
    target_raw = args.target.read_bytes()
    bundle_raw = args.bundle.read_bytes()
    key_raw = args.key.read_bytes()
    with tempfile.TemporaryDirectory(prefix="aidefend-d004003-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        target_snapshot = root / "target.json"
        bundle_snapshot = root / "target.bundle"
        key_snapshot = root / "verify-key.pub"
        write_private(target_snapshot, target_raw)
        write_private(bundle_snapshot, bundle_raw)
        write_private(key_snapshot, key_raw)
        captured = [
            (target_snapshot, target_raw),
            (bundle_snapshot, bundle_raw),
            (key_snapshot, key_raw),
        ]
        try:
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", str(key_snapshot),
                    "--bundle", str(bundle_snapshot), str(target_snapshot),
                ],
                check=True,
                capture_output=True,
                text=True,
                timeout=timeout,
            )
            if any(
                not hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private target-verification snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)

    document = strict_json(target_raw)
    if (
        not isinstance(document, dict)
        or set(document) != FIELDS
        or document["schema_version"] != "aidefend.argocd-drift-target.v1"
    ):
        raise ValueError("signed drift-target schema differs")
    for field in FIELDS - {"minimum_resource_count"}:
        if not isinstance(document[field], str) or not document[field].strip():
            raise ValueError("signed drift-target field is empty: " + field)
    if REVISION.fullmatch(document["target_revision"]) is None:
        raise ValueError("target_revision is not an immutable Git object ID")
    minimum = document["minimum_resource_count"]
    if isinstance(minimum, bool) or not isinstance(minimum, int) or minimum < 1:
        raise ValueError("minimum_resource_count must be positive")
    verified = dict(document)
    verified["target_sha256"] = hashlib.sha256(target_raw).hexdigest()
    output = (
        json.dumps(verified, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
        + "\n"
    ).encode("utf-8")
    write_private(args.output, output)
    if not hmac.compare_digest(args.output.read_bytes(), output):
        raise RuntimeError("verified target readback differs")


if __name__ == "__main__":
    main()
Register the exact target in manual mode
#!/usr/bin/env bash
set -euo pipefail
snapshot="$(mktemp -d)"
trap 'rm -rf -- "$snapshot"' EXIT
chmod 0700 "$snapshot"
python detection/load_drift_target.py \
  --target policy/ai-serving-drift-target.json \
  --bundle policy/ai-serving-drift-target.bundle \
  --key /etc/aidefend/trust/d004003-target.pub \
  --output "$snapshot/target.json"

app="$(jq -er '.app_name' "$snapshot/target.json")"
project="$(jq -er '.project' "$snapshot/target.json")"
repo="$(jq -er '.repo_url' "$snapshot/target.json")"
path_in_repo="$(jq -er '.repo_path' "$snapshot/target.json")"
revision="$(jq -er '.target_revision' "$snapshot/target.json")"
server="$(jq -er '.cluster_server' "$snapshot/target.json")"
namespace="$(jq -er '.namespace' "$snapshot/target.json")"

argocd app create "$app" --upsert --sync-policy manual \
  --sync-option Prune=false --project "$project" --repo "$repo" \
  --path "$path_in_repo" --revision "$revision" \
  --dest-server "$server" --dest-namespace "$namespace"
argocd app get "$app" --hard-refresh -o json >"$snapshot/application.json"
jq -e \
  --arg project "$project" --arg repo "$repo" --arg path "$path_in_repo" \
  --arg revision "$revision" --arg server "$server" --arg namespace "$namespace" \
  '(.spec.project == $project) and (.spec.source.repoURL == $repo) and
   (.spec.source.path == $path) and (.spec.source.targetRevision == $revision) and
   (.spec.destination.server == $server) and
   (.spec.destination.namespace == $namespace) and
   ((.spec.syncPolicy.automated // null) == null)' \
  "$snapshot/application.json" >/dev/null
Emit a population-aware finding
#!/usr/bin/env bash
set -euo pipefail
snapshot="$(mktemp -d)"
trap 'rm -rf -- "$snapshot"' EXIT
chmod 0700 "$snapshot"
python detection/load_drift_target.py \
  --target policy/ai-serving-drift-target.json \
  --bundle policy/ai-serving-drift-target.bundle \
  --key /etc/aidefend/trust/d004003-target.pub \
  --output "$snapshot/target.json"
app="$(jq -er '.app_name' "$snapshot/target.json")"
window_id="$(jq -er '.window_id' "$snapshot/target.json")"
minimum="$(jq -er '.minimum_resource_count' "$snapshot/target.json")"
target_sha256="$(jq -er '.target_sha256' "$snapshot/target.json")"
argocd app get "$app" --hard-refresh -o json >"$snapshot/live.json"
live_sha256="$(sha256sum -- "$snapshot/live.json" | awk '{print $1}')"

jq -nS \
  --slurpfile target "$snapshot/target.json" \
  --slurpfile live "$snapshot/live.json" \
  --arg window_id "$window_id" --arg target_sha256 "$target_sha256" \
  --arg live_sha256 "$live_sha256" --argjson minimum "$minimum" \
  '($target[0]) as $t | ($live[0]) as $l |
   ([ $l.status.resources[]? ] | length) as $count |
   ([ $l.status.resources[]? | select((.status // "Unknown") != "Synced") ] | length) as $drifted |
   ([ $l.status.resources[]? |
      select((.health.status // "Unknown") == "Degraded" or
             (.health.status // "Unknown") == "Missing" or
             (.health.status // "Unknown") == "Unknown") ] | length) as $unhealthy |
   (($l.spec.project != $t.project) or ($l.spec.source.repoURL != $t.repo_url) or
    ($l.spec.source.path != $t.repo_path) or
    ($l.spec.source.targetRevision != $t.target_revision) or
    ($l.spec.destination.server != $t.cluster_server) or
    ($l.spec.destination.namespace != $t.namespace) or
    (($l.spec.syncPolicy.automated // null) != null)) as $spec_drift |
   (if $count < $minimum then
      {control_status:"INSUFFICIENT_DATA",finding_status:"UNKNOWN",
       reason:"observed resource population is below signed minimum"}
    elif (($l.status.sync.status // "Unknown") != "Synced") or
         (($l.status.health.status // "Unknown") == "Degraded") or
         (($l.status.health.status // "Unknown") == "Missing") or
         (($l.status.health.status // "Unknown") == "Unknown") or
         $drifted > 0 or $unhealthy > 0 or $spec_drift then
      {control_status:"PASS",finding_status:"FINDING"}
    else {control_status:"PASS",finding_status:"NO_FINDING"} end) +
   {schema_version:"aidefend.d004003-drift-result.v1",
    window_id:$window_id,app_name:$t.app_name,policy_version:$t.policy_version,
    target_sha256:$target_sha256,live_sha256:$live_sha256,
    sync_status:($l.status.sync.status // "Unknown"),
    health_status:($l.status.health.status // "Unknown"),
    observed_resource_count:$count,minimum_resource_count:$minimum,
    drifted_resource_count:$drifted,unhealthy_resource_count:$unhealthy,
    application_spec_drift:$spec_drift}' \
  >evidence/AID-D-004.003-argocd-drift.tmp
mv -- evidence/AID-D-004.003-argocd-drift.tmp \
  evidence/AID-D-004.003-argocd-drift.json
Replay and route

Run at startup and the signed cadence, retain the captured target/bundle, Argo response digest, complete resource population, result, and exceptions in the authenticated evidence path, and independently repeat the read with read-only Argo credentials. Test missing resources, an empty or partial population, per-resource and application-level drift, target revision/spec drift, a changed target between runs, corrupt bundle/key, Argo timeout, malformed JSON, and evidence-write failure. A finding is routed for investigation; configuration rollback or reconciliation remains a separately approved Harden or response action.

Open-source 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-004.003: Runtime Configuration & Policy Drift Detection and Monitoring | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service