AID-D-011.004: Non-Human Identity & Delegated Token Abuse Detection

Version 1.20261015 · Actionable sub-technique

Detect abuse of valid agent, workload, service-principal, OAuth, and connector credentials by correlating identity-provider issuance, exchange, refresh, revocation, and active-token evidence with downstream AI, MCP, SaaS, API, and bulk-export activity. This sub-technique is designed for cases where authentication succeeds but the non-human identity, delegated token, client, audience, resource, scope, workload, fan-out, or data volume is inconsistent with signed policy.

Pillars: app, infra

Lifecycle: operation

Scope boundary

Owns detection of abuse involving valid agent, workload, service-principal, OAuth, connector, and delegated credentials by reconciling token lifecycle and downstream AI, MCP, SaaS, API, and bulk-export activity.

AID-D-011.001: AID-D-011.004 focuses on identity and token-plane inconsistencies despite successful authentication; AID-D-011.001 focuses on broad longitudinal behavior of the registered agent. Valid-token abuse may preserve normal tool sequences, while behavior anomalies do not necessarily involve credential misuse.

AID-D-016: AID-D-011.004 detects abuse by known valid identities and credentials; AID-D-016 discovers and scores unknown or emergent agent identities. A compromised registered principal and an unregistered rogue agent require different identity baselines.

Implementation guidance

AID-D-011.004-G001

Reconcile identity-bound token-lifecycle and downstream-resource events for every scoped AI non-human identity and delegated-token integration.
Use one shared event and coverage contract

The governance-owned signed policy enumerates every required telemetry cell as an exact principal, integration, lifecycle source, resource source, and probe tuple. Adapters emit the strict JSONL envelope below and use a gateway-issued HMAC correlation identifier with a signed key version; never export bearer tokens or a direct bearer-derived hash.

from __future__ import annotations

# File: monitoring/token_coverage.py
import argparse
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
import ctypes
import errno
import hashlib
import hmac
import json
import math
import os
from pathlib import Path
import re
import stat
import subprocess
import tempfile
from typing import Any

COSIGN = "/usr/local/bin/cosign"
POLICY_KEY = Path("/etc/aidefend/d011004/trust/coverage-policy.pub")
RECEIPT_KEY = Path("/etc/aidefend/d011004/trust/collector.pub")
# Non-tunable process-safety ceilings (input sizes are bytes); signed policy and callers cannot raise them.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
HARD_INPUT_CEILING = 128 * 1024 * 1024
ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+-]*$")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
EVENT_TYPES = {
    "TOKEN_ISSUED", "TOKEN_REFRESH", "TOKEN_REVOKED", "RESOURCE_ACCESS",
}
EVENT_FIELDS = {
    "schema_version", "event_id", "event_type", "occurred_at", "source_id",
    "source_sequence", "principal_id", "integration_id", "token_family_id",
    "token_instance_id", "parent_token_instance_id", "token_correlation_id",
    "correlation_key_version", "client_id", "audience", "workload_id",
    "resource_id", "scopes", "bytes_out", "expires_at",
}
EXIT_CODES = {
    "PASS": 0, "FAIL": 1, "INSUFFICIENT_DATA": 2,
    "NOT_APPLICABLE": 0, "ERROR": 3,
}

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

    def evidence(self):
        return {
            "payload_sha256": hashlib.sha256(self.raw).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, label):
    try:
        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)
            ),
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise ValueError(label + " is not strict UTF-8 JSON") from error

def strict_jsonl(raw, label):
    if not raw:
        return []
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if any(not line.strip() for line in lines):
        raise ValueError(label + " contains a blank JSONL row")
    return [strict_json(line.encode("utf-8"), label) for line in lines]

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

def text(value, label):
    if (
        not isinstance(value, str) or len(value) > 256
        or not ID.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.tzinfo is None:
        raise ValueError(label + " lacks a timezone")
    return parsed.astimezone(timezone.utc)

def integer(value, label):
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise ValueError(label + " must be a non-negative integer")
    return value

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 read_stable(path, *, trust=False, allow_empty=False):
    path = Path(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 and not allow_empty)
            or before.st_size > HARD_INPUT_CEILING
        ):
            raise ValueError("input is not a bounded regular file")
        if trust and (before.st_uid != 0 or before.st_mode & 0o022):
            raise PermissionError("trust root is not root-owned and protected")
        chunks = []
        while True:
            block = os.read(descriptor, 1024 * 1024)
            if not block:
                break
            chunks.append(block)
        after = os.fstat(descriptor)
    finally:
        os.close(descriptor)
    identity = lambda item: (
        item.st_dev, item.st_ino, item.st_size,
        item.st_mtime_ns, item.st_ctime_ns,
    )
    raw = b"".join(chunks)
    if identity(before) != identity(after) or len(raw) != before.st_size:
        raise RuntimeError("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_signed(
    label, payload, bundle, key, *, expected_payload_sha256=None,
    expected_bundle_sha256=None, expected_key_sha256=None,
):
    captured = Captured(
        read_stable(payload), read_stable(bundle), read_stable(key, trust=True)
    )
    evidence = captured.evidence()
    expected = {
        "payload_sha256": expected_payload_sha256,
        "bundle_sha256": expected_bundle_sha256,
        "verification_key_sha256": expected_key_sha256,
    }
    for field, digest in expected.items():
        if digest is not None:
            if not isinstance(digest, str) or not SHA256.fullmatch(digest):
                raise ValueError(label + " expected " + field + " is invalid")
            if not hmac.compare_digest(evidence[field], digest):
                raise ValueError(label + " captured " + field + " differs")
    with tempfile.TemporaryDirectory(prefix="aidefend-d011004-") as name:
        root = Path(name)
        os.chmod(root, 0o700)
        payload_copy = root / (label + ".payload")
        bundle_copy = root / (label + ".sigstore.json")
        key_copy = root / (label + ".pub")
        for path, raw in (
            (payload_copy, captured.raw), (bundle_copy, captured.bundle),
            (key_copy, captured.key),
        ):
            write_private(path, 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=COSIGN_DEADLINE_CEILING_SECONDS,
        )
        if any(
            not hmac.compare_digest(path.read_bytes(), raw)
            for path, raw in (
                (payload_copy, captured.raw), (bundle_copy, captured.bundle),
                (key_copy, captured.key),
            )
        ):
            raise RuntimeError("private verification snapshot changed")
    return captured

