Scope boundary

Owns release-bound monitoring of executable code and modules actually loaded by each applicable runtime, including identity, origin, digest or version, load context, required and allowed membership, and explicit coverage limits for file-backed, anonymous, or JIT code.

AID-D-004.001: AID-D-004.008 observes loaded executable code in the running process; AID-D-004.001 verifies stored artifacts before deployment or promotion. Runtime load provenance catches unexpected modules and dynamic behavior that static artifact verification alone cannot establish.

AID-D-004.002: AID-D-004.008 monitors module-load membership and provenance; AID-D-004.002 verifies a fresh confidential-runtime attestation measurement. Module visibility and platform-backed attestation are complementary runtime integrity signals.

AID-D-004.003: AID-D-004.008 observes executable code and modules; AID-D-004.003 observes runtime configuration and policy state. A process may load unauthorized code without configuration drift, or run approved code with altered policy.

Implementation guidance

• AID-D-004.008-G001: On applicable Linux serving or agent runtimes, use eBPF mmap events to trigger signed-manifest verification of file-backed executable mappings without claiming anonymous, JIT, or in-memory byte integrity.

AID-D-004.008-G001

Prerequisites and scope

Use this Linux host adapter only when the signed release policy marks it applicable and the verifier runs in the target host PID and mount namespaces. pidfd_open requires Linux 5.3 or later. Access to /proc/<pid>/map_files is subject to the kernel ptrace read check; on current Linux, resolving those links also requires CAP_SYS_ADMIN or CAP_CHECKPOINT_RESTORE in the initial user namespace. A non-Linux, anonymous/JIT-only, or policy-routed runtime is NOT_APPLICABLE to this adapter; a supposedly applicable deployment that lacks these prerequisites is ERROR, not a pass.

Verify one exact process generation

The observer signs host boot ID, host PID, process starttime, cgroup digest, release identity, sequence, and window. The verifier captures payload, Sigstore bundle, and fixed root-owned trust key before Cosign verification, opens a pidfd, briefly quiesces the exact process generation, requires two identical maps captures, and hashes each actual mapped object through map_files while checking its device and inode. It retains the raw maps and every module byte stream in a content-addressed result directory.

from __future__ import annotations

# File: runtime_mapping/verify.py
import hashlib
import hmac
import json
import math
import os
import posixpath
import re
import signal
import shutil
import stat
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

COSIGN = "/usr/local/bin/cosign"  # Pin this path in the verifier image.
MANIFEST_KEY = Path("/etc/aidefend/d004008/trust/mapping-manifest.pub")
EVENT_KEY = Path("/etc/aidefend/d004008/trust/runtime-observer.pub")
RESULT_KEY = Path("/etc/aidefend/d004008/trust/mapping-result.pub")
def required_profile_value(name):
    value = os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required signed runtime-profile field is absent: {name}")
    return value.strip()

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

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

RUNTIME_PROFILE_VERSION = required_profile_value("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = required_profile_value("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if len(RUNTIME_PROFILE_SHA256) != 64 or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef"):
    raise RuntimeError("signed runtime-profile digest must be lowercase SHA-256")
VERIFY_TIMEOUT_SECONDS = positive_profile_float("D004008_VERIFY_TIMEOUT_SECONDS")
MAX_SIGNED_INPUT_BYTES = positive_profile_int("D004008_MAX_SIGNED_INPUT_BYTES")
CHUNK_BYTES = positive_profile_int("D004008_CHUNK_BYTES")
if CHUNK_BYTES > MAX_SIGNED_INPUT_BYTES:
    raise RuntimeError("signed runtime-profile chunk size exceeds the input bound")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
OCI_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+-]*$")
PROC_ESCAPE = re.compile(r"\\([0-7]{3})")

