AID-D-011.003: Registered Agent Population Drift

Version 1.20261015 · Actionable sub-technique

Monitor registered agent identities for replica-count, namespace, workload, and image-lineage drift against the approved registry baseline.

Pillars: infra, app

Lifecycle: operation

Scope boundary

Owns drift findings for replica count, namespace, workload, and image lineage within the registered agent population. It does not discover or score previously unseen identities and does not execute containment.

AID-D-011.001: AID-D-011.003 measures registered-population composition and deployment drift; AID-D-011.001 measures longitudinal behavior of an individual registered agent. Population identity drift and behavioral drift are independently observable.

AID-D-016.001: AID-D-011.003 evaluates only identities already present in the approved registry baseline; AID-D-016.001 discovers unknown or newly emergent agents and their graph relationships. An unregistered identity is routed to rogue-agent discovery rather than counted as ordinary registered-population drift.

Implementation guidance

AID-D-011.003-G001

Compare only registered agent identities against the signed approved baseline and emit evidence for replica, namespace, workload, or image-lineage drift; route unknown identities to AID-D-016.001.
Compare one complete signed registry window

The registry owner signs the approved agent population and its deployment-specific replica ranges. A separately credentialed collector closes the live window, hashes the exact snapshot bytes, and signs a receipt that binds the baseline digest, subject, time window, source coverage, and row count. The evaluator captures each file once, verifies only private snapshots, and never reopens producer-controlled paths.

from __future__ import annotations

# File: monitoring/registered_agent_population.py

import argparse
from collections import defaultdict
import hashlib
import hmac
import json
import math
import os
import re
import stat
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


AGENT_FIELDS = {
    "agent_id", "namespace", "workload", "image_digest",
    "min_replicas", "max_replicas",
}
LIVE_FIELDS = {
    "source_id", "agent_id", "namespace", "workload", "image_digest",
    "replica_count", "source_sequence",
}
EXIT_CODES = {
    "PASS": 0,
    "FAIL": 1,
    "INSUFFICIENT_DATA": 2,
    "NOT_APPLICABLE": 0,
    "ERROR": 3,
}
IMAGE_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
BASELINE_KEY = Path("/etc/aidefend/d011003/trust/registry-owner.pub")
RECEIPT_KEY = Path("/etc/aidefend/d011003/trust/collector.pub")
RESULT_KEY = Path("/etc/aidefend/d011003/trust/detector.pub")
def required_profile_value(name: str) -> str:
    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: str) -> float:
    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: str) -> int:
    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("D011003_VERIFY_TIMEOUT_SECONDS")
MAX_INPUT_BYTES = positive_profile_int("D011003_MAX_INPUT_BYTES")