def validate_event(row, label):
    if (
        not isinstance(row, dict) or set(row) != EVENT_FIELDS
        or row.get("schema_version") != "aidefend.delegated-token-event.v2"
        or row.get("event_type") not in EVENT_TYPES
    ):
        raise ValueError(label + ": event schema differs")
    for field in (
        "event_id", "source_id", "principal_id", "integration_id",
        "token_family_id", "token_instance_id", "token_correlation_id",
        "correlation_key_version", "client_id", "audience", "workload_id",
    ):
        text(row[field], label + ":" + field)
    integer(row["source_sequence"], label + ":source_sequence")
    occurred = timestamp(row["occurred_at"], label + ":occurred_at")
    scopes = row["scopes"]
    if (
        not isinstance(scopes, list) or not scopes
        or len(scopes) > 256
        or len(scopes) != len(set(scopes))
        or any(not isinstance(item, str) or not ID.fullmatch(item) for item in scopes)
    ):
        raise ValueError(label + ": scopes are invalid")
    integer(row["bytes_out"], label + ":bytes_out")
    event_type = row["event_type"]
    if event_type == "TOKEN_REFRESH":
        text(row["parent_token_instance_id"], label + ":parent_token_instance_id")
    elif row["parent_token_instance_id"] is not None:
        raise ValueError(label + ": only refresh may name a parent token")
    if event_type in {"TOKEN_ISSUED", "TOKEN_REFRESH"}:
        expires = timestamp(row["expires_at"], label + ":expires_at")
        if expires <= occurred or row["resource_id"] is not None or row["bytes_out"] != 0:
            raise ValueError(label + ": issued/refresh payload is inconsistent")
    elif event_type == "RESOURCE_ACCESS":
        text(row["resource_id"], label + ":resource_id")
        if row["expires_at"] is not None:
            raise ValueError(label + ": resource event must not redefine expiry")
    elif row["resource_id"] is not None or row["expires_at"] is not None or row["bytes_out"] != 0:
        raise ValueError(label + ": revocation payload is inconsistent")
    return row

def atomic_write(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    raw = canonical(value) + b"\n"
    descriptor, temporary = tempfile.mkstemp(prefix="." + path.name, dir=path.parent)
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(raw)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
        if not hmac.compare_digest(read_stable(path), raw):
            raise RuntimeError("result readback differs")
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)

def fsync_directory(path):
    if os.name == "nt":
        return
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)

def fsync_file(path):
    if os.name == "nt":
        return
    descriptor = os.open(
        path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    )
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)

def rename_noreplace(source, target):
    if os.name == "nt":
        os.rename(source, target)
        return
    libc = ctypes.CDLL(None, use_errno=True)
    renameat2 = getattr(libc, "renameat2", None)
    if renameat2 is None:
        raise RuntimeError("atomic no-replace directory rename is unavailable")
    renameat2.argtypes = [
        ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p,
        ctypes.c_uint,
    ]
    renameat2.restype = ctypes.c_int
    result = renameat2(
        -100, os.fsencode(source), -100, os.fsencode(target), 1
    )
    if result != 0:
        number = ctypes.get_errno()
        if number == errno.EEXIST:
            raise FileExistsError(number, os.strerror(number), str(target))
        raise OSError(number, os.strerror(number), str(target))

def publish_result_directory(staging, target):
    staging = Path(staging)
    target = Path(target)
    if (
        not staging.is_dir() or staging.is_symlink()
        or staging.parent.is_symlink() or target.parent.is_symlink()
        or staging.parent.resolve(strict=True) != target.parent.resolve(strict=True)
        or not SHA256.fullmatch(target.name)
    ):
        raise ValueError("publication paths are not a safe same-parent staging pair")
    expected_names = {"result.json", "result.sigstore.json", "result.pub"}
    if {path.name for path in staging.iterdir()} != expected_names:
        raise ValueError("staging directory is incomplete or contains extra files")
    if any((staging / name).is_symlink() for name in expected_names):
        raise ValueError("staging directory contains a symlink")
    files = {
        name: read_stable(staging / name)
        for name in sorted(expected_names)
    }
    for name in expected_names:
        os.chmod(staging / name, 0o440)
        fsync_file(staging / name)
    digests = {
        name: hashlib.sha256(raw).hexdigest()
        for name, raw in files.items()
    }
    if not hmac.compare_digest(digests["result.json"], target.name):
        raise ValueError("target directory does not match captured result bytes")
    completion = {
        "schema_version": "aidefend.signed-result-publication.v1",
        "payload_sha256": digests["result.json"],
        "bundle_sha256": digests["result.sigstore.json"],
        "verification_key_sha256": digests["result.pub"],
    }
    completion_raw = canonical(completion) + b"\n"
    write_private(staging / "complete.json", completion_raw)
    os.chmod(staging / "complete.json", 0o440)
    os.chmod(staging, 0o750)
    fsync_directory(staging)
    rename_noreplace(staging, target)
    fsync_directory(target.parent)
    if {path.name for path in target.iterdir()} != expected_names | {"complete.json"}:
        raise RuntimeError("published directory contents differ")
    if not hmac.compare_digest(read_stable(target / "complete.json"), completion_raw):
        raise RuntimeError("publication manifest readback differs")
    for name, raw in files.items():
        if not hmac.compare_digest(read_stable(target / name), raw):
            raise RuntimeError("published " + name + " readback differs")
    return completion