MANIFEST_FIELDS = {
    "schema_version", "policy_version", "manifest_id", "runtime_id",
    "deployment_id", "release_id", "container_image_digest",
    "process_executable", "required_mappings", "allowed_mappings",
    "platform_exemptions", "adapter_applicable", "not_applicable_reason",
    "issued_at", "expires_at", "max_event_age_seconds",
    "max_future_skew_seconds", "quiesce_timeout_seconds", "max_mapping_bytes",
}
EVENT_FIELDS = {
    "schema_version", "event_id", "window_id", "runtime_id",
    "deployment_id", "release_id", "container_image_digest", "boot_id",
    "host_pid", "process_start_time_ticks", "cgroup_sha256",
    "observed_at", "observer_revision", "observer_sequence",
}

@dataclass(frozen=True)
class Captured:
    payload: bytes
    bundle: bytes
    key: bytes

    def digests(self):
        return {
            "payload_sha256": hashlib.sha256(self.payload).hexdigest(),
            "bundle_sha256": hashlib.sha256(self.bundle).hexdigest(),
            "verification_key_sha256": hashlib.sha256(self.key).hexdigest(),
        }

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

def strict_json(raw):
    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 bounded(value, label):
    if not isinstance(value, str) or not IDENTIFIER.fullmatch(value):
        raise ValueError(label + " is not a bounded identifier")
    return value

def timestamp(value, label):
    if not isinstance(value, str) or not value:
        raise ValueError(label + " is absent")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.utcoffset() is None:
        raise ValueError(label + " lacks a timezone")
    return parsed.astimezone(timezone.utc)

def finite(value, label, positive):
    if (
        isinstance(value, bool) or not isinstance(value, (int, float))
        or not math.isfinite(float(value))
        or (float(value) <= 0 if positive else float(value) < 0)
    ):
        raise ValueError(label + " is invalid")
    return float(value)

def canonical_path(value):
    if (
        not isinstance(value, str) or not value.startswith("/")
        or "\x00" in value or posixpath.normpath(value) != value
    ):
        raise ValueError("mapping path is not canonical and absolute")
    return value

def read_regular(path, *, trust_key=False):
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    with os.fdopen(descriptor, "rb") as stream:
        before = os.fstat(stream.fileno())
        if (
            not stat.S_ISREG(before.st_mode) or before.st_size < 1
            or before.st_size > MAX_SIGNED_INPUT_BYTES
        ):
            raise ValueError("signed input is not a bounded regular file")
        if trust_key and (before.st_uid != 0 or before.st_mode & 0o022):
            raise PermissionError("trust key is not root-owned and write-protected")
        raw = stream.read()
        after = os.fstat(stream.fileno())
    identity = lambda item: (
        item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns
    )
    if identity(before) != identity(after) or len(raw) != before.st_size:
        raise RuntimeError("signed input changed during capture")
    return raw

