AID-D-004.001: Static Artifact Integrity Re-verification & Drift Monitoring

Version 1.20261015 · Actionable sub-technique

Independently re-verify a declared population of candidate or deployed model artifacts, datasets, container image layers, libraries, and related files against the authorized signed manifests or immutable registry records established by AID-M-002.002. Emit findings for unauthorized modification, signature failure, missing or unexpected members, and drift from the approved baseline; do not turn the same release-gate verification into a second integrity PASS.

Pillars: infra, model, app

Lifecycle: building, validation, operation

Scope boundary

Owns independently scheduled or event-triggered detective reconciliation of a declared candidate or deployed static-artifact snapshot against authorized manifests or immutable registry records, including complete-population coverage and integrity-drift findings. It does not create the authoritative digest or signature verdict, admit an artifact, or decide promotion.

AID-D-004.008: AID-D-004.001 verifies static stored artifacts against approved manifests; AID-D-004.008 monitors which executable code and modules an applicable runtime actually loads. A valid file before deployment does not prove that only approved modules were loaded in the running process.

AID-D-004.002: AID-D-004.001 verifies artifact bytes and signatures; AID-D-004.002 verifies a fresh platform-backed confidential-runtime attestation measurement. Artifact integrity and attested execution environment identity are distinct cryptographic claims.

AID-M-002.002: AID-M-002.002 owns authoritative whole-artifact digest and signature creation and verification at ingestion, transfer, load, or admission; AID-D-004.001 independently replays those checks over a declared candidate or deployed snapshot and emits findings for drift. The same promotion-time verification cannot earn a second PASS under Detect, and detector coverage cannot replace artifact admission.

Implementation guidance

AID-D-004.001-G001

Independently re-verify a complete candidate snapshot against the authorized manifest and emit findings without recreating the release-admission verdict.
Consume, do not recreate, the authoritative integrity manifest

AID-M-002.002 creates and verifies the whole-artifact manifest and signature used by admission. This detector receives that exact immutable manifest and a separately mounted read-only candidate snapshot, re-enumerates the complete file population, and emits a drift finding. It must not publish a replacement manifest or claim a second promotion-time integrity verdict.

Reconcile the exact read-only candidate snapshot
# File: detection/reverify_candidate_snapshot.py
from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import stat
from pathlib import Path, PurePosixPath
from typing import Any

SHA256 = re.compile(r"^[a-f0-9]{64}$")


class DetectorError(Exception):
    pass


def positive_limit(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise DetectorError(name + " must be positive")
    return value


MAX_FILES = positive_limit("D004001_MAX_FILES")
MAX_FILE_BYTES = positive_limit("D004001_MAX_FILE_BYTES")
MAX_TOTAL_BYTES = positive_limit("D004001_MAX_TOTAL_BYTES")
MAX_MANIFEST_BYTES = positive_limit("D004001_MAX_MANIFEST_BYTES")


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise DetectorError("duplicate JSON key: " + key)
        result[key] = value
    return result


def bounded_regular(path: Path, maximum: int) -> bytes:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0))
    try:
        before = os.fstat(descriptor)
        if not stat.S_ISREG(before.st_mode) or before.st_size <= 0 or before.st_size > maximum:
            raise DetectorError("input is not a bounded regular file")
        chunks, remaining = [], before.st_size
        while remaining:
            chunk = os.read(descriptor, min(1024 * 1024, remaining))
            if not chunk:
                raise DetectorError("input ended before its declared size")
            chunks.append(chunk)
            remaining -= len(chunk)
        if os.read(descriptor, 1):
            raise DetectorError("input grew while being captured")
        after = os.fstat(descriptor)
        if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (
            after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns
        ):
            raise DetectorError("input changed while being captured")
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def manifest(path: Path) -> tuple[dict[str, Any], bytes]:
    raw = bounded_regular(path, MAX_MANIFEST_BYTES)
    try:
        value = json.loads(
            raw.decode("utf-8"), object_pairs_hook=unique_object,
            parse_constant=lambda token: (_ for _ in ()).throw(DetectorError("non-finite JSON number: " + token)),
        )
    except (UnicodeError, json.JSONDecodeError) as error:
        raise DetectorError("manifest is not strict UTF-8 JSON") from error
    if not isinstance(value, dict) or set(value) != {"schema_version", "release_id", "artifact_version", "files"}:
        raise DetectorError("manifest schema differs")
    if value["schema_version"] != "aidefend.artifact_manifest.v1":
        raise DetectorError("manifest version differs")
    if not isinstance(value["release_id"], str) or not value["release_id"]:
        raise DetectorError("release identity is absent")
    if not isinstance(value["artifact_version"], str) or not value["artifact_version"]:
        raise DetectorError("artifact version is absent")
    if not isinstance(value["files"], list) or not value["files"] or len(value["files"]) > MAX_FILES:
        raise DetectorError("manifest population is invalid")
    return value, raw