def reject_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number is forbidden: {value}")


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 ValueError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def strict_json(raw: bytes, label: str) -> Any:
    try:
        return json.loads(
            raw.decode("utf-8"),
            object_pairs_hook=unique_object,
            parse_constant=reject_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise ValueError(f"{label} is not strict UTF-8 JSON") from error


def canonical(value: Any) -> bytes:
    return json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")


def required_text(value: Any, label: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{label} must be a non-empty string")
    return value.strip()


def parse_time(value: Any, label: str) -> datetime:
    text = required_text(value, label)
    try:
        parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError as error:
        raise ValueError(f"{label} is not ISO 8601") from error
    if parsed.tzinfo is None:
        raise ValueError(f"{label} must include a timezone")
    return parsed.astimezone(timezone.utc)


def nonnegative_integer(value: Any, label: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise ValueError(f"{label} must be a non-negative integer")
    return value


def finite_number(value: Any, label: str, *, positive: bool) -> float:
    if (
        isinstance(value, bool)
        or not isinstance(value, (int, float))
        or not math.isfinite(float(value))
    ):
        raise ValueError(f"{label} must be a finite number")
    number = float(value)
    if (positive and number <= 0) or (not positive and number < 0):
        raise ValueError(f"{label} is outside its allowed range")
    return number


def read_stable_file(path: Path, *, require_trust: bool = False) -> bytes:
    requested = Path(os.path.abspath(path))
    resolved = path.resolve(strict=True)
    if requested != resolved or path.is_symlink():
        raise ValueError(f"symlinked or non-canonical input: {path}")
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    try:
        before = os.fstat(descriptor)
        if (
            not stat.S_ISREG(before.st_mode)
            or before.st_size < 1
            or before.st_size > MAX_INPUT_BYTES
        ):
            raise ValueError(f"input is not a bounded regular file: {path}")
        if require_trust and (before.st_uid != 0 or before.st_mode & 0o022):
            raise PermissionError(f"trust key is not root-owned and protected: {path}")
        chunks: list[bytes] = []
        while True:
            chunk = os.read(descriptor, 1024 * 1024)
            if not chunk:
                break
            chunks.append(chunk)
        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 ValueError(f"input changed during capture: {path}")
        raw = b"".join(chunks)
        if len(raw) != before.st_size:
            raise ValueError(f"captured length differs: {path}")
        return raw
    finally:
        os.close(descriptor)


def write_private(path: Path, raw: bytes) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        view = memoryview(raw)
        while view:
            written = os.write(descriptor, view)
            if written <= 0:
                raise OSError("private snapshot write made no progress")
            view = view[written:]
        os.fsync(descriptor)
    finally:
        os.close(descriptor)
    os.chmod(path, 0o400)


def capture_signed(
    *,
    payload: Path,
    bundle: Path,
    public_key: Path,
    private_root: Path,
    label: str,
    timeout_seconds: float,
) -> tuple[bytes, dict[str, str]]:
    payload_raw = read_stable_file(payload)
    bundle_raw = read_stable_file(bundle)
    key_raw = read_stable_file(public_key, require_trust=True)
    payload_copy = private_root / f"{label}.payload"
    bundle_copy = private_root / f"{label}.bundle.json"
    key_copy = private_root / f"{label}.pub"
    write_private(payload_copy, payload_raw)
    write_private(bundle_copy, bundle_raw)
    write_private(key_copy, key_raw)
    subprocess.run(
        [
            "cosign", "verify-blob", "--key", str(key_copy),
            "--bundle", str(bundle_copy), str(payload_copy),
        ],
        check=True,
        capture_output=True,
        text=True,
        timeout=timeout_seconds,
    )
    if any(
        not hmac.compare_digest(path.read_bytes(), raw)
        for path, raw in (
            (payload_copy, payload_raw), (bundle_copy, bundle_raw),
            (key_copy, key_raw),
        )
    ):
        raise RuntimeError("private verification snapshot changed")
    return payload_raw, {
        "payload_sha256": hashlib.sha256(payload_raw).hexdigest(),
        "bundle_sha256": hashlib.sha256(bundle_raw).hexdigest(),
        "verification_key_sha256": hashlib.sha256(key_raw).hexdigest(),
    }


def index_baseline(value: Any) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]:
    fields = {
        "schema_version", "baseline_version", "subject",
        "maximum_evidence_age_seconds", "maximum_future_skew_seconds",
        "required_sources", "agents",
    }
    if not isinstance(value, dict) or set(value) != fields:
        raise ValueError("baseline field set differs")
    if value["schema_version"] != "aidefend.registered_population.v2":
        raise ValueError("unsupported registered-population baseline")
    required_text(value["baseline_version"], "baseline.baseline_version")
    if not isinstance(value["subject"], dict) or set(value["subject"]) != {
        "cluster_id", "deployment_revision"
    }:
        raise ValueError("baseline subject field set differs")
    for field in ("cluster_id", "deployment_revision"):
        required_text(value["subject"][field], f"baseline.subject.{field}")
    finite_number(
        value["maximum_evidence_age_seconds"],
        "baseline.maximum_evidence_age_seconds",
        positive=True,
    )
    finite_number(
        value["maximum_future_skew_seconds"],
        "baseline.maximum_future_skew_seconds",
        positive=False,
    )
    required_sources = value["required_sources"]
    if (
        not isinstance(required_sources, list) or not required_sources
        or len(required_sources) != len(set(required_sources))
        or any(not isinstance(item, str) or not item.strip() for item in required_sources)
    ):
        raise ValueError("baseline.required_sources is invalid")
    if not isinstance(value["agents"], list):
        raise ValueError("baseline agents must be an array")
    approved: dict[str, dict[str, Any]] = {}
    for number, agent in enumerate(value["agents"], 1):
        label = f"baseline.agents:{number}"
        if not isinstance(agent, dict) or set(agent) != AGENT_FIELDS:
            raise ValueError(f"{label} field set differs")
        agent_id = required_text(agent["agent_id"], f"{label}.agent_id")
        if agent_id in approved:
            raise ValueError(f"duplicate approved agent: {agent_id}")
        for field in ("namespace", "workload"):
            required_text(agent[field], f"{label}.{field}")
        if IMAGE_DIGEST.fullmatch(str(agent["image_digest"])) is None:
            raise ValueError(f"{label}.image_digest must be an immutable SHA-256 digest")
        minimum = nonnegative_integer(agent["min_replicas"], f"{label}.min_replicas")
        maximum = nonnegative_integer(agent["max_replicas"], f"{label}.max_replicas")
        if maximum < minimum:
            raise ValueError(f"{label} maximum replicas is below minimum")
        approved[agent_id] = agent
    return value, approved


def index_live(
    value: Any,
) -> tuple[dict[str, dict[str, Any]], set[str], dict[str, list[int]]]:
    if not isinstance(value, list):
        raise ValueError("live snapshot must be an array")
    live: dict[str, dict[str, Any]] = {}
    sources: set[str] = set()
    sequences: dict[str, list[int]] = defaultdict(list)
    for number, row in enumerate(value, 1):
        label = f"live:{number}"
        if not isinstance(row, dict) or set(row) != LIVE_FIELDS:
            raise ValueError(f"{label} field set differs")
        agent_id = required_text(row["agent_id"], f"{label}.agent_id")
        if agent_id in live:
            raise ValueError(f"duplicate live agent: {agent_id}")
        source_id = required_text(row["source_id"], f"{label}.source_id")
        sequence = nonnegative_integer(
            row["source_sequence"], f"{label}.source_sequence"
        )
        sources.add(source_id)
        sequences[source_id].append(sequence)
        for field in ("namespace", "workload"):
            required_text(row[field], f"{label}.{field}")
        if IMAGE_DIGEST.fullmatch(str(row["image_digest"])) is None:
            raise ValueError(f"{label}.image_digest must be an immutable SHA-256 digest")
        nonnegative_integer(row["replica_count"], f"{label}.replica_count")
        live[agent_id] = row
    if any(len(items) != len(set(items)) for items in sequences.values()):
        raise ValueError("duplicate source_sequence in live snapshot")
    return live, sources, sequences


def evaluate(
    *,
    baseline_path: Path,
    baseline_bundle: Path,
    snapshot_path: Path,
    receipt_path: Path,
    receipt_bundle: Path,
    evaluated_at: datetime,
) -> dict[str, Any]:
    with tempfile.TemporaryDirectory(prefix="aidefend-d011003-") as directory:
        private_root = Path(directory)
        os.chmod(private_root, 0o700)
        baseline_raw, baseline_evidence = capture_signed(
            payload=baseline_path,
            bundle=baseline_bundle,
            public_key=BASELINE_KEY,
            private_root=private_root,
            label="baseline",
            timeout_seconds=VERIFY_TIMEOUT_SECONDS,
        )
        receipt_raw, receipt_evidence = capture_signed(
            payload=receipt_path,
            bundle=receipt_bundle,
            public_key=RECEIPT_KEY,
            private_root=private_root,
            label="receipt",
            timeout_seconds=VERIFY_TIMEOUT_SECONDS,
        )
        snapshot_raw = read_stable_file(snapshot_path)

    baseline, approved = index_baseline(strict_json(baseline_raw, "baseline"))
    live, live_sources, live_sequences = index_live(
        strict_json(snapshot_raw, "live snapshot")
    )
    receipt = strict_json(receipt_raw, "collection receipt")
    receipt_fields = {
        "schema_version", "window_id", "window_start", "window_end",
        "subject", "baseline_sha256", "snapshot_sha256", "snapshot_row_count",
        "expected_sources", "observed_sources", "source_checkpoints",
        "collection_complete",
    }
    if not isinstance(receipt, dict) or set(receipt) != receipt_fields:
        raise ValueError("collection receipt field set differs")
    if receipt["schema_version"] != "aidefend.population_collection.v2":
        raise ValueError("unsupported collection receipt")
    required_text(receipt["window_id"], "receipt.window_id")
    start = parse_time(receipt["window_start"], "receipt.window_start")
    end = parse_time(receipt["window_end"], "receipt.window_end")
    if start >= end:
        raise ValueError("collection window is empty or reversed")
    if receipt["subject"] != baseline["subject"]:
        raise ValueError("receipt subject differs from baseline subject")
    snapshot_digest = hashlib.sha256(snapshot_raw).hexdigest()
    if receipt["baseline_sha256"] != baseline_evidence["payload_sha256"]:
        raise ValueError("receipt baseline digest differs")
    if receipt["snapshot_sha256"] != snapshot_digest:
        raise ValueError("receipt snapshot digest differs")
    if nonnegative_integer(receipt["snapshot_row_count"], "snapshot_row_count") != len(live):
        raise ValueError("receipt snapshot row count differs")
    source_sets: dict[str, set[str]] = {}
    for field in ("expected_sources", "observed_sources"):
        values = receipt[field]
        if (
            not isinstance(values, list)
            or any(not isinstance(item, str) or not item.strip() for item in values)
            or len(values) != len(set(values))
        ):
            raise ValueError(f"receipt.{field} is invalid")
        source_sets[field] = set(values)
    baseline_sources = set(baseline["required_sources"])
    if source_sets["expected_sources"] != baseline_sources:
        raise ValueError("receipt expected sources differ from signed baseline")
    if live_sources - source_sets["observed_sources"]:
        raise ValueError("live rows name a source absent from observed_sources")
    checkpoints = receipt["source_checkpoints"]
    if not isinstance(checkpoints, dict) or set(checkpoints) != baseline_sources:
        raise ValueError("source checkpoint population differs from baseline")
    checkpoint_gaps = []
    for source in sorted(baseline_sources):
        checkpoint = checkpoints[source]
        if (
            not isinstance(checkpoint, dict)
            or set(checkpoint) != {"first_sequence", "last_sequence", "row_count", "complete"}
            or not isinstance(checkpoint["complete"], bool)
        ):
            raise ValueError("source checkpoint schema differs: " + source)
        sequences = sorted(live_sequences.get(source, []))
        row_count = nonnegative_integer(
            checkpoint["row_count"], source + ".row_count"
        )
        if row_count != len(sequences):
            raise ValueError("source checkpoint row count differs: " + source)
        if sequences:
            first = nonnegative_integer(
                checkpoint["first_sequence"], source + ".first_sequence"
            )
            last = nonnegative_integer(
                checkpoint["last_sequence"], source + ".last_sequence"
            )
            if sequences != list(range(first, last + 1)):
                raise ValueError("source sequence population is not contiguous: " + source)
        elif checkpoint["first_sequence"] is not None or checkpoint["last_sequence"] is not None:
            raise ValueError("empty source checkpoint has sequence bounds: " + source)
        if checkpoint["complete"] is not True:
            checkpoint_gaps.append({
                "reason": "source_checkpoint_incomplete", "source": source,
            })
    if not isinstance(receipt["collection_complete"], bool):
        raise ValueError("collection_complete must be a boolean")

    age = (evaluated_at - end).total_seconds()
    coverage_gaps: list[dict[str, Any]] = list(checkpoint_gaps)
    if receipt["collection_complete"] is not True:
        coverage_gaps.append({"reason": "collection_incomplete"})
    if source_sets["expected_sources"] != source_sets["observed_sources"]:
        coverage_gaps.append({
            "reason": "source_population_mismatch",
            "missing": sorted(source_sets["expected_sources"] - source_sets["observed_sources"]),
            "extra": sorted(source_sets["observed_sources"] - source_sets["expected_sources"]),
        })
    if age > float(baseline["maximum_evidence_age_seconds"]):
        coverage_gaps.append({"reason": "snapshot_stale"})
    if age < -float(baseline["maximum_future_skew_seconds"]):
        coverage_gaps.append({"reason": "snapshot_future_dated"})

    if not approved:
        status = "NOT_APPLICABLE"
        findings: list[dict[str, Any]] = []
        reason = "signed registry contains no governed agent identities"
    else:
        findings = []
        for agent_id, expected in approved.items():
            observed = live.get(agent_id)
            if observed is None:
                if expected["min_replicas"] > 0:
                    findings.append({
                        "agent_id": agent_id,
                        "reasons": ["missing_registered_identity"],
                    })
                continue
            reasons = []
            for field, drift_reason in (
                ("namespace", "namespace_drift"),
                ("workload", "workload_drift"),
                ("image_digest", "image_lineage_drift"),
            ):
                if observed[field] != expected[field]:
                    reasons.append(drift_reason)
            if not (
                expected["min_replicas"]
                <= observed["replica_count"]
                <= expected["max_replicas"]
            ):
                reasons.append("replica_count_drift")
            if reasons:
                findings.append({
                    "agent_id": agent_id,
                    "reasons": reasons,
                    "expected": expected,
                    "observed": observed,
                })
        if findings:
            status, reason = "FAIL", "registered agent population drift was confirmed"
        elif coverage_gaps:
            status, reason = "INSUFFICIENT_DATA", "registered population coverage is incomplete or stale"
        else:
            status, reason = "PASS", "complete registered population is within policy"

    delegated_unknown = sorted(set(live) - set(approved))
    return {
        "schema_version": "aidefend.registered_population_result.v2",
        "control": "AID-D-011.003",
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
        "status": status,
        "status_reason": reason,
        "subject": baseline["subject"],
        "baseline_version": baseline["baseline_version"],
        "window_id": receipt["window_id"],
        "window_start": receipt["window_start"],
        "window_end": receipt["window_end"],
        "evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
        "input_evidence": {
            "baseline": baseline_evidence,
            "snapshot": {
                "payload_sha256": snapshot_digest,
                "receipt_bound": True,
            },
            "receipt": receipt_evidence,
        },
        "registered_identity_count": len(approved),
        "observed_registered_identity_count": len(set(live) & set(approved)),
        "delegated_unknown_identities": delegated_unknown,
        "findings": findings,
        "coverage_gaps": coverage_gaps,
    }


def atomic_write(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    raw = canonical(value) + b"\n"
    descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    try:
        with os.fdopen(descriptor, "wb") as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        if read_stable_file(path) != raw:
            raise OSError("result readback differs")
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--baseline", type=Path, required=True)
    parser.add_argument("--baseline-bundle", type=Path, required=True)
    parser.add_argument("--snapshot", type=Path, required=True)
    parser.add_argument("--receipt", type=Path, required=True)
    parser.add_argument("--receipt-bundle", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    arguments = parser.parse_args()
    evaluated_at = datetime.now(timezone.utc)
    try:
        result = evaluate(
            baseline_path=arguments.baseline,
            baseline_bundle=arguments.baseline_bundle,
            snapshot_path=arguments.snapshot,
            receipt_path=arguments.receipt,
            receipt_bundle=arguments.receipt_bundle,
            evaluated_at=evaluated_at,
        )
    except Exception as error:
        result = {
            "schema_version": "aidefend.registered_population_result.v2",
            "control": "AID-D-011.003",
            "runtime_profile_version": RUNTIME_PROFILE_VERSION,
            "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
            "status": "ERROR",
            "status_reason": "signature, schema, binding, or evaluation failed",
            "error_class": type(error).__name__,
            "evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
        }
    atomic_write(arguments.output, result)
    return EXIT_CODES[result["status"]]


if __name__ == "__main__":
    raise SystemExit(main())
Sign and independently replay the result
set -euo pipefail
snapshot="$(mktemp -d)"
trap 'rm -rf -- "$snapshot"' EXIT
chmod 0700 "$snapshot"
set +e
python monitoring/registered_agent_population.py \
  --baseline policy/registered-agent-population.json \
  --baseline-bundle policy/registered-agent-population.sigstore.json \
  --snapshot evidence/registered-agent-live-snapshot.json \
  --receipt evidence/registered-agent-collection-receipt.json \
  --receipt-bundle evidence/registered-agent-collection-receipt.sigstore.json \
  --output "$snapshot/result.json"
status=$?
set -e
test -f /etc/aidefend/d011003/trust/detector.pub
test ! -L /etc/aidefend/d011003/trust/detector.pub
install -m 0400 -- /etc/aidefend/d011003/trust/detector.pub "$snapshot/detector.pub"
cosign sign-blob --yes --key env://D011003_RESULT_SIGNING_KEY \
  --bundle "$snapshot/result.sigstore.json" "$snapshot/result.json"
chmod 0400 "$snapshot/result.sigstore.json"
cosign verify-blob --key "$snapshot/detector.pub" \
  --bundle "$snapshot/result.sigstore.json" "$snapshot/result.json"
result_sha="$(sha256sum -- "$snapshot/result.json" | awk '{print $1}')"
bundle_sha="$(sha256sum -- "$snapshot/result.sigstore.json" | awk '{print $1}')"
install -m 0440 -- "$snapshot/result.sigstore.json" \
  evidence/AID-D-011.003-result.sigstore.json.next
install -m 0440 -- "$snapshot/result.json" \
  evidence/AID-D-011.003-result.json.next
mv -f -- evidence/AID-D-011.003-result.sigstore.json.next \
  evidence/AID-D-011.003-result.sigstore.json
mv -f -- evidence/AID-D-011.003-result.json.next \
  evidence/AID-D-011.003-result.json
test "$result_sha" = "$(sha256sum -- evidence/AID-D-011.003-result.json | awk '{print $1}')"
test "$bundle_sha" = "$(sha256sum -- evidence/AID-D-011.003-result.sigstore.json | awk '{print $1}')"
exit "$status"

Retain the exact baseline, live snapshot, receipt, bundles, trust keys, result, and readback digests in append-only storage. A separate identity verifies all bundles, reruns the comparison, and byte-compares a canonical replay made at the recorded evaluation time. Test missing and duplicate identities, changed namespace/workload/image lineage, low and high replica counts, missing sources, stale and future windows, a swapped snapshot, duplicate JSON keys, NaN/Infinity, and an empty signed registry.

Confirmed registered-population drift is FAIL. Incomplete authenticated source coverage produces INSUFFICIENT_DATA; unknown identities are retained and delegated to AID-D-016.001 instead of being silently treated as registered-agent drift.

A signed empty governed registry produces NOT_APPLICABLE, while processing or verification faults produce ERROR. The comparison emits PASS only when every registered identity is present and within its signed replica range.

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-011.003: Registered Agent Population Drift | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service