def write_private(path, raw):
    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 capture_verified(label, payload_path, bundle_path, key_path):
    captured = Captured(
        read_regular(Path(payload_path)),
        read_regular(Path(bundle_path)),
        read_regular(Path(key_path), trust_key=True),
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-d004008-verify-") as name:
        root = Path(name)
        os.chmod(root, 0o700)
        payload = root / (label + ".payload")
        bundle = root / (label + ".bundle")
        key = root / (label + ".pub")
        for path, raw in (
            (payload, captured.payload), (bundle, captured.bundle),
            (key, captured.key),
        ):
            write_private(path, raw)
        subprocess.run(
            [COSIGN, "verify-blob", "--key", str(key),
             "--bundle", str(bundle), str(payload)],
            check=True, capture_output=True, text=True,
            timeout=VERIFY_TIMEOUT_SECONDS,
        )
        if any(
            not hmac.compare_digest(path.read_bytes(), raw)
            for path, raw in (
                (payload, captured.payload), (bundle, captured.bundle),
                (key, captured.key),
            )
        ):
            raise RuntimeError("private Cosign snapshot changed")
    return captured

def parse_mapping_entries(manifest, field):
    entries = manifest[field]
    if not isinstance(entries, list):
        raise ValueError(field + " must be an array")
    parsed = {}
    for number, entry in enumerate(entries, 1):
        if (
            not isinstance(entry, dict) or set(entry) != {"path", "sha256"}
            or not isinstance(entry["sha256"], str)
            or not SHA256.fullmatch(entry["sha256"])
        ):
            raise ValueError(f"{field}:{number}: schema differs")
        path = canonical_path(entry["path"])
        if path in parsed:
            raise ValueError(field + " contains a duplicate path")
        parsed[path] = entry["sha256"]
    return parsed

def load_manifest(captured):
    value = strict_json(captured.payload)
    if (
        not isinstance(value, dict) or set(value) != MANIFEST_FIELDS
        or value.get("schema_version") != "aidefend.executable-mappings.v2"
    ):
        raise ValueError("mapping manifest schema differs")
    for field in ("policy_version", "manifest_id", "runtime_id",
                  "deployment_id", "release_id"):
        bounded(value[field], field)
    if not OCI_DIGEST.fullmatch(str(value["container_image_digest"])):
        raise ValueError("container image digest is invalid")
    canonical_path(value["process_executable"])
    if not isinstance(value["adapter_applicable"], bool):
        raise ValueError("adapter applicability is not boolean")
    reason = value["not_applicable_reason"]
    if value["adapter_applicable"] and reason is not None:
        raise ValueError("applicable manifest has a not-applicable reason")
    if not value["adapter_applicable"] and (
        not isinstance(reason, str) or not reason.strip()
    ):
        raise ValueError("not-applicable manifest lacks a reason")
    required = parse_mapping_entries(value, "required_mappings")
    allowed = parse_mapping_entries(value, "allowed_mappings")
    exemptions = value["platform_exemptions"]
    if (
        not isinstance(exemptions, list) or len(exemptions) != len(set(exemptions))
        or any(item not in {"[vdso]", "[vsyscall]"} for item in exemptions)
    ):
        raise ValueError("platform exemption list is invalid")
    if value["adapter_applicable"] and (
        not required or not set(required) <= set(allowed)
        or value["process_executable"] not in required
        or any(allowed[path] != digest for path, digest in required.items())
    ):
        raise ValueError("required mapping coverage is invalid")
    issued = timestamp(value["issued_at"], "issued_at")
    expires = timestamp(value["expires_at"], "expires_at")
    now = datetime.now(timezone.utc)
    if issued > now or expires <= issued or now >= expires:
        raise ValueError("mapping manifest is not currently valid")
    for field, positive in (
        ("max_event_age_seconds", True),
        ("max_future_skew_seconds", False),
        ("quiesce_timeout_seconds", True),
    ):
        finite(value[field], field, positive)
    maximum = value["max_mapping_bytes"]
    if isinstance(maximum, bool) or not isinstance(maximum, int) or maximum < 1:
        raise ValueError("max_mapping_bytes is invalid")
    return value, required, allowed, set(exemptions)

def load_event(captured, manifest):
    value = strict_json(captured.payload)
    if (
        not isinstance(value, dict) or set(value) != EVENT_FIELDS
        or value.get("schema_version") != "aidefend.exec-mapping-event.v2"
    ):
        raise ValueError("runtime observer event schema differs")
    for field in (
        "event_id", "window_id", "runtime_id", "deployment_id",
        "release_id", "boot_id", "observer_revision",
    ):
        bounded(value[field], field)
    if not OCI_DIGEST.fullmatch(str(value["container_image_digest"])):
        raise ValueError("event container image digest is invalid")
    if not SHA256.fullmatch(str(value["cgroup_sha256"])):
        raise ValueError("event cgroup digest is invalid")
    for field in ("host_pid", "observer_sequence"):
        if isinstance(value[field], bool) or not isinstance(value[field], int) or value[field] < 0:
            raise ValueError(field + " is invalid")
    if value["host_pid"] < 1 or not str(value["process_start_time_ticks"]).isdigit():
        raise ValueError("event process generation is invalid")
    for field in ("runtime_id", "deployment_id", "release_id", "container_image_digest"):
        if value[field] != manifest[field]:
            raise ValueError("event release identity differs: " + field)
    age = (datetime.now(timezone.utc) - timestamp(value["observed_at"], "observed_at")).total_seconds()
    if age > manifest["max_event_age_seconds"] or age < -manifest["max_future_skew_seconds"]:
        raise ValueError("runtime observer event is stale or future-dated")
    return value

def store_blob(stage, label, raw):
    digest = hashlib.sha256(raw).hexdigest()
    objects = stage / "objects"
    objects.mkdir(mode=0o700, exist_ok=True)
    target = objects / digest
    if target.exists():
        if not hmac.compare_digest(target.read_bytes(), raw):
            raise RuntimeError("content-addressed evidence collision")
    else:
        temporary = objects / ("." + digest + ".tmp")
        write_private(temporary, raw)
        os.replace(temporary, target)
    return {
        "label": label, "sha256": digest, "size_bytes": len(raw),
        "relative_path": "objects/" + digest,
    }

def record_capture(stage, label, captured):
    return {
        "payload": store_blob(stage, label + ":payload", captured.payload),
        "bundle": store_blob(stage, label + ":bundle", captured.bundle),
        "verification_key": store_blob(stage, label + ":key", captured.key),
    }

def process_identity(pid):
    if isinstance(pid, bool) or not isinstance(pid, int) or pid < 1:
        raise ValueError("pid must be a positive integer")
    proc = Path("/proc") / str(pid)
    stat_text = (proc / "stat").read_bytes().decode("utf-8", errors="strict").strip()
    prefix = str(pid) + " ("
    close = stat_text.rfind(")")
    if not stat_text.startswith(prefix) or close < len(prefix):
        raise ValueError("process stat record has an invalid comm field")
    suffix = stat_text[close + 1:].strip().split()
    # suffix[0] is field 3, therefore suffix[19] is field 22 (starttime).
    if len(suffix) < 20 or not suffix[19].isdigit():
        raise ValueError("process stat record lacks a valid starttime")
    executable = os.readlink(proc / "exe")
    deleted = executable.endswith(" (deleted)")
    if deleted:
        executable = executable[:-10]
    return {
        "boot_id": (Path("/proc/sys/kernel/random/boot_id")
                    .read_text(encoding="ascii").strip()),
        "pid": pid,
        "start_time_ticks": suffix[19],
        "state": suffix[0],
        "executable": canonical_path(executable),
        "executable_deleted": deleted,
        "cgroup_sha256": hashlib.sha256((proc / "cgroup").read_bytes()).hexdigest(),
    }

def wait_for_stopped(pid, timeout_seconds):
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        status = (Path("/proc") / str(pid) / "status").read_text(
            encoding="utf-8", errors="strict"
        )
        state_line = next(
            (line for line in status.splitlines() if line.startswith("State:")), None
        )
        if state_line is None:
            raise RuntimeError("process state is unavailable")
        state = state_line.split()[1]
        if state in {"T", "t"}:
            return
        time.sleep(0.01)
    raise TimeoutError("process did not quiesce within signed policy")

def decode_proc_path(value):
    decoded = PROC_ESCAPE.sub(lambda match: chr(int(match.group(1), 8)), value)
    return canonical_path(decoded)

def parse_map_row(line, number):
    parts = line.split(maxsplit=5)
    if len(parts) < 5:
        raise ValueError(f"maps:{number}: row is malformed")
    address, permissions, offset_text, device_text, inode_text = parts[:5]
    try:
        start_text, end_text = address.split("-", 1)
        start, end = int(start_text, 16), int(end_text, 16)
        offset = int(offset_text, 16)
        major_text, minor_text = device_text.split(":", 1)
        device = (int(major_text, 16), int(minor_text, 16))
        inode = int(inode_text, 10)
    except (ValueError, TypeError) as error:
        raise ValueError(f"maps:{number}: numeric field is malformed") from error
    if (
        start >= end or offset < 0 or inode < 0 or min(device) < 0
        or re.fullmatch(r"[r-][w-][x-][ps]", permissions) is None
    ):
        raise ValueError(f"maps:{number}: range or permissions are invalid")
    return {
        "range": address, "start": start, "end": end,
        "permissions": permissions, "offset": offset,
        "device": device, "inode": inode,
        "raw_path": parts[5] if len(parts) == 6 else None,
    }

def snapshot_mapping_fd(descriptor, stage, maximum_bytes):
    before = os.fstat(descriptor)
    if not stat.S_ISREG(before.st_mode):
        raise ValueError("mapped backing object is not regular")
    hasher = hashlib.sha256()
    output, temporary_name = tempfile.mkstemp(prefix=".mapping.", dir=stage)
    temporary = Path(temporary_name)
    total = 0
    try:
        with os.fdopen(output, "wb") as stream:
            while True:
                block = os.read(descriptor, CHUNK_BYTES)
                if not block:
                    break
                total += len(block)
                if total > maximum_bytes:
                    raise ValueError("mapped object exceeds signed size bound")
                hasher.update(block)
                stream.write(block)
            stream.flush()
            os.fsync(stream.fileno())
    except Exception:
        temporary.unlink(missing_ok=True)
        raise
    after = os.fstat(descriptor)
    identity = lambda item: (
        item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns
    )
    if identity(before) != identity(after) or total != before.st_size:
        temporary.unlink(missing_ok=True)
        raise RuntimeError("mapped object changed during same-fd capture")
    digest = hasher.hexdigest()
    objects = stage / "objects"
    objects.mkdir(mode=0o700, exist_ok=True)
    target = objects / digest
    if target.exists():
        temporary.unlink()
    else:
        os.chmod(temporary, 0o400)
        os.replace(temporary, target)
    return before, {
        "sha256": digest, "size_bytes": total,
        "relative_path": "objects/" + digest,
    }

def inspect_executable_maps(pid, maps_raw, stage, exemptions, maximum_bytes):
    observed = {}
    rows = []
    problems = []
    for number, line in enumerate(
        maps_raw.decode("utf-8", errors="strict").splitlines(), 1
    ):
        row = parse_map_row(line, number)
        if "x" not in row["permissions"]:
            continue
        raw_path = row.pop("raw_path")
        if raw_path in {"[vdso]", "[vsyscall]"}:
            row["platform_exemption"] = raw_path
            rows.append(row)
            if raw_path not in exemptions:
                problems.append("unapproved_platform_exemption:" + raw_path)
            continue
        if raw_path is None or not raw_path.startswith("/"):
            row["anonymous_label"] = raw_path or "unlabeled"
            rows.append(row)
            problems.append("anonymous_exec:" + (raw_path or "unlabeled"))
            continue
        if raw_path.endswith(" (deleted)"):
            row["deleted_path"] = raw_path
            rows.append(row)
            problems.append("deleted_exec:" + raw_path)
            continue
        path = decode_proc_path(raw_path)
        map_file = Path("/proc") / str(pid) / "map_files" / row["range"]
        descriptor = os.open(map_file, os.O_RDONLY | getattr(os, "O_CLOEXEC", 0))
        try:
            metadata, artifact = snapshot_mapping_fd(
                descriptor, stage, maximum_bytes
            )
        finally:
            os.close(descriptor)
        if (
            os.major(metadata.st_dev), os.minor(metadata.st_dev)
        ) != row["device"] or metadata.st_ino != row["inode"]:
            raise RuntimeError("map_files object differs from maps dev/inode")
        row.update({
            "path": path,
            "device": f"{row['device'][0]:x}:{row['device'][1]:x}",
            "artifact": artifact,
        })
        rows.append(row)
        prior = observed.setdefault(path, artifact["sha256"])
        if prior != artifact["sha256"]:
            raise RuntimeError("one mapped path resolved to different objects")
    return observed, rows, sorted(set(problems))

def verify_generation(manifest, required, allowed, exemptions, event, stage):
    pid = event["host_pid"]
    if not hasattr(os, "pidfd_open") or not hasattr(signal, "pidfd_send_signal"):
        raise RuntimeError("pidfd support is unavailable on this Linux host")
    pidfd = os.pidfd_open(pid, 0)
    stopped = False
    try:
        before = process_identity(pid)
        generation = (
            before["boot_id"], before["pid"], before["start_time_ticks"]
        )
        expected = (
            event["boot_id"], event["host_pid"],
            str(event["process_start_time_ticks"]),
        )
        if generation != expected:
            raise RuntimeError("event refers to a different process generation")
        if before["state"] in {"T", "t"}:
            raise RuntimeError("process was already stopped by another controller")
        signal.pidfd_send_signal(pidfd, signal.SIGSTOP)
        stopped = True
        wait_for_stopped(pid, manifest["quiesce_timeout_seconds"])
        stopped_identity = process_identity(pid)
        if (
            stopped_identity["boot_id"], stopped_identity["pid"],
            stopped_identity["start_time_ticks"],
        ) != expected:
            raise RuntimeError("process generation changed before maps capture")
        maps_path = Path("/proc") / str(pid) / "maps"
        maps_first = maps_path.read_bytes()
        if not maps_first:
            raise RuntimeError("executable maps population is empty")
        observed, rows, mapping_problems = inspect_executable_maps(
            pid, maps_first, stage, exemptions, manifest["max_mapping_bytes"]
        )
        maps_second = maps_path.read_bytes()
        after = process_identity(pid)
        if not hmac.compare_digest(maps_first, maps_second):
            raise RuntimeError("maps generation changed while process was quiesced")
        stable_fields = (
            "boot_id", "pid", "start_time_ticks", "executable",
            "executable_deleted", "cgroup_sha256",
        )
        if any(before[field] != after[field] for field in stable_fields):
            raise RuntimeError("process identity changed during mapping capture")
    finally:
        try:
            if stopped:
                signal.pidfd_send_signal(pidfd, signal.SIGCONT)
        finally:
            os.close(pidfd)

    identity_findings = []
    if before["executable"] != manifest["process_executable"]:
        identity_findings.append("process_executable_mismatch")
    if before["executable_deleted"]:
        identity_findings.append("process_executable_deleted")
    if before["cgroup_sha256"] != event["cgroup_sha256"]:
        identity_findings.append("cgroup_identity_mismatch")
    missing = sorted(set(required) - set(observed))
    unexpected = sorted(set(observed) - set(allowed))
    mismatches = [
        {
            "path": path, "expected_sha256": allowed[path],
            "actual_sha256": observed[path],
        }
        for path in sorted(set(observed) & set(allowed))
        if observed[path] != allowed[path]
    ]
    findings = {
        "identity_findings": identity_findings,
        "missing_required_mappings": missing,
        "unexpected_mappings": unexpected,
        "deleted_anonymous_or_exemption_findings": mapping_problems,
        "digest_mismatches": mismatches,
    }
    failed = any(findings.values())
    return {
        "schema_version": "aidefend.executable-mapping-result.v2",
        "control": "AID-D-004.008",
        "status": "FAIL" if failed else "PASS",
        "event_id": event["event_id"],
        "window_id": event["window_id"],
        "runtime_id": manifest["runtime_id"],
        "deployment_id": manifest["deployment_id"],
        "release_id": manifest["release_id"],
        "container_image_digest": manifest["container_image_digest"],
        "policy_version": manifest["policy_version"],
        "manifest_id": manifest["manifest_id"],
        "process_identity": before,
        "maps_artifact": store_blob(stage, "proc_maps", maps_first),
        "observed_mapping_rows": rows,
        "required_mapping_count": len(required),
        "allowed_mapping_count": len(allowed),
        "observed_mapping_count": len(observed),
        "findings": findings,
        "evaluated_at": datetime.now(timezone.utc).isoformat(),
    }

def publish_result(stage, final_directory, result):
    result_key_raw = read_regular(RESULT_KEY, trust_key=True)
    result["result_verification_key_sha256"] = hashlib.sha256(
        result_key_raw
    ).hexdigest()
    result_path = stage / "result.json"
    bundle_path = stage / "result.sigstore.json"
    key_snapshot = stage / "result-key.pub"
    expected = (
        (json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n")
        .encode("utf-8")
    )
    write_private(result_path, expected)
    write_private(key_snapshot, result_key_raw)
    subprocess.run(
        [COSIGN, "sign-blob", "--yes", "--key",
         "env://D004008_RESULT_SIGNING_KEY", "--bundle", str(bundle_path),
         str(result_path)],
        check=True, timeout=VERIFY_TIMEOUT_SECONDS,
    )
    os.chmod(bundle_path, 0o400)
    bundle_expected = read_regular(bundle_path)
    key_expected = key_snapshot.read_bytes()
    publication = {
        "schema_version": "aidefend.create-once-publication.v1",
        "result_sha256": hashlib.sha256(expected).hexdigest(),
        "bundle_sha256": hashlib.sha256(bundle_expected).hexdigest(),
        "verification_key_sha256": hashlib.sha256(key_expected).hexdigest(),
    }
    publication_path = stage / "publication-readback.json"
    publication_expected = (
        json.dumps(publication, sort_keys=True, separators=(",", ":")) + "\n"
    ).encode("utf-8")
    write_private(publication_path, publication_expected)
    subprocess.run(
        [COSIGN, "verify-blob", "--key", str(key_snapshot),
         "--bundle", str(bundle_path), str(result_path)],
        check=True, capture_output=True, text=True,
        timeout=VERIFY_TIMEOUT_SECONDS,
    )
    if any((
        not hmac.compare_digest(result_path.read_bytes(), expected),
        not hmac.compare_digest(bundle_path.read_bytes(), bundle_expected),
        not hmac.compare_digest(key_snapshot.read_bytes(), key_expected),
    )):
        raise RuntimeError("verified result, bundle, or key changed before publication")
    os.replace(stage, final_directory)
    parent_fd = os.open(final_directory.parent, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(parent_fd)
    finally:
        os.close(parent_fd)
    for name, wanted in (
        ("result.json", expected),
        ("result.sigstore.json", bundle_expected),
        ("result-key.pub", key_expected),
        ("publication-readback.json", publication_expected),
    ):
        if not hmac.compare_digest((final_directory / name).read_bytes(), wanted):
            raise RuntimeError("create-once publication readback differs: " + name)

def main():
    if len(sys.argv) != 6:
        raise SystemExit(
            "usage: verify.py MANIFEST MANIFEST_BUNDLE EVENT EVENT_BUNDLE OUTPUT_DIR"
        )
    final_directory = Path(sys.argv[5]).resolve(strict=False)
    if final_directory.exists() or final_directory.is_symlink():
        raise SystemExit("OUTPUT_DIR must be a new create-once result directory")
    parent = final_directory.parent
    if not parent.is_dir() or parent.is_symlink():
        raise SystemExit("OUTPUT_DIR parent must be an existing real directory")
    stage = Path(tempfile.mkdtemp(prefix=".d004008-", dir=parent))
    os.chmod(stage, 0o700)
    inputs = {}
    try:
        try:
            manifest_capture = capture_verified(
                "manifest", Path(sys.argv[1]), Path(sys.argv[2]), MANIFEST_KEY
            )
            inputs["manifest"] = record_capture(stage, "manifest", manifest_capture)
            manifest, required, allowed, exemptions = load_manifest(manifest_capture)
            if not manifest["adapter_applicable"]:
                result = {
                    "schema_version": "aidefend.executable-mapping-result.v2",
                    "control": "AID-D-004.008",
                    "status": "NOT_APPLICABLE",
                    "status_reason": manifest["not_applicable_reason"],
                    "manifest_id": manifest["manifest_id"],
                    "policy_version": manifest["policy_version"],
                    "evaluated_at": datetime.now(timezone.utc).isoformat(),
                }
            else:
                event_capture = capture_verified(
                    "event", Path(sys.argv[3]), Path(sys.argv[4]), EVENT_KEY
                )
                inputs["event"] = record_capture(stage, "event", event_capture)
                event = load_event(event_capture, manifest)
                result = verify_generation(
                    manifest, required, allowed, exemptions, event, stage
                )
        except Exception as error:
            result = {
                "schema_version": "aidefend.executable-mapping-result.v2",
                "control": "AID-D-004.008",
                "status": "ERROR",
                "status_reason": (
                    "signed-input, process-generation, quiesce, map-files, "
                    "or evidence processing failed"
                ),
                "error_class": type(error).__name__,
                "evaluated_at": datetime.now(timezone.utc).isoformat(),
            }
        result["input_artifacts"] = inputs
        publish_result(stage, final_directory, result)
    finally:
        if stage.exists():
            shutil.rmtree(stage)

if __name__ == "__main__":
    main()

Run, for example: python runtime_mapping/verify.py policy/executable-mappings.json policy/executable-mappings.sigstore.json evidence/runtime-event.json evidence/runtime-event.sigstore.json /var/lib/aidefend/results/<event-id>. The local result directory is a create-once staging publication, not WORM storage. It contains canonical result.json, its Cosign v3 bundle, the verification key snapshot, readback digest receipt, raw maps, signed inputs, and content-addressed mapped objects; the job reads back result, bundle, key, and receipt before returning. A separate retention identity must copy and read back this directory from append-only or WORM storage. A definite missing, unexpected, deleted, anonymous, unapproved-exemption, identity, or digest mismatch is FAIL; only complete zero-finding capture is PASS.

Reconcile and replay the complete population

At every signed window close, a separate read-only verifier must capture and verify the deployment controller inventory and observer-health receipt, then require one unique signed result for every expected (boot_id, host_pid, starttime) generation. Include startup and cadence scans plus mmap(PROT_EXEC), mprotect(PROT_EXEC), and pkey_mprotect(PROT_EXEC) triggers. Lost events, queue drops, missing generations, an unstable maps capture, or an unreadable applicable process prevents PASS; missing population evidence is INSUFFICIENT_DATA, while tool, signature, schema, capability, PID-reuse, or capture failure is ERROR. Replay from retained bytes only: re-verify every bundle and digest, reparse the raw maps, rehash stored objects, recompute findings, and compare canonical result bytes. This adapter verifies file-backed provenance; it does not claim the already-resident executable pages or JIT memory match disk.

Open-source tools

• Falco
• Cilium Tetragon
• osquery (loaded-module and process inventory readback)
• Cosign

Commercial / hosted tools

• Microsoft Sysmon
• Microsoft Defender for Endpoint

Threat mappings

MITRE ATLAS

• AML.T0010 AI Supply Chain Compromise
• AML.T0011 User Execution
• AML.T0018 Manipulate AI Model
• AML.T0110 AI Agent Tool Poisoning
• AML.T0110.001 AI Agent Tool Poisoning: Implementation
• AML.T0011.002 User Execution: Poisoned AI Agent Tool
• AML.T0011.000 User Execution: Unsafe AI Artifacts
• AML.T0018.002 Manipulate AI Model: Embed Malware
• AML.T0010.001 AI Supply Chain Compromise: AI Software
• AML.T0107 Exploitation for Defense Evasion

MAESTRO

• Compromised Container Images (L4)
• Supply Chain Attacks (Cross-Layer)

OWASP LLM Top 10 2026

• LLM04:2026 Supply Chain

OWASP ML Top 10 2023

• ML06:2023 AI Supply Chain Attacks

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities
• ASI05:2026 Unexpected Code Execution (RCE)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-9.3.1 Malicious Package / Tool Injection
• AISubtech-9.3.3 Dependency Replacement / Rug Pull
• AITech-9.3 Dependency / Plugin Compromise

Google Secure AI Framework 2.0 - Risks

• MDT: Model Deployment Tampering
• IIC: Insecure Integrated Component

Databricks AI Security Framework 3.0

• Algorithms 5.4: Malicious libraries
• Agents - Core 13.11: Unexpected RCE and Code Attacks