def safe_relative(value: Any) -> str:
    if not isinstance(value, str) or not value or "\\" in value or "\x00" in value:
        raise DetectorError("manifest path is invalid")
    relative = PurePosixPath(value)
    if relative.is_absolute() or any(part in {"", ".", ".."} for part in relative.parts):
        raise DetectorError("manifest path escapes the artifact root")
    canonical = relative.as_posix()
    if canonical != value:
        raise DetectorError("manifest path is not canonical")
    return canonical


def opened_digest(path: Path, device: int) -> tuple[int, str]:
    before = path.lstat()
    if not stat.S_ISREG(before.st_mode) or before.st_dev != device or before.st_size > MAX_FILE_BYTES:
        raise DetectorError("snapshot member is not an allowed regular file")
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0))
    try:
        opened = os.fstat(descriptor)
        if (opened.st_dev, opened.st_ino, opened.st_size) != (before.st_dev, before.st_ino, before.st_size):
            raise DetectorError("snapshot member changed before open")
        digest = hashlib.sha256()
        remaining = opened.st_size
        while remaining:
            chunk = os.read(descriptor, min(1024 * 1024, remaining))
            if not chunk:
                raise DetectorError("snapshot member ended before its declared size")
            digest.update(chunk)
            remaining -= len(chunk)
        if os.read(descriptor, 1):
            raise DetectorError("snapshot member grew while being read")
        after = os.fstat(descriptor)
        if (opened.st_dev, opened.st_ino, opened.st_size, opened.st_mtime_ns, opened.st_ctime_ns) != (
            after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns
        ):
            raise DetectorError("snapshot member changed while being read")
        return opened.st_size, digest.hexdigest()
    finally:
        os.close(descriptor)


def capture(root: Path) -> dict[str, dict[str, Any]]:
    if root.is_symlink():
        raise DetectorError("snapshot root is a symlink")
    root = root.resolve(strict=True)
    root_state = root.stat()
    if not stat.S_ISDIR(root_state.st_mode) or not os.statvfs(root).f_flag & os.ST_RDONLY:
        raise DetectorError("candidate snapshot must be a read-only directory")
    observed: dict[str, dict[str, Any]] = {}
    total_bytes = 0
    for path in sorted(root.rglob("*")):
        state = path.lstat()
        if stat.S_ISDIR(state.st_mode):
            if state.st_dev != root_state.st_dev:
                raise DetectorError("snapshot crosses a filesystem boundary")
            continue
        relative = PurePosixPath(path.relative_to(root).as_posix()).as_posix()
        size, digest = opened_digest(path, root_state.st_dev)
        if relative in observed:
            raise DetectorError("duplicate snapshot path")
        observed[relative] = {"size": size, "sha256": digest}
        total_bytes += size
        if len(observed) > MAX_FILES or total_bytes > MAX_TOTAL_BYTES:
            raise DetectorError("snapshot exceeds verified collection limits")
    if not observed:
        raise DetectorError("candidate snapshot is empty")
    return observed