def load_coverage_policy(captured):
    policy = strict_json(captured.raw, "coverage policy")
    policy_fields = {
        "schema_version", "policy_version", "applicable",
        "not_applicable_reason", "correlation_id_scheme", "required_paths",
        "maximum_evidence_age_seconds", "maximum_future_skew_seconds",
    }
    if (
        not isinstance(policy, dict) or set(policy) != policy_fields
        or policy.get("schema_version") != "aidefend.token-coverage-policy.v2"
        or not isinstance(policy.get("applicable"), bool)
    ):
        raise ValueError("coverage policy schema differs")
    text(policy["policy_version"], "policy_version")
    reason = policy["not_applicable_reason"]
    if policy["applicable"] and reason is not None:
        raise ValueError("applicable policy contains a not-applicable reason")
    if not policy["applicable"] and (
        not isinstance(reason, str) or not reason.strip()
    ):
        raise ValueError("not-applicable policy lacks a reason")
    scheme = policy["correlation_id_scheme"]
    if (
        not isinstance(scheme, dict) or set(scheme) != {"name", "key_version"}
        or scheme.get("name") != "gateway_hmac_id"
    ):
        raise ValueError("privacy-safe correlation scheme differs")
    text(scheme["key_version"], "correlation key version")
    stale = finite(
        policy["maximum_evidence_age_seconds"], "maximum_evidence_age_seconds", True
    )
    future = finite(
        policy["maximum_future_skew_seconds"], "maximum_future_skew_seconds", False
    )
    required_paths = policy["required_paths"]
    path_fields = {
        "path_id", "principal_id", "integration_id",
        "lifecycle_source", "resource_source", "probe_id",
    }
    if not isinstance(required_paths, list):
        raise ValueError("required_paths must be an array")
    path_index = {}
    path_signatures = set()
    for number, row in enumerate(required_paths, 1):
        if not isinstance(row, dict) or set(row) != path_fields:
            raise ValueError(f"required_path:{number}: schema differs")
        for field in path_fields:
            text(row[field], f"required_path:{number}:{field}")
        if row["path_id"] in path_index:
            raise ValueError("duplicate required path_id")
        signature = tuple(row[field] for field in sorted(path_fields - {"path_id"}))
        if signature in path_signatures:
            raise ValueError("duplicate required telemetry tuple")
        path_signatures.add(signature)
        path_index[row["path_id"]] = row
    if policy["applicable"] and not path_index:
        raise ValueError("applicable policy has no required telemetry path")
    if not policy["applicable"] and path_index:
        raise ValueError("not-applicable policy must not enumerate telemetry paths")
    return policy, scheme, stale, future, path_index

