Scope boundary

Owns unknown or newly emergent agent identity and communication-edge discovery against a versioned approved graph. It does not score reputation or classify drift within identities already registered.

AID-D-011.003: AID-D-016.001 discovers identities and edges absent from the approved graph; AID-D-011.003 detects replica, namespace, workload, and image-lineage drift for identities already registered. A new identity is a rogue-agent discovery candidate rather than an ordinary registered-population variance.

AID-D-016.002: AID-D-016.001 establishes that an identity or edge is new or unknown; AID-D-016.002 combines that and other security signals into reputation and escalation evidence. Discovery is one input to scoring, not the containment recommendation itself.

AID-M-001.006: AID-D-016.001 discovers active or observed agent identities and communication edges from runtime graph behavior; AID-M-001.006 discovers organization-owned internal AI endpoints from private-network and control-plane scope. A reachable service endpoint is not necessarily an active agent node, and agent behavior can reveal identities that endpoint enumeration misses.

Implementation guidance

• AID-D-016.001-G001: Maintain a versioned agent registry baseline and alert on unknown agent identities or new communication edges, using stable identity keys (SPIFFE IDs/service accounts) and time-windowed anomaly detection.

AID-D-016.001-G001

When and where to run this detector

Apply this detector where independently authenticated AI-agent workloads communicate through a service mesh, gateway, message bus, or agent protocol. The workload-identity/platform owner exports source-authenticated observations; the agent-governance owner signs the approved identity/edge baseline; detection engineering evaluates a fixed complete window at the telemetry fan-in point. Bind every result to the exact application or agent ecosystem asset/version, baseline, identity schemes, required collectors, and window. A signed architecture inventory proving no agent-to-agent communication path exists yields NOT_APPLICABLE.

Production graph-baseline evaluator

from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


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


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


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
AGENT_GRAPH_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D016001_AGENT_GRAPH_COMMAND_TIMEOUT_SECONDS")

# File: detection/evaluate_agent_graph.py

import hashlib
import json
import os
import subprocess
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
import hmac
import re
import tempfile


BASELINE = Path("policy/approved-agent-graph.json")
EVENTS = Path("artifacts/agent-graph/observations.jsonl")
RECEIPT = Path("artifacts/agent-graph/collection-receipt.json")
EVIDENCE = Path("artifacts/agent-graph/AID-D-016.001-evidence.json")


def digest(path: Path) -> str:
    captured = _VERIFIED_INPUTS.get(Path(path))
    payload = captured if captured is not None else Path(path).read_bytes()
    return hashlib.sha256(payload).hexdigest()


_VERIFIED_INPUTS: dict[Path, bytes] = {}