def evaluate(manifest_path: Path, root: Path) -> dict[str, Any]:
    document, raw = manifest(manifest_path)
    expected: dict[str, dict[str, Any]] = {}
    for row in document["files"]:
        if not isinstance(row, dict) or set(row) != {"path", "size", "sha256"}:
            raise DetectorError("manifest file row differs")
        relative = safe_relative(row["path"])
        size, digest = row["size"], row["sha256"]
        if isinstance(size, bool) or not isinstance(size, int) or size < 0 or size > MAX_FILE_BYTES:
            raise DetectorError("manifest file size is invalid")
        if relative in expected or not isinstance(digest, str) or not SHA256.fullmatch(digest):
            raise DetectorError("manifest path or digest is invalid or duplicated")
        expected[relative] = {"size": size, "sha256": digest}
    observed = capture(root)
    missing = sorted(set(expected) - set(observed))
    unexpected = sorted(set(observed) - set(expected))
    changed = sorted(path for path in set(expected) & set(observed) if expected[path] != observed[path])
    population = [
        {"path": path, "size": observed[path]["size"], "sha256": observed[path]["sha256"]}
        for path in sorted(observed)
    ]
    return {
        "schema_version": "aidefend.d004001-candidate-reverification.v1",
        "control": "AID-D-004.001",
        "detector_status": "FINDING" if missing or unexpected or changed else "CLEAR",
        "release_id": document["release_id"],
        "artifact_version": document["artifact_version"],
        "manifest_sha256": hashlib.sha256(raw).hexdigest(),
        "observed_population_sha256": hashlib.sha256(
            json.dumps(population, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
        ).hexdigest(),
        "expected_count": len(expected), "observed_count": len(observed),
        "missing": missing, "unexpected": unexpected, "changed": changed,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--manifest", type=Path, required=True)
    parser.add_argument("--snapshot-root", type=Path, required=True)
    parser.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    try:
        result = evaluate(args.manifest, args.snapshot_root)
    except Exception as error:
        result = {
            "schema_version": "aidefend.d004001-candidate-reverification.v1",
            "control": "AID-D-004.001", "detector_status": "ERROR",
            "reason": "manifest or candidate-snapshot verification failed",
            "error_class": type(error).__name__,
        }
    payload = (json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n").encode("utf-8")
    descriptor = os.open(args.out, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o400)
    try:
        with os.fdopen(descriptor, "wb", closefd=False) as stream:
            stream.write(payload); stream.flush(); os.fsync(stream.fileno())
    finally:
        os.close(descriptor)
    if args.out.read_bytes() != payload:
        raise RuntimeError("detector result readback differs")
    return {"CLEAR": 0, "FINDING": 1, "ERROR": 2}[result["detector_status"]]


if __name__ == "__main__":
    raise SystemExit(main())
Verify producer evidence and retain the detector result
set -euo pipefail
umask 077
: "${D004001_MAX_FILES:?load from the verified detector profile}"
: "${D004001_MAX_FILE_BYTES:?load from the verified detector profile}"
: "${D004001_MAX_TOTAL_BYTES:?load from the verified detector profile}"
: "${D004001_MAX_MANIFEST_BYTES:?load from the verified detector profile}"
STAGE="$(mktemp -d)"
trap 'rm -rf -- "$STAGE"' EXIT HUP INT TERM
for INPUT in artifact-manifest.json artifact-manifest.sigstore.json; do
  test -f "evidence/$INPUT" && test ! -L "evidence/$INPUT"
  install -m 0400 -- "evidence/$INPUT" "$STAGE/$INPUT"
done
cosign verify-blob --key keys/artifact-integrity-authority.pub \
  --bundle "$STAGE/artifact-manifest.sigstore.json" \
  "$STAGE/artifact-manifest.json"
set +e
python detection/reverify_candidate_snapshot.py \
  --manifest "$STAGE/artifact-manifest.json" \
  --snapshot-root /mnt/read-only-candidate \
  --out "$STAGE/detector-result.json"
STATUS=$?
set -e
cosign sign-blob --yes --key env://D004001_DETECTOR_RESULT_KEY \
  --bundle "$STAGE/detector-result.sigstore.json" "$STAGE/detector-result.json"
install -m 0400 -- "$STAGE/detector-result.json" "$STAGE/detector-result.sigstore.json" evidence/
exit "$STATUS"

The manifest, bundle, trust key, detector profile, and candidate snapshot identity are deployment-selected inputs. A separate read-only verifier re-verifies the AID-M-002.002 signature and recomputes the population. Missing, unexpected, or changed members produce a finding; trust, snapshot, parsing, or publication failures produce an error. This detector publishes evidence and never promotes the artifact. AID-H-003.002 owns any release decision.

Test clean, missing, extra, modified, symlink, special-file, cross-filesystem, empty, oversized, duplicate-key, duplicate-path, wrong-key, bad-bundle, writable-snapshot, mutation-during-read, result-collision, and result-readback fixtures. The leaf may PASS only for complete detector coverage with no finding; that is not a second artifact-admission verdict.

AID-D-004.001-G002

Detect drift from approved manifests across deployed models, framework libraries, system binaries, and runtime configs using file integrity monitoring.
Evaluate an actual read-only storage snapshot

Run this detector under a dedicated read-only identity only after the storage or deployment orchestrator has mounted an immutable filesystem snapshot read-only and placed its attested snapshot ID in the signature-verified runtime profile. The release-authority manifest binds that snapshot ID, root, release, and complete regular-file population. A live writable release directory is not a stable snapshot and must return ERROR.

from __future__ import annotations

# File: /usr/local/bin/check_model_integrity.py

import hashlib
import hmac
import json
import math
import os
import posixpath
import stat
import subprocess
import tempfile
from pathlib import Path, PurePosixPath
from urllib.parse import urlsplit

import requests


def required(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError("verified runtime-profile field is absent: " + name)
    return value


def positive_float(name: str) -> float:
    value = float(required(name))
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError("verified runtime-profile field is invalid: " + name)
    return value


PROFILE_VERSION = required("AIDEFEND_RUNTIME_PROFILE_VERSION")
PROFILE_SHA256 = required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
COMMAND_TIMEOUT = positive_float("D004001_COMMAND_TIMEOUT_SECONDS")
CONNECT_TIMEOUT = positive_float("D004001_ALERT_CONNECT_TIMEOUT_SECONDS")
READ_TIMEOUT = positive_float("D004001_ALERT_READ_TIMEOUT_SECONDS")
OBSERVED_SNAPSHOT_ID = required("D004001_OBSERVED_SNAPSHOT_ID")
MANIFEST_PATH = Path(required("D004001_MANIFEST_PATH"))
MANIFEST_BUNDLE = Path(required("D004001_MANIFEST_BUNDLE_PATH"))
MANIFEST_VERIFY_KEY = Path("/etc/aidefend/d004001/trust/deployment-manifest.pub")
EVIDENCE_PATH = Path(required("D004001_EVIDENCE_PATH"))
WEBHOOK_URL = required("D004001_WEBHOOK_URL")
HEX = set("0123456789abcdef")
if len(PROFILE_SHA256) != 64 or set(PROFILE_SHA256) - HEX or PROFILE_SHA256 == "0" * 64:
    raise RuntimeError("runtime-profile digest is not a nonzero lowercase SHA-256")


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

    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )


def 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 load_signed_manifest() -> tuple[dict, str]:
    if MANIFEST_VERIFY_KEY.is_symlink():
        raise RuntimeError("manifest verification key must not be a symlink")
    key_path = MANIFEST_VERIFY_KEY.resolve(strict=True)
    if not key_path.is_file() or key_path.stat().st_mode & 0o022:
        raise RuntimeError("manifest verification key must be non-writable")
    manifest_raw = MANIFEST_PATH.read_bytes()
    bundle_raw = MANIFEST_BUNDLE.read_bytes()
    key_raw = key_path.read_bytes()
    with tempfile.TemporaryDirectory(prefix="aidefend-d004001-manifest-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        manifest_snapshot = root / "manifest.json"
        bundle_snapshot = root / "manifest.bundle"
        key_snapshot = root / "verify-key.pub"
        write_private(manifest_snapshot, manifest_raw)
        write_private(bundle_snapshot, bundle_raw)
        write_private(key_snapshot, key_raw)
        captured = [
            (manifest_snapshot, manifest_raw),
            (bundle_snapshot, bundle_raw),
            (key_snapshot, key_raw),
        ]
        try:
            subprocess.run(
                ["cosign", "verify-blob", "--key", str(key_snapshot),
                 "--bundle", str(bundle_snapshot), str(manifest_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=COMMAND_TIMEOUT,
            )
            if any(
                not hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private manifest snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)
    manifest = strict_json(manifest_raw)
    fields = {
        "schema_version", "release_id", "snapshot_id", "model_root", "artifacts",
    }
    if (
        not isinstance(manifest, dict) or set(manifest) != fields
        or manifest["schema_version"] != "aidefend.model-deployment-manifest.v2"
        or any(
            not isinstance(manifest[field], str) or not manifest[field]
            for field in ("release_id", "snapshot_id", "model_root")
        )
    ):
        raise ValueError("signed model manifest schema differs")
    return manifest, hashlib.sha256(manifest_raw).hexdigest()


def safe_relative(value: object) -> str:
    if not isinstance(value, str) or not value or "\\" in value:
        raise ValueError("manifest path is not a nonempty POSIX relative path")
    normalized = posixpath.normpath(value)
    path = PurePosixPath(value)
    if normalized != value or path.is_absolute() or any(
        part in {"", ".", ".."} for part in path.parts
    ):
        raise ValueError("manifest path escapes the signed model root")
    return path.as_posix()


def enumerate_regular_files(root: Path) -> dict[str, Path]:
    observed = {}
    for current_root, directory_names, file_names in os.walk(root, followlinks=False):
        current = Path(current_root)
        for name in directory_names:
            candidate = current / name
            if candidate.is_symlink() or not candidate.is_dir():
                raise ValueError("snapshot contains a linked or invalid directory")
        for name in file_names:
            candidate = current / name
            relative = candidate.relative_to(root).as_posix()
            if candidate.is_symlink() or not candidate.is_file():
                raise ValueError("snapshot contains a linked or special member: " + relative)
            if relative in observed:
                raise ValueError("snapshot contains a duplicate relative path")
            observed[relative] = candidate
    return observed


def read_stable_regular(path: Path) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    try:
        before = os.fstat(descriptor)
        if not stat.S_ISREG(before.st_mode):
            raise ValueError("snapshot member is not a regular file")
        chunks = []
        while True:
            chunk = os.read(descriptor, 1024 * 1024)
            if not chunk:
                break
            chunks.append(chunk)
        after = os.fstat(descriptor)
    finally:
        os.close(descriptor)
    identity_before = (
        before.st_dev, before.st_ino, before.st_size,
        before.st_mtime_ns, before.st_ctime_ns,
    )
    identity_after = (
        after.st_dev, after.st_ino, after.st_size,
        after.st_mtime_ns, after.st_ctime_ns,
    )
    raw = b"".join(chunks)
    if identity_before != identity_after or len(raw) != before.st_size:
        raise RuntimeError("snapshot member changed while being captured")
    return raw


def capture_tree(root: Path) -> tuple[tempfile.TemporaryDirectory, dict[str, str], str]:
    first_population = enumerate_regular_files(root)
    owner = tempfile.TemporaryDirectory(prefix="aidefend-d004001-tree-")
    snapshot = Path(owner.name)
    os.chmod(snapshot, 0o700)
    observed = {}
    for relative_text in sorted(first_population):
        raw = read_stable_regular(first_population[relative_text])
        relative = PurePosixPath(relative_text)
        destination = snapshot.joinpath(*relative.parts)
        destination.parent.mkdir(parents=True, exist_ok=True)
        current = destination.parent
        while current != snapshot:
            os.chmod(current, 0o700)
            current = current.parent
        write_private(destination, raw)
        if not hmac.compare_digest(destination.read_bytes(), raw):
            raise RuntimeError("private artifact snapshot readback differs")
        observed[relative_text] = hashlib.sha256(raw).hexdigest()
    second_population = enumerate_regular_files(root)
    if set(second_population) != set(first_population):
        raise RuntimeError("snapshot population changed during capture")
    canonical = (json.dumps(
        [{"path": path, "sha256": observed[path]} for path in sorted(observed)],
        sort_keys=True, separators=(",", ":"),
    ) + "\n").encode("utf-8")
    return owner, observed, hashlib.sha256(canonical).hexdigest()


def evaluate() -> dict:
    manifest, manifest_sha256 = load_signed_manifest()
    if manifest["snapshot_id"] != OBSERVED_SNAPSHOT_ID:
        raise ValueError("mounted snapshot identity differs from signed manifest")
    source_root = Path(manifest["model_root"])
    if source_root.is_symlink():
        raise ValueError("model snapshot root must not be a symlink")
    root = source_root.resolve(strict=True)
    if not root.is_dir():
        raise ValueError("model snapshot root is not a directory")
    if not os.statvfs(root).f_flag & os.ST_RDONLY:
        raise ValueError("model root must be a filesystem snapshot mounted read-only")

    artifacts = manifest["artifacts"]
    if not isinstance(artifacts, list) or not artifacts:
        raise ValueError("signed artifact population is empty")
    expected = {}
    for item in artifacts:
        if (
            not isinstance(item, dict) or set(item) != {"path", "sha256"}
            or not isinstance(item["sha256"], str)
        ):
            raise ValueError("artifact entry schema differs")
        relative = safe_relative(item["path"])
        digest = item["sha256"].lower()
        if relative in expected or len(digest) != 64 or set(digest) - HEX or digest == "0" * 64:
            raise ValueError("artifact path or digest is invalid or duplicated")
        expected[relative] = digest

    owner, observed, captured_tree_sha256 = capture_tree(root)
    missing = sorted(set(expected) - set(observed))
    unexpected = sorted(set(observed) - set(expected))
    changed = sorted(
        path for path in set(expected) & set(observed)
        if expected[path] != observed[path]
    )
    result = {
        "schema_version": "aidefend.d004001-integrity-result.v2",
        "control": "AID-D-004.001",
        "status": "FAIL" if missing or unexpected or changed else "PASS",
        "release_id": manifest["release_id"],
        "snapshot_id": manifest["snapshot_id"],
        "manifest_sha256": manifest_sha256,
        "captured_tree_sha256": captured_tree_sha256,
        "runtime_profile_version": PROFILE_VERSION,
        "runtime_profile_sha256": PROFILE_SHA256,
        "expected_count": len(expected),
        "observed_count": len(observed),
        "missing": missing,
        "unexpected": unexpected,
        "changed": changed,
    }
    owner.cleanup()
    return result


def post_finding(result: dict) -> None:
    parsed = urlsplit(WEBHOOK_URL)
    if (
        parsed.scheme != "https" or not parsed.hostname or parsed.username
        or parsed.password or parsed.fragment
    ):
        raise ValueError("alert endpoint must be one canonical HTTPS URL")
    with requests.Session() as session:
        with session.post(
            WEBHOOK_URL, json=result,
            cert=(required("D004001_ALERT_CLIENT_CERT"), required("D004001_ALERT_CLIENT_KEY")),
            verify=required("D004001_ALERT_CA_BUNDLE"),
            timeout=(CONNECT_TIMEOUT, READ_TIMEOUT),
            allow_redirects=False, stream=True,
        ) as response:
            if 300 <= response.status_code < 400:
                raise RuntimeError("alert redirect is forbidden")
            response.raise_for_status()


def publish_result(result: dict) -> None:
    payload = (json.dumps(
        result, sort_keys=True, separators=(",", ":"), allow_nan=False,
    ) + "\n").encode("utf-8")
    parent = EVIDENCE_PATH.parent.resolve(strict=True)
    parent_stat = parent.stat()
    if parent_stat.st_uid != os.geteuid() or parent_stat.st_mode & 0o022:
        raise RuntimeError("evidence directory ownership or mode is unsafe")
    destination = parent / EVIDENCE_PATH.name
    descriptor, temporary_name = tempfile.mkstemp(prefix=".d004001-result-", dir=parent)
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(payload)
            stream.flush()
            os.fsync(stream.fileno())
        os.chmod(temporary, 0o400)
        os.replace(temporary, destination)
        if not hmac.compare_digest(destination.read_bytes(), payload):
            raise RuntimeError("local integrity evidence readback differs")
    finally:
        if temporary.exists():
            temporary.unlink()


try:
    result = evaluate()
except Exception as error:
    result = {
        "schema_version": "aidefend.d004001-integrity-result.v2",
        "control": "AID-D-004.001",
        "status": "ERROR",
        "runtime_profile_version": PROFILE_VERSION,
        "runtime_profile_sha256": PROFILE_SHA256,
        "error_class": type(error).__name__,
    }
publish_result(result)
if result["status"] != "PASS":
    post_finding(result)
    raise SystemExit(1)
Schedule and independently verify

Run from a hardened systemd timer, EDR/FIM job, or deployment agent. Retain the current manifest bundle, runtime-profile identity, storage-snapshot attestation, exact population result, captured-tree digest, and local evidence digest in storage the monitored node cannot alter. An independent read-only identity mounts the same snapshot ID, re-enumerates it, captures every member once, and compares expected, observed, missing, unexpected, and changed members.

Exercise clean, missing, modified, extra, unreadable, symlink, special-file, duplicate-key, bad-bundle, wrong snapshot ID, writable mount, mutation-during-read, timeout, redirect, TLS, evidence-publication, and alert-failure fixtures. Population or digest drift is FAIL; detector, trust, snapshot, or handoff failure is ERROR. Neither state establishes a clean node, and incident response owns containment or restoration.

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.001: Static Artifact Integrity Re-verification & Drift Monitoring | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service