def evaluate(policy_path, policy_bundle, events_path, receipt_path, receipt_bundle,
             evaluated_at):
    policy_capture = capture_signed(
        "coverage-policy", policy_path, policy_bundle, POLICY_KEY
    )
    policy, scheme, stale, future, path_index = load_coverage_policy(
        policy_capture
    )
    policy_sha = policy_capture.evidence()["payload_sha256"]
    required_paths = policy["required_paths"]
    if not policy["applicable"]:
        return {
            "schema_version": "aidefend.delegated-token-coverage-result.v2",
            "control": "AID-D-011.004", "guidance": "G001",
            "status": "NOT_APPLICABLE",
            "status_reason": policy["not_applicable_reason"],
            "policy_version": policy["policy_version"],
            "coverage_policy_sha256": policy_sha,
            "required_path_count": 0, "required_paths": [],
            "population_complete": False,
            "input_evidence": {"policy": policy_capture.evidence()},
            "evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
        }

    receipt_capture = capture_signed(
        "coverage-receipt", receipt_path, receipt_bundle, RECEIPT_KEY
    )
    events_raw = read_stable(events_path, allow_empty=True)

    receipt = strict_json(receipt_capture.raw, "coverage receipt")
    receipt_fields = {
        "schema_version", "window_id", "collection_start", "window_start",
        "window_end", "observation_through", "policy_sha256", "events_sha256",
        "event_count", "source_checkpoints", "probe_results",
        "rejected_row_count", "dead_letter_count", "collection_complete",
    }
    if (
        not isinstance(receipt, dict) or set(receipt) != receipt_fields
        or receipt.get("schema_version") != "aidefend.token-coverage-receipt.v2"
    ):
        raise ValueError("coverage receipt schema differs")
    text(receipt["window_id"], "window_id")
    collection_start = timestamp(receipt["collection_start"], "collection_start")
    window_start = timestamp(receipt["window_start"], "window_start")
    window_end = timestamp(receipt["window_end"], "window_end")
    observation_through = timestamp(
        receipt["observation_through"], "observation_through"
    )
    if not collection_start <= window_start < window_end <= observation_through:
        raise ValueError("coverage window bounds are not ordered")
    events_sha = hashlib.sha256(events_raw).hexdigest()
    if receipt["policy_sha256"] != policy_sha or receipt["events_sha256"] != events_sha:
        raise ValueError("receipt does not bind exact policy/events bytes")
    events = [
        validate_event(row, f"event:{number}")
        for number, row in enumerate(strict_jsonl(events_raw, "events"), 1)
    ]
    if integer(receipt["event_count"], "event_count") != len(events):
        raise ValueError("receipt event count differs")
    if len({row["event_id"] for row in events}) != len(events):
        raise ValueError("duplicate event_id")
    if len({(row["source_id"], row["source_sequence"]) for row in events}) != len(events):
        raise ValueError("duplicate source sequence")
    if any(row["correlation_key_version"] != scheme["key_version"] for row in events):
        raise ValueError("event correlation key version differs from policy")
    if any(
        not collection_start <= timestamp(row["occurred_at"], "occurred_at")
        <= observation_through
        for row in events
    ):
        raise ValueError("event lies outside collected interval")

    expected_sources = {
        row[field] for row in required_paths
        for field in ("lifecycle_source", "resource_source")
    }
    checkpoints = receipt["source_checkpoints"]
    if not isinstance(checkpoints, dict) or set(checkpoints) != expected_sources:
        raise ValueError("source checkpoint population differs")
    by_source = defaultdict(list)
    for row in events:
        if row["source_id"] not in expected_sources:
            raise ValueError("event names an unapproved source")
        by_source[row["source_id"]].append(row["source_sequence"])
    source_gaps = []
    for source in sorted(expected_sources):
        checkpoint = checkpoints[source]
        if (
            not isinstance(checkpoint, dict)
            or set(checkpoint) != {"first", "last", "row_count", "complete"}
            or not isinstance(checkpoint["complete"], bool)
        ):
            raise ValueError("source checkpoint schema differs: " + source)
        sequences = sorted(by_source[source])
        if integer(checkpoint["row_count"], source + ":row_count") != len(sequences):
            raise ValueError("source row count differs: " + source)
        if sequences:
            first = integer(checkpoint["first"], source + ":first")
            last = integer(checkpoint["last"], source + ":last")
            if sequences != list(range(first, last + 1)):
                raise ValueError("source sequence population is not contiguous: " + source)
        elif checkpoint["first"] is not None or checkpoint["last"] is not None:
            raise ValueError("empty source has non-empty sequence bounds")
        if checkpoint["complete"] is not True:
            source_gaps.append({"source": source, "reason": "checkpoint_incomplete"})

    probes = receipt["probe_results"]
    if not isinstance(probes, list):
        raise ValueError("probe_results must be an array")
    probe_index = {}
    for number, row in enumerate(probes, 1):
        if not isinstance(row, dict) or set(row) != {
            "path_id", "probe_id", "observed", "checked_at"
        }:
            raise ValueError(f"probe:{number}: schema differs")
        path_id = text(row["path_id"], f"probe:{number}:path_id")
        probe_id = text(row["probe_id"], f"probe:{number}:probe_id")
        if path_id in probe_index or not isinstance(row["observed"], bool):
            raise ValueError(f"probe:{number}: duplicate or invalid")
        if path_id not in path_index or probe_id != path_index[path_id]["probe_id"]:
            raise ValueError(f"probe:{number}: policy binding differs")
        checked = timestamp(row["checked_at"], f"probe:{number}:checked_at")
        if not window_start <= checked <= observation_through:
            raise ValueError(f"probe:{number}: timestamp outside window")
        probe_index[path_id] = row
    if set(probe_index) != set(path_index):
        raise ValueError("probe population differs from required path matrix")
    if not isinstance(receipt["collection_complete"], bool):
        raise ValueError("collection_complete must be boolean")
    rejected = integer(receipt["rejected_row_count"], "rejected_row_count")
    dead_letters = integer(receipt["dead_letter_count"], "dead_letter_count")

    coverage_gaps = list(source_gaps)
    if receipt["collection_complete"] is not True:
        coverage_gaps.append({"reason": "collection_incomplete"})
    if rejected or dead_letters:
        coverage_gaps.append({
            "reason": "rejected_or_dead_letter_events",
            "rejected_row_count": rejected,
            "dead_letter_count": dead_letters,
        })
    age = (evaluated_at - observation_through).total_seconds()
    if age > stale:
        coverage_gaps.append({"reason": "evidence_stale"})
    if age < -future:
        coverage_gaps.append({"reason": "evidence_future_dated"})
    broken_paths = sorted(
        path_id for path_id, probe in probe_index.items() if probe["observed"] is not True
    )
    principals = {row["principal_id"] for row in required_paths}
    delegated = sorted({row["principal_id"] for row in events} - principals)
    population_complete = not coverage_gaps and not broken_paths
    if broken_paths:
        status, status_reason = "FAIL", "required telemetry probe failed"
    elif coverage_gaps:
        status, status_reason = (
            "INSUFFICIENT_DATA",
            "event population is incomplete, stale, or future-dated",
        )
    else:
        status, status_reason = "PASS", "all required telemetry paths are complete"
    return {
        "schema_version": "aidefend.delegated-token-coverage-result.v2",
        "control": "AID-D-011.004", "guidance": "G001",
        "status": status, "status_reason": status_reason,
        "policy_version": policy["policy_version"],
        "coverage_policy_sha256": policy_sha,
        "window_id": receipt["window_id"],
        "collection_start": receipt["collection_start"],
        "window_start": receipt["window_start"],
        "window_end": receipt["window_end"],
        "observation_through": receipt["observation_through"],
        "events_sha256": events_sha, "event_count": len(events),
        "required_path_count": len(path_index),
        "required_paths": required_paths,
        "probe_results": probes, "broken_path_ids": broken_paths,
        "source_checkpoints": checkpoints, "coverage_gaps": coverage_gaps,
        "rejected_row_count": rejected, "dead_letter_count": dead_letters,
        "population_complete": population_complete,
        "delegated_unknown_principals": delegated,
        "input_evidence": {
            "policy": policy_capture.evidence(),
            "receipt": receipt_capture.evidence(),
            "events": {"payload_sha256": events_sha, "receipt_bound": True},
        },
        "evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
    }

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--policy", type=Path, required=True)
    parser.add_argument("--policy-bundle", type=Path, required=True)
    parser.add_argument("--events", 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(
            arguments.policy, arguments.policy_bundle, arguments.events,
            arguments.receipt, arguments.receipt_bundle, evaluated_at,
        )
    except Exception as error:
        result = {
            "schema_version": "aidefend.delegated-token-coverage-result.v2",
            "control": "AID-D-011.004", "guidance": "G001",
            "status": "ERROR",
            "status_reason": "signature, schema, population, or evaluation failed",
            "error_class": type(error).__name__,
            "population_complete": False,
            "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 the exact coverage result

The signed collector receipt binds the exact event bytes, policy, source checkpoints, probe population, cohort window, and observation horizon. The evaluator keeps delegated unknown principals separate so G002 can still consume a complete population. A confirmed failed required-path probe is FAIL; incomplete checkpoints, stale evidence, or unfinished collection are INSUFFICIENT_DATA; a signed absent architecture is NOT_APPLICABLE; cryptographic, schema, or parser faults are ERROR.

#!/usr/bin/env bash
set -euo pipefail
snapshot="$(mktemp -d)"
publish_stage=""
cleanup() {
  rm -rf -- "$snapshot"
  if [[ -n "$publish_stage" && -d "$publish_stage" ]]; then
    rm -rf -- "$publish_stage"
  fi
}
trap cleanup EXIT
chmod 0700 "$snapshot"
set +e
python monitoring/token_coverage.py \
  --policy policy/token-coverage.json \
  --policy-bundle policy/token-coverage.sigstore.json \
  --events evidence/delegated-token-events.jsonl \
  --receipt evidence/token-coverage-receipt.json \
  --receipt-bundle evidence/token-coverage-receipt.sigstore.json \
  --output "$snapshot/coverage-result.json"
status=$?
set -e
result_key=/etc/aidefend/d011004/trust/coverage-result.pub
test -f "$result_key" && test ! -L "$result_key"
test "$(stat -c %u -- "$result_key")" -eq 0
key_mode="$(stat -c %a -- "$result_key")"
test "$((8#$key_mode & 8#022))" -eq 0
install -m 0400 -- "$result_key" "$snapshot/result.pub"
cosign sign-blob --yes --key env://D011004_COVERAGE_RESULT_SIGNING_KEY \
  --bundle "$snapshot/coverage-result.sigstore.json" \
  "$snapshot/coverage-result.json"
chmod 0400 "$snapshot/coverage-result.sigstore.json"
cosign verify-blob --key "$snapshot/result.pub" \
  --bundle "$snapshot/coverage-result.sigstore.json" \
  "$snapshot/coverage-result.json"
result_sha="$(sha256sum -- "$snapshot/coverage-result.json" | awk '{print $1}')"
test -d evidence && test ! -L evidence
result_base=evidence/token-coverage-results
test ! -L "$result_base"
if test -e "$result_base"; then
  test -d "$result_base"
else
  mkdir -m 0750 -- "$result_base"
fi
target="$result_base/$result_sha"
publish_stage="$(mktemp -d "$result_base/.$result_sha.tmp.XXXXXX")"
chmod 0700 "$publish_stage"
install -m 0440 -- "$snapshot/coverage-result.json" "$publish_stage/result.json"
install -m 0440 -- "$snapshot/coverage-result.sigstore.json" \
  "$publish_stage/result.sigstore.json"
install -m 0440 -- "$snapshot/result.pub" "$publish_stage/result.pub"
cosign verify-blob --key "$publish_stage/result.pub" \
  --bundle "$publish_stage/result.sigstore.json" \
  "$publish_stage/result.json"
python - "$publish_stage" "$target" <<'PY'
import sys
from pathlib import Path
from monitoring.token_coverage import publish_result_directory

publish_result_directory(Path(sys.argv[1]), Path(sys.argv[2]))
PY
publish_stage=""
printf '%s\n' "$target"
exit "$status"

Exercise every required path with synthetic canaries and retain source cursor bounds, probe IDs, rejected/dead-letter counts in the collector receipt, and exact payload/bundle/key digests. Publish the three signed artifacts plus their completion manifest from a same-parent private directory with one no-replace rename. The live evaluator takes evaluation time only from its trusted local UTC clock and binds it into the signed result. A separate identity may capture all four published files, verify their digest bindings and bundle, and independently recompute a historical replay from the signed receipt and result; replay output is audit evidence and cannot emit or replace a live PASS.

AID-D-011.004-G002

Correlate valid-token lifecycle and resource events to find unapproved client, audience, resource, scope, or workload combinations; use after revocation or expiry; lifecycle gaps; abnormal resource fan-out; and policy-defined bulk export.
Consume the exact complete G001 population

This detector imports the strict event parser and capture helpers from monitoring/token_coverage.py. It accepts only an aidefend.delegated-token-coverage-result.v2 G001 result from an aidefend.signed-result-publication.v1 directory, with the same coverage-policy digest, window, event digest and count, and population_complete=true. Delegated unknown principals remain analyzable; an actual G001 telemetry failure makes this detector INSUFFICIENT_DATA, not a false clean result.

from __future__ import annotations

# File: monitoring/token_behavior.py
import argparse
from collections import defaultdict, deque
from datetime import datetime, timedelta, timezone
import hashlib
import hmac
import os
from pathlib import Path
import tempfile

from token_coverage import (
    EXIT_CODES, SHA256, atomic_write, canonical, capture_signed, integer,
    read_stable, strict_json, strict_jsonl, text, timestamp, validate_event,
)

DETECTOR_POLICY_KEY = Path("/etc/aidefend/d011004/trust/detector-policy.pub")
COVERAGE_RESULT_KEY = Path("/etc/aidefend/d011004/trust/coverage-result.pub")
POLICY_FIELDS = {
    "schema_version", "policy_version", "applicable", "not_applicable_reason",
    "coverage_policy_sha256", "allowed_bindings", "rolling_window_seconds",
    "maximum_distinct_resources", "maximum_bytes_out",
    "required_lookback_seconds", "decision_horizon_seconds",
    "maximum_coverage_result_age_seconds", "maximum_future_skew_seconds",
}
BINDING_FIELDS = {
    "binding_id", "principal_id", "integration_id", "client_id", "audience",
    "resource_id", "workload_id", "scopes",
}
PUBLICATION_FIELDS = {
    "schema_version", "payload_sha256", "bundle_sha256",
    "verification_key_sha256",
}

def positive_integer(value, label):
    result = integer(value, label)
    if result < 1:
        raise ValueError(label + " must be positive")
    return result

def load_policy(captured):
    policy = strict_json(captured.raw, "detector policy")
    if (
        not isinstance(policy, dict) or set(policy) != POLICY_FIELDS
        or policy.get("schema_version") != "aidefend.token-behavior-policy.v2"
        or not isinstance(policy.get("applicable"), bool)
    ):
        raise ValueError("detector policy schema differs")
    text(policy["policy_version"], "policy_version")
    if not SHA256.fullmatch(str(policy["coverage_policy_sha256"])):
        raise ValueError("coverage policy digest is invalid")
    reason = policy["not_applicable_reason"]
    if policy["applicable"] and reason is not None:
        raise ValueError("applicable policy has a not-applicable reason")
    if not policy["applicable"] and (
        not isinstance(reason, str) or not reason.strip()
    ):
        raise ValueError("not-applicable policy lacks a reason")
    for field in (
        "rolling_window_seconds", "maximum_distinct_resources",
        "maximum_bytes_out", "required_lookback_seconds",
        "decision_horizon_seconds", "maximum_coverage_result_age_seconds",
    ):
        positive_integer(policy[field], field)
    integer(policy["maximum_future_skew_seconds"], "maximum_future_skew_seconds")
    if policy["required_lookback_seconds"] < policy["rolling_window_seconds"]:
        raise ValueError("required lookback is shorter than the rolling window")
    bindings = policy["allowed_bindings"]
    if not isinstance(bindings, list):
        raise ValueError("allowed_bindings must be an array")
    indexed = {}
    signatures = set()
    for number, row in enumerate(bindings, 1):
        if not isinstance(row, dict) or set(row) != BINDING_FIELDS:
            raise ValueError(f"allowed_binding:{number}: schema differs")
        for field in BINDING_FIELDS - {"scopes"}:
            text(row[field], f"allowed_binding:{number}:{field}")
        scopes = row["scopes"]
        if (
            not isinstance(scopes, list) or not scopes
            or len(scopes) != len(set(scopes))
        ):
            raise ValueError(f"allowed_binding:{number}: scopes are invalid")
        for scope in scopes:
            text(scope, f"allowed_binding:{number}:scope")
        signature = (
            row["principal_id"], row["integration_id"], row["client_id"],
            row["audience"], row["resource_id"], row["workload_id"],
            tuple(sorted(scopes)),
        )
        if row["binding_id"] in indexed or signature in signatures:
            raise ValueError("duplicate binding ID or exact binding tuple")
        indexed[row["binding_id"]] = row
        signatures.add(signature)
    if policy["applicable"] and not indexed:
        raise ValueError("applicable detector has no allowed binding")
    return policy, signatures

def capture_published_coverage_result(coverage_path, coverage_bundle):
    payload = Path(os.path.abspath(coverage_path))
    bundle = Path(os.path.abspath(coverage_bundle))
    parent = payload.parent
    stored_key = parent / "result.pub"
    completion_path = parent / "complete.json"
    expected_paths = {
        payload, bundle, stored_key, completion_path,
    }
    if (
        payload.name != "result.json"
        or bundle != parent / "result.sigstore.json"
        or not SHA256.fullmatch(parent.name)
        or not parent.is_dir() or parent.is_symlink()
        or {path.name for path in parent.iterdir()}
        != {"result.json", "result.sigstore.json", "result.pub", "complete.json"}
        or any(path.is_symlink() or not path.is_file() for path in expected_paths)
    ):
        raise ValueError("coverage publication directory is not exact and complete")
    completion_raw = read_stable(completion_path)
    completion = strict_json(completion_raw, "coverage publication manifest")
    if (
        not isinstance(completion, dict) or set(completion) != PUBLICATION_FIELDS
        or completion.get("schema_version")
        != "aidefend.signed-result-publication.v1"
        or any(
            not SHA256.fullmatch(str(completion.get(field)))
            for field in (
                "payload_sha256", "bundle_sha256", "verification_key_sha256"
            )
        )
        or not hmac.compare_digest(completion["payload_sha256"], parent.name)
    ):
        raise ValueError("coverage publication manifest is invalid")
    stored_key_raw = read_stable(stored_key)
    captured = capture_signed(
        "coverage-result", payload, bundle, COVERAGE_RESULT_KEY,
        expected_payload_sha256=parent.name,
        expected_bundle_sha256=completion["bundle_sha256"],
        expected_key_sha256=completion["verification_key_sha256"],
    )
    if not hmac.compare_digest(stored_key_raw, captured.key):
        raise ValueError("published coverage key differs from the trusted key")
    evidence = captured.evidence()
    evidence["content_address"] = parent.name
    evidence["publication_manifest_sha256"] = hashlib.sha256(
        completion_raw
    ).hexdigest()
    return captured, evidence

def gated_result(policy, coverage, reason, status, evaluated_at):
    return {
        "schema_version": "aidefend.delegated-token-behavior-result.v2",
        "control": "AID-D-011.004", "guidance": "G002",
        "status": status, "status_reason": reason,
        "policy_version": policy["policy_version"],
        "coverage_status": coverage.get("status") if coverage else None,
        "population_complete": False,
        "evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
    }

def binding_signature(row):
    return (
        row["principal_id"], row["integration_id"], row["client_id"],
        row["audience"], row["resource_id"], row["workload_id"],
        tuple(sorted(row["scopes"])),
    )

def token_binding(row):
    result = {
        field: row[field]
        for field in (
            "principal_id", "integration_id", "token_family_id",
            "token_instance_id", "token_correlation_id", "client_id",
            "audience", "workload_id", "scopes",
        )
    }
    result["scopes"] = tuple(sorted(row["scopes"]))
    return result

def lifecycle_matches(state, row):
    return state["scopes"] == tuple(sorted(row["scopes"])) and all(
        state[field] == row[field]
        for field in (
            "principal_id", "integration_id", "token_family_id",
            "token_instance_id", "token_correlation_id", "client_id",
            "audience", "workload_id",
        )
    )

def analyze(
    policy, allowed_bindings, coverage, events, input_evidence, evaluated_at
):
    window_start = timestamp(coverage["window_start"], "window_start")
    window_end = timestamp(coverage["window_end"], "window_end")
    collection_start = timestamp(coverage["collection_start"], "collection_start")
    observation_through = timestamp(
        coverage["observation_through"], "observation_through"
    )
    if not collection_start <= window_start < window_end <= observation_through:
        raise ValueError("signed coverage window bounds are not ordered")
    rolling_seed_start = window_start - timedelta(
        seconds=policy["rolling_window_seconds"]
    )
    if collection_start > window_start - timedelta(
        seconds=policy["required_lookback_seconds"]
    ):
        return gated_result(
            policy, coverage, "signed collection lacks required lifecycle lookback",
            "INSUFFICIENT_DATA", evaluated_at,
        )
    if observation_through < window_end + timedelta(
        seconds=policy["decision_horizon_seconds"]
    ):
        return gated_result(
            policy, coverage, "signed observation horizon is right-censored",
            "INSUFFICIENT_DATA", evaluated_at,
        )

    ordered = sorted(events, key=lambda row: (
        timestamp(row["occurred_at"], "occurred_at"),
        row["source_id"], row["source_sequence"], row["event_id"],
    ))
    same_time = defaultdict(set)
    for row in ordered:
        key = (row["token_instance_id"], row["occurred_at"])
        same_time[key].add(row["event_type"])
    if any(len(types) > 1 for types in same_time.values()):
        raise ValueError("ambiguous equal-time lifecycle transition")

    states = {}
    findings = []
    resource_windows = defaultdict(deque)
    finding_keys = set()
    def finding(row, reason, details=None):
        key = (row["event_id"], reason)
        if key not in finding_keys:
            finding_keys.add(key)
            item = {"event_id": row["event_id"], "reason": reason}
            if details is not None:
                item["details"] = details
            findings.append(item)

    for row in ordered:
        occurred = timestamp(row["occurred_at"], "occurred_at")
        event_type = row["event_type"]
        token_id = row["token_instance_id"]
        if event_type == "TOKEN_ISSUED":
            if token_id in states:
                finding(row, "duplicate_token_issue")
                continue
            states[token_id] = {
                **token_binding(row),
                "issued_at": occurred,
                "expires_at": timestamp(row["expires_at"], "expires_at"),
                "revoked_at": None,
                "parent_token_instance_id": None,
            }
            continue
        if event_type == "TOKEN_REFRESH":
            parent_id = row["parent_token_instance_id"]
            parent = states.get(parent_id)
            if parent is None:
                finding(row, "refresh_without_parent_lifecycle")
                continue
            if token_id in states:
                finding(row, "duplicate_refreshed_token")
                continue
            if any(
                parent[field] != row[field]
                for field in (
                    "principal_id", "integration_id", "token_family_id",
                    "client_id", "audience", "workload_id",
                )
            ) or parent["scopes"] != tuple(sorted(row["scopes"])):
                finding(row, "refresh_binding_mismatch")
            if parent["revoked_at"] is not None and occurred >= parent["revoked_at"]:
                finding(row, "refresh_after_parent_revocation")
            if occurred >= parent["expires_at"]:
                finding(row, "refresh_after_parent_expiry")
            states[token_id] = {
                **token_binding(row),
                "issued_at": occurred,
                "expires_at": timestamp(row["expires_at"], "expires_at"),
                "revoked_at": None,
                "parent_token_instance_id": parent_id,
            }
            continue
        state = states.get(token_id)
        if state is None:
            finding(
                row,
                "revocation_without_issue" if event_type == "TOKEN_REVOKED"
                else "resource_use_without_lifecycle",
            )
            continue
        if not lifecycle_matches(state, row):
            finding(row, "token_lifecycle_binding_mismatch")
        if event_type == "TOKEN_REVOKED":
            if state["revoked_at"] is not None:
                finding(row, "duplicate_revocation")
            else:
                state["revoked_at"] = occurred
            continue

        if state["revoked_at"] is not None and occurred >= state["revoked_at"]:
            finding(row, "resource_use_after_revocation")
        if occurred >= state["expires_at"]:
            finding(row, "resource_use_after_expiry")
        if binding_signature(row) not in allowed_bindings:
            finding(row, "unapproved_exact_binding")
        if not rolling_seed_start <= occurred < window_end:
            continue
        identity = (row["principal_id"], row["integration_id"])
        window = resource_windows[identity]
        cutoff = occurred - timedelta(seconds=policy["rolling_window_seconds"])
        while window and window[0][0] < cutoff:
            window.popleft()
        window.append((occurred, row["resource_id"], row["bytes_out"]))
        if occurred < window_start:
            continue
        distinct = len({item[1] for item in window})
        total_bytes = sum(item[2] for item in window)
        if distinct > policy["maximum_distinct_resources"]:
            finding(row, "resource_fanout_exceeded", {"observed": distinct})
        if total_bytes > policy["maximum_bytes_out"]:
            finding(row, "bulk_export_exceeded", {"observed_bytes": total_bytes})

    return {
        "schema_version": "aidefend.delegated-token-behavior-result.v2",
        "control": "AID-D-011.004", "guidance": "G002",
        "status": "FAIL" if findings else "PASS",
        "status_reason": (
            "delegated-token behavior findings were confirmed" if findings
            else "complete delegated-token population has no finding"
        ),
        "policy_version": policy["policy_version"],
        "coverage_policy_sha256": coverage["coverage_policy_sha256"],
        "window_id": coverage["window_id"],
        "window_start": coverage["window_start"],
        "window_end": coverage["window_end"],
        "observation_through": coverage["observation_through"],
        "events_sha256": coverage["events_sha256"],
        "event_count": len(events),
        "token_instance_count": len(states),
        "finding_count": len(findings),
        "findings": findings,
        "population_complete": True,
        "input_evidence": input_evidence,
        "evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
    }

def evaluate(policy_path, policy_bundle, events_path, coverage_path,
             coverage_bundle, evaluated_at):
    policy_capture = capture_signed(
        "detector-policy", policy_path, policy_bundle, DETECTOR_POLICY_KEY
    )
    policy, allowed_bindings = load_policy(policy_capture)
    policy_evidence = policy_capture.evidence()
    if not policy["applicable"]:
        result = gated_result(
            policy, None, policy["not_applicable_reason"], "NOT_APPLICABLE",
            evaluated_at,
        )
        result["input_evidence"] = {"policy": policy_evidence}
        return result

    coverage_capture, coverage_evidence = capture_published_coverage_result(
        coverage_path, coverage_bundle
    )
    coverage = strict_json(coverage_capture.raw, "coverage result")
    gate_evidence = {
        "policy": policy_evidence,
        "coverage_result": coverage_evidence,
    }
    def gate(reason, status):
        result = gated_result(policy, coverage, reason, status, evaluated_at)
        result["input_evidence"] = gate_evidence
        return result
    if (
        not isinstance(coverage, dict)
        or coverage.get("schema_version")
        != "aidefend.delegated-token-coverage-result.v2"
        or coverage.get("control") != "AID-D-011.004"
        or coverage.get("guidance") != "G001"
        or coverage.get("status")
        not in {"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"}
        or not isinstance(coverage.get("population_complete"), bool)
    ):
        raise ValueError("signed G001 coverage result is invalid")
    coverage_evaluated_at = timestamp(
        coverage.get("evaluated_at"), "coverage evaluated_at"
    )
    future_skew = timedelta(seconds=policy["maximum_future_skew_seconds"])
    if coverage_evaluated_at > evaluated_at + future_skew:
        raise ValueError("signed coverage evaluation is future-dated")
    if coverage["status"] == "ERROR":
        return gate("coverage evaluation failed", "ERROR")
    if coverage["status"] == "NOT_APPLICABLE":
        return gate(
            "coverage adapter is not applicable to this scope", "INSUFFICIENT_DATA"
        )
    required_coverage = {
        "coverage_policy_sha256", "window_id", "collection_start",
        "window_start", "window_end", "observation_through",
        "events_sha256", "event_count", "evaluated_at",
    }
    if not required_coverage <= set(coverage):
        raise ValueError("measured coverage result lacks a population binding")
    if coverage["status"] == "PASS" and coverage["population_complete"] is not True:
        raise ValueError("PASS coverage result is not population-complete")
    if coverage["coverage_policy_sha256"] != policy["coverage_policy_sha256"]:
        raise ValueError("detector and coverage policy digests differ")
    if not SHA256.fullmatch(str(coverage["events_sha256"])):
        raise ValueError("coverage event digest is invalid")
    observation_through = timestamp(
        coverage["observation_through"], "coverage observation_through"
    )
    if observation_through > evaluated_at + future_skew:
        raise ValueError("signed coverage observation is future-dated")
    maximum_age = timedelta(
        seconds=policy["maximum_coverage_result_age_seconds"]
    )
    if (
        evaluated_at - coverage_evaluated_at > maximum_age
        or evaluated_at - observation_through > maximum_age
    ):
        return gate("signed coverage result is stale", "INSUFFICIENT_DATA")
    if coverage["status"] in {"FAIL", "INSUFFICIENT_DATA"}:
        return gate(
            "required telemetry population is unavailable", "INSUFFICIENT_DATA"
        )

    events_raw = read_stable(events_path, allow_empty=True)
    events_sha = hashlib.sha256(events_raw).hexdigest()
    if events_sha != coverage["events_sha256"]:
        raise ValueError("events differ from signed G001 coverage bytes")
    events = [
        validate_event(row, f"event:{number}")
        for number, row in enumerate(strict_jsonl(events_raw, "events"), 1)
    ]
    if integer(coverage["event_count"], "coverage event_count") != len(events):
        raise ValueError("event count differs from signed G001 result")
    if len({row["event_id"] for row in events}) != len(events):
        raise ValueError("duplicate event_id")
    if len({(row["source_id"], row["source_sequence"]) for row in events}) != len(events):
        raise ValueError("duplicate source sequence")
    return analyze(
        policy, allowed_bindings, coverage, events,
        {
            "policy": policy_evidence,
            "coverage_result": coverage_evidence,
            "events": {"payload_sha256": events_sha, "coverage_bound": True},
        },
        evaluated_at,
    )

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--policy", type=Path, required=True)
    parser.add_argument("--policy-bundle", type=Path, required=True)
    parser.add_argument("--events", type=Path, required=True)
    parser.add_argument("--coverage-result", type=Path, required=True)
    parser.add_argument("--coverage-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(
            arguments.policy, arguments.policy_bundle, arguments.events,
            arguments.coverage_result, arguments.coverage_bundle, evaluated_at,
        )
    except Exception as error:
        result = {
            "schema_version": "aidefend.delegated-token-behavior-result.v2",
            "control": "AID-D-011.004", "guidance": "G002",
            "status": "ERROR",
            "status_reason": "signature, schema, lifecycle, or evaluation failed",
            "error_class": type(error).__name__,
            "population_complete": False,
            "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())
Publish and replay a create-once result

The signed detector policy lists exact allowed tuples, not independent allowlists. The state machine requires issue before refresh or revocation, binds refreshed and used tokens to their family, client, audience, workload and scopes, rejects ambiguous equal-time transitions, detects use after expiry/revocation, and evaluates policy-defined rolling fan-out and byte volume. The signed coverage horizon must include both required lookback and post-window decision time so right-censored cohorts cannot pass.

set -euo pipefail
: "\${COVERAGE_RESULT_DIR:?set COVERAGE_RESULT_DIR to the immutable G001 result directory}"
[[ "$COVERAGE_RESULT_DIR" =~ ^evidence/token-coverage-results/[0-9a-f]{64}$ ]] || {
  echo "COVERAGE_RESULT_DIR must identify a 64-hex content-addressed G001 result" >&2
  exit 12
}
test -d evidence && test ! -L evidence
test -d evidence/token-coverage-results
test ! -L evidence/token-coverage-results
test -d "$COVERAGE_RESULT_DIR" && test ! -L "$COVERAGE_RESULT_DIR"
for source in "$COVERAGE_RESULT_DIR/result.json" \
              "$COVERAGE_RESULT_DIR/result.sigstore.json" \
              "$COVERAGE_RESULT_DIR/result.pub" \
              "$COVERAGE_RESULT_DIR/complete.json"; do
  test -f "$source" && test ! -L "$source"
done
coverage_sha="\${COVERAGE_RESULT_DIR##*/}"
test "$coverage_sha" = "$(sha256sum -- "$COVERAGE_RESULT_DIR/result.json" | awk '{print $1}')"
snapshot="$(mktemp -d)"
publish_stage=""
cleanup() {
  rm -rf -- "$snapshot"
  if [[ -n "$publish_stage" && -d "$publish_stage" ]]; then
    rm -rf -- "$publish_stage"
  fi
}
trap cleanup EXIT
chmod 0700 "$snapshot"
set +e
python monitoring/token_behavior.py \
  --policy policy/token-behavior.json \
  --policy-bundle policy/token-behavior.sigstore.json \
  --events evidence/delegated-token-events.jsonl \
  --coverage-result "$COVERAGE_RESULT_DIR/result.json" \
  --coverage-bundle "$COVERAGE_RESULT_DIR/result.sigstore.json" \
  --output "$snapshot/behavior-result.json"
status=$?
set -e
result_key=/etc/aidefend/d011004/trust/behavior-result.pub
test -f "$result_key" && test ! -L "$result_key"
test "$(stat -c %u -- "$result_key")" -eq 0
key_mode="$(stat -c %a -- "$result_key")"
test "$((8#$key_mode & 8#022))" -eq 0
install -m 0400 -- "$result_key" "$snapshot/result.pub"
cosign sign-blob --yes --key env://D011004_BEHAVIOR_RESULT_SIGNING_KEY \
  --bundle "$snapshot/behavior-result.sigstore.json" \
  "$snapshot/behavior-result.json"
chmod 0400 "$snapshot/behavior-result.sigstore.json"
cosign verify-blob --key "$snapshot/result.pub" \
  --bundle "$snapshot/behavior-result.sigstore.json" \
  "$snapshot/behavior-result.json"
result_sha="$(sha256sum -- "$snapshot/behavior-result.json" | awk '{print $1}')"
test -d evidence && test ! -L evidence
result_base=evidence/token-behavior-results
test ! -L "$result_base"
if test -e "$result_base"; then
  test -d "$result_base"
else
  mkdir -m 0750 -- "$result_base"
fi
target="$result_base/$result_sha"
publish_stage="$(mktemp -d "$result_base/.$result_sha.tmp.XXXXXX")"
chmod 0700 "$publish_stage"
install -m 0440 -- "$snapshot/behavior-result.json" "$publish_stage/result.json"
install -m 0440 -- "$snapshot/behavior-result.sigstore.json" \
  "$publish_stage/result.sigstore.json"
install -m 0440 -- "$snapshot/result.pub" "$publish_stage/result.pub"
cosign verify-blob --key "$publish_stage/result.pub" \
  --bundle "$publish_stage/result.sigstore.json" \
  "$publish_stage/result.json"
python - "$publish_stage" "$target" <<'PY'
import sys
from pathlib import Path
from monitoring.token_coverage import publish_result_directory

publish_result_directory(Path(sys.argv[1]), Path(sys.argv[2]))
PY
publish_stage=""
printf '%s\n' "$target"
exit "$status"

The publisher commits a same-parent private staging directory with a completion manifest in one no-replace rename, so readers never accept partial result, bundle, or key sets. The live detector takes evaluation time only from its trusted local UTC clock and binds it into the signed result. This local create-once directory is not WORM; a separate retention identity must copy it to append-only storage, read back all four files, verify their digests and bundle, and independently recompute historical findings from the signed G001 result and exact event bytes. Historical replay is audit evidence and cannot emit or replace a live PASS. Test unauthorized combinations that independent allowlists would incorrectly accept, refresh without a parent, revoke without issue, duplicate issue/revoke, binding drift, expiry and revocation use, equal-time transitions, stale or future-dated G001 results, lookback/horizon truncation, fan-out and byte boundaries, and every five-state handoff.

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.004: Non-Human Identity & Delegated Token Abuse Detection | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service