def _write_private_snapshot(path: Path, payload: bytes) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def _strict_json(raw: bytes | str) -> object:
    text = raw.decode("utf-8", errors="strict") if isinstance(raw, bytes) else raw

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

    return json.loads(
        text,
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def verify(path: Path, key_env: str) -> None:
    payload_path = Path(path)
    signature_path = Path(str(payload_path) + ".sig")
    payload_raw = payload_path.read_bytes()
    signature_raw = signature_path.read_bytes()
    key_reference = os.environ[key_env]
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
    )) or key_reference.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else Path(key_reference).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-d016001-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        signature_snapshot = root / "signature"
        _write_private_snapshot(payload_snapshot, payload_raw)
        _write_private_snapshot(signature_snapshot, signature_raw)
        captured = [(payload_snapshot, payload_raw), (signature_snapshot, signature_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _write_private_snapshot(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(signature_snapshot), str(payload_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=AGENT_GRAPH_COMMAND_TIMEOUT_SECONDS,
            )
            if any(
                not hmac.compare_digest(candidate.read_bytes(), expected)
                for candidate, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for candidate, _ in captured:
                os.chmod(candidate, 0o600)
    _VERIFIED_INPUTS[payload_path] = payload_raw


def _verified_text(path: Path) -> str:
    try:
        return _VERIFIED_INPUTS[Path(path)].decode("utf-8", errors="strict")
    except KeyError as error:
        raise RuntimeError("signed input was not captured and verified") from error


def timestamp(value: object) -> datetime:
    if not isinstance(value, str):
        raise ValueError("timestamp is absent")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def load_inputs() -> tuple[dict, list[dict], dict]:
    verify(BASELINE, "D016001_BASELINE_VERIFY_KEY")
    verify(EVENTS, "D016001_COLLECTOR_VERIFY_KEY")
    verify(RECEIPT, "D016001_COMPLETENESS_VERIFY_KEY")
    baseline = _strict_json(_VERIFIED_INPUTS[BASELINE])
    receipt = _strict_json(_VERIFIED_INPUTS[RECEIPT])
    baseline_fields = {
        "schema_version", "baseline_version", "asset_id", "asset_version",
        "directed", "identity_schemes", "identities", "edges",
        "required_sources", "window_seconds", "minimum_observation_events",
        "maximum_evidence_age_seconds", "issued_at", "expires_at",
    }
    if (
        not isinstance(baseline, dict)
        or set(baseline) != baseline_fields
        or baseline["schema_version"] != "aidefend.agent-graph-baseline.v1"
        or not isinstance(baseline["directed"], bool)
    ):
        raise ValueError("agent graph baseline schema differs")
    for field in ("baseline_version", "asset_id", "asset_version"):
        if not isinstance(baseline[field], str) or not baseline[field]:
            raise ValueError(f"baseline {field} is absent")
    schemes = baseline["identity_schemes"]
    identities = baseline["identities"]
    sources = baseline["required_sources"]
    if (
        not isinstance(schemes, list)
        or not schemes
        or any(not isinstance(value, str) or not value.endswith("://") for value in schemes)
        or len(schemes) != len(set(schemes))
        or not isinstance(identities, list)
        or not identities
        or any(
            not isinstance(value, str)
            or not value
            or not any(value.startswith(scheme) for scheme in schemes)
            for value in identities
        )
        or len(identities) != len(set(identities))
        or not isinstance(sources, list)
        or not sources
        or any(not isinstance(value, str) or not value for value in sources)
        or len(sources) != len(set(sources))
    ):
        raise ValueError("baseline identity or source registry is invalid")
    for field in (
        "window_seconds", "minimum_observation_events",
        "maximum_evidence_age_seconds",
    ):
        if (
            isinstance(baseline[field], bool)
            or not isinstance(baseline[field], int)
            or baseline[field] < 1
        ):
            raise ValueError(f"baseline {field} is invalid")
    now = datetime.now(timezone.utc)
    issued, expires = timestamp(baseline["issued_at"]), timestamp(baseline["expires_at"])
    if not issued <= now < expires:
        raise ValueError("agent graph baseline is not currently valid")
    identity_set = set(identities)
    edge_set = set()
    if not isinstance(baseline["edges"], list):
        raise ValueError("baseline edges are not a list")
    for edge in baseline["edges"]:
        if (
            not isinstance(edge, list)
            or len(edge) != 2
            or edge[0] not in identity_set
            or edge[1] not in identity_set
        ):
            raise ValueError("baseline edge is invalid")
        key = tuple(edge) if baseline["directed"] else tuple(sorted(edge))
        if key in edge_set:
            raise ValueError("baseline edge is duplicated")
        edge_set.add(key)

    receipt_fields = {
        "schema_version", "asset_id", "asset_version", "events_sha256",
        "window_start", "window_end", "expected_sources", "source_counts",
        "source_checkpoints", "event_count", "collection_complete",
    }
    if (
        not isinstance(receipt, dict)
        or set(receipt) != receipt_fields
        or receipt["schema_version"] != "aidefend.agent-graph-collection.v1"
        or receipt["asset_id"] != baseline["asset_id"]
        or receipt["asset_version"] != baseline["asset_version"]
        or receipt["events_sha256"] != digest(EVENTS)
        or receipt["expected_sources"] != sources
        or not isinstance(receipt["collection_complete"], bool)
        or isinstance(receipt["event_count"], bool)
        or not isinstance(receipt["event_count"], int)
        or receipt["event_count"] < 0
        or not isinstance(receipt["source_counts"], dict)
        or set(receipt["source_counts"]) != set(sources)
        or any(
            isinstance(value, bool) or not isinstance(value, int) or value < 0
            for value in receipt["source_counts"].values()
        )
        or not isinstance(receipt["source_checkpoints"], dict)
        or set(receipt["source_checkpoints"]) != set(sources)
    ):
        raise ValueError("agent graph receipt is invalid or misbound")
    start, end = timestamp(receipt["window_start"]), timestamp(receipt["window_end"])
    if int((end - start).total_seconds()) != baseline["window_seconds"]:
        raise ValueError("observation window differs from signed baseline")
    if (
        end > now
        or (now - end).total_seconds()
        > baseline["maximum_evidence_age_seconds"]
    ):
        raise ValueError("agent graph window is future-dated or stale")

    lines = _verified_text(EVENTS).splitlines()
    if any(not line.strip() for line in lines):
        raise ValueError("agent graph JSONL contains a blank record")
    events = [_strict_json(line) for line in lines]
    if len(events) != receipt["event_count"]:
        raise ValueError("agent graph event count differs")
    event_fields = {
        "schema_version", "event_id", "source", "source_sequence",
        "observed_at", "source_identity", "target_identity",
        "source_authn_evidence_sha256", "target_authn_evidence_sha256",
    }
    if any(
        not isinstance(item, dict)
        or set(item) != event_fields
        or item["schema_version"] != "aidefend.agent-graph-observation.v1"
        for item in events
    ):
        raise ValueError("agent graph event schema differs")
    event_ids = [item["event_id"] for item in events]
    if len(event_ids) != len(set(event_ids)) or any(not isinstance(value, str) or not value for value in event_ids):
        raise ValueError("agent graph event IDs are absent or duplicated")
    for item in events:
        if item["source"] not in sources or not start <= timestamp(item["observed_at"]) < end:
            raise ValueError("agent graph event source or time is outside receipt")
        if (
            isinstance(item["source_sequence"], bool)
            or not isinstance(item["source_sequence"], int)
            or item["source_sequence"] < 0
        ):
            raise ValueError("agent graph source sequence is invalid")
        for field in ("source_identity", "target_identity"):
            value = item[field]
            if (
                not isinstance(value, str)
                or not any(value.startswith(scheme) for scheme in schemes)
            ):
                raise ValueError("agent identity scheme is invalid")
        for field in ("source_authn_evidence_sha256", "target_authn_evidence_sha256"):
            value = item[field]
            if (
                not isinstance(value, str)
                or re.fullmatch(r"[0-9a-f]{64}", value) is None
            ):
                raise ValueError("authentication evidence digest is invalid")
    observed_counts = dict(sorted(Counter(item["source"] for item in events).items()))
    expected_counts = {
        source: receipt["source_counts"][source]
        for source in sources
        if receipt["source_counts"][source] > 0
    }
    if observed_counts != dict(sorted(expected_counts.items())):
        raise ValueError("per-source population differs")
    checkpoint_fields = {"first", "last", "complete"}
    for source in sources:
        checkpoint = receipt["source_checkpoints"][source]
        if (
            not isinstance(checkpoint, dict)
            or set(checkpoint) != checkpoint_fields
            or not isinstance(checkpoint["complete"], bool)
        ):
            raise ValueError("source checkpoint schema differs")
        source_sequences = sorted(
            item["source_sequence"] for item in events if item["source"] == source
        )
        if source_sequences:
            if (
                isinstance(checkpoint["first"], bool)
                or not isinstance(checkpoint["first"], int)
                or isinstance(checkpoint["last"], bool)
                or not isinstance(checkpoint["last"], int)
                or source_sequences
                != list(range(checkpoint["first"], checkpoint["last"] + 1))
            ):
                raise ValueError("source sequence population is incomplete")
        elif checkpoint["first"] is not None or checkpoint["last"] is not None:
            raise ValueError("empty source has non-empty checkpoint bounds")
    return baseline, events, receipt


def evaluate(baseline: dict, events: list[dict], receipt: dict) -> dict:
    if (
        receipt["collection_complete"] is not True
        or any(
            checkpoint["complete"] is not True
            for checkpoint in receipt["source_checkpoints"].values()
        )
    ):
        return {"status": "INSUFFICIENT_DATA", "reason": "required source collection is incomplete", "findings": []}
    if len(events) < baseline["minimum_observation_events"]:
        return {"status": "INSUFFICIENT_DATA", "reason": "observation population is below signed minimum", "findings": []}
    known_identities = set(baseline["identities"])
    known_edges = {
        tuple(edge) if baseline["directed"] else tuple(sorted(edge))
        for edge in baseline["edges"]
    }
    unknown_identities, new_edges, findings = set(), set(), []
    for item in events:
        source, target = item["source_identity"], item["target_identity"]
        for identity in (source, target):
            if identity not in known_identities:
                unknown_identities.add(identity)
        edge = (source, target) if baseline["directed"] else tuple(sorted((source, target)))
        if edge not in known_edges:
            new_edges.add(edge)
    if unknown_identities:
        findings.append("unknown_agent_identity")
    if new_edges:
        findings.append("new_agent_communication_edge")
    return {
        "status": "FAIL" if findings else "PASS",
        "reason": "agent graph baseline finding exists" if findings
        else "complete agent graph window matched the signed baseline",
        "findings": findings,
        "unknown_identities": sorted(unknown_identities),
        "new_edges": [list(edge) for edge in sorted(new_edges)],
    }


def main() -> None:
    try:
        baseline, events, receipt = load_inputs()
        result = evaluate(baseline, events, receipt)
        evidence = {
            "schema_version": "aidefend.d016001-evidence.v1",
            "control": "AID-D-016.001",
            "asset_id": baseline["asset_id"],
            "asset_version": baseline["asset_version"],
            "baseline_version": baseline["baseline_version"],
            "window_start": receipt["window_start"],
            "window_end": receipt["window_end"],
            "event_population": len(events),
            "baseline_sha256": digest(BASELINE),
            "events_sha256": digest(EVENTS),
            "receipt_sha256": digest(RECEIPT),
            **result,
            "evaluated_at": datetime.now(timezone.utc).isoformat(),
        }
    except Exception as error:
        evidence = {
            "schema_version": "aidefend.d016001-evidence.v1",
            "control": "AID-D-016.001",
            "status": "ERROR",
            "reason": f"{type(error).__name__}: {error}",
            "evaluated_at": datetime.now(timezone.utc).isoformat(),
        }
    EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
    EVIDENCE.write_text(json.dumps(evidence, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8")
    subprocess.run(
        [
            "cosign", "sign-blob", "--yes",
            "--key", "env://D016001_EVIDENCE_SIGNING_KEY",
            "--bundle", str(EVIDENCE) + ".sig", str(EVIDENCE),
        ],
        check=True, timeout=AGENT_GRAPH_COMMAND_TIMEOUT_SECONDS,
    )
    verify(EVIDENCE, "D016001_EVIDENCE_VERIFY_KEY")


if __name__ == "__main__":
    main()

Replay and interpret results

Action: At the authenticated agent-graph collector, compare every observed identity and communication edge in the reconciled window with the signed baseline, persist the signed delta result, and emit unknown-identity and new-edge findings to the agent-security queue; containment remains with AID-I-003.

Cross-run mutable-state detection

Include authenticated observations for applicable shared registries, mutable tags, manifests, notes, queues, and coordination metadata. Key them by tenant, principal, run, object generation, and operation, and flag a new communication or state edge when one run writes state another run can consume without an approved handoff. Do not flag a digest-addressed immutable cache merely because several runs read it. Authorization and write denial remain with AID-H-018.002 and service IAM; this guidance emits the finding and preserves the observed edge only.

Open-source tools

• Istio
• Kiali (service-graph visualization from Istio telemetry)
• Envoy
• SPIRE
• OpenTelemetry
• Prometheus
• Cosign

Commercial / hosted tools

• Datadog
• Splunk Enterprise Security
• Microsoft Sentinel

Threat mappings

MITRE ATLAS

• AML.T0103 Deploy AI Agent (baseline detects unauthorized agent deployments)
• AML.T0118 Autonomous AI Agent Communication
• AML.T0118.001 Autonomous AI Agent Communication: Direct Agent Communication (unapproved agent-to-agent communication edges surface against the baseline graph)
• AML.T0124 Autonomous Attack Orchestration (new orchestrator-to-executor edges surface against the approved agent graph)
• AML.T0125 Create Account (newly created agent identities surface against the baseline graph)

MAESTRO

• Compromised Agent Registry (L7)
• Lateral Movement (Cross-Layer)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI10:2026 Rogue Agents
• ASI07:2026 Insecure Inter-Agent Communication
• ASI03:2026 Identity and Privilege Abuse (agent graph detects identity anomalies)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-4.1.1 Rogue Agent Introduction
• AISubtech-4.2.2 Session Boundary Violation (unapproved cross-run mutable-state handoffs observed as new state edges)
• AISubtech-5.1.1 Long-term / Short-term Memory Injection (unapproved cross-run mutable-state handoffs observed as new state edges)
• AITech-4.1 Agent Injection
• AITech-4.2 Context Boundary Attacks (unapproved cross-run mutable-state handoffs observed as new state edges)
• AITech-5.1 Memory System Persistence (unapproved cross-run mutable-state handoffs observed as new state edges)

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions (graph baseline detects unauthorized new agent introductions)
• IIC: Insecure Integrated Component (new-agent discovery detects unauthorized component additions)

Databricks AI Security Framework 3.0

• Agents - Core 13.13: Rogue Agents in Multi-Agent Systems
• Agents - Core 13.9: Identity Spoofing & Impersonation