AID-D-011.002: Authenticated Inter-Agent Security Event Contract Monitoring

Version 1.20261015 · Actionable sub-technique

Consume authenticated security events from every applicable A2A, MCP-adjacent, or custom inter-agent interaction and validate a versioned event contract covering event and trace identity, sender and receiver identity, credential or key identity, protocol and schema version, message or action digest, discovery registry and route, delegated scope, decision and canonical reason, policy version, and observation time. Reconcile discovery, route selection, message-envelope validation, delegation, peer-review, quorum, denial, and execution-receipt events against their authoritative request ledgers; emit findings for missing, forged, replayed, misbound, contradictory, or out-of-policy interaction evidence.

Pillars: app

Lifecycle: operation

Scope boundary

Owns validation and complete-population reconciliation of authenticated inter-agent security-event contracts across discovery, routing, messaging, delegation, peer review, quorum, denial, and execution receipts. It does not authorize, block, establish factual truth, or judge goal alignment.

AID-D-009.002: AID-D-011.002 verifies producer, transport, route, delegation, sequence, and event-contract integrity; AID-D-009.002 reconstructs propagation of a claim's content state across those interactions. Transport authenticity does not establish factual truth, and claim lineage cannot replace event authenticity.

AID-D-010: AID-D-011.002 verifies that goal and action events are authentic, correctly bound, and contract-complete; AID-D-010 evaluates whether their semantic content remains aligned with the approved goal. A valid signed event may describe a goal-divergent action.

AID-D-005.006: AID-D-011.002 authenticates discovery and routing event contracts and reconciles their complete population against authoritative interaction ledgers; AID-D-005.006 detects registry poisoning indicators, namespace churn, reconnaissance, and other anomalies within available registry and resolution telemetry. Missing, forged, replayed, or misbound events are scored here, while anomalous patterns in valid events are scored under AID-D-005.006.

Implementation guidance

AID-D-011.002-G001

Define and validate one canonical authenticated security-event contract across inter-agent discovery, routing, messaging, delegation, approval, denial, and execution receipts.
Canonical event contract

Require a versioned envelope with event_id, trace_id, interaction_id, event type, producer identity, sender and receiver identity, credential or key ID, protocol and schema version, registry and route version, message or action SHA-256, delegated scope, decision, canonical reason code, policy version, sequence, and signed observation time. The collector validates producer authentication, signature, schema, digest shape, identity binding, freshness, and monotonic sequence before accepting the event. Raw credentials and message bodies stay outside ordinary telemetry.

Failure semantics

Unknown event type, unsupported schema, invalid signature, missing required field, stale clock evidence, or unavailable key material is INSUFFICIENT_DATA or ERROR, not a valid denial and never a clean interaction. The monitor publishes a contract finding and does not retry, drop, authorize, or execute the interaction.

Independent verification

Replay valid, malformed, unknown-producer, wrong receiver, digest mismatch, duplicate, replay, sequence gap, clock-skew, schema-downgrade, and key-rotation fixtures against the same signed contract and compare canonical event digests and outcomes.

Executable authenticated event validation
# File: a2a_events/validate_event.py
from __future__ import annotations

import base64
import hashlib
import json
import math
import re
from datetime import datetime, timezone

from jsonschema import Draft202012Validator
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


SHA256 = re.compile(r"[0-9a-f]{64}")
REQUIRED_PAYLOAD = {
    "event_id", "trace_id", "interaction_id", "event_type", "producer_id",
    "source_agent_id", "destination_agent_id", "key_id", "protocol_version",
    "schema_version", "registry_version", "route_version", "action_sha256",
    "delegated_scope", "decision", "reason_code", "policy_version",
    "sequence", "observed_at",
}
VALID_DECISIONS = {"APPROVE", "REJECT", "ABSTAIN", "NONE"}


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


def validate_event(
    envelope: dict,
    verified_schema: dict,
    active_keys: dict[tuple[str, str], Ed25519PublicKey],
    producer_high_water: dict[str, int],
    *,
    evaluated_at: datetime,
    maximum_age_seconds: float,
    maximum_future_skew_seconds: float,
) -> dict:
    if set(envelope) != {"payload", "signature_b64"}:
        raise ValueError("event envelope differs")
    if (
        not math.isfinite(maximum_age_seconds) or maximum_age_seconds <= 0
        or not math.isfinite(maximum_future_skew_seconds)
        or maximum_future_skew_seconds < 0
    ):
        raise ValueError("event freshness policy is invalid")
    payload = envelope["payload"]
    Draft202012Validator(verified_schema).validate(payload)
    if not isinstance(payload, dict) or set(payload) != REQUIRED_PAYLOAD:
        raise ValueError("event semantic field set differs")
    producer = payload["producer_id"]
    key_id = payload["key_id"]
    active_key = active_keys.get((producer, key_id))
    if active_key is None:
        raise PermissionError("event producer/key pair is absent, unknown, or revoked")
    for name in REQUIRED_PAYLOAD - {"sequence", "delegated_scope"}:
        if not isinstance(payload[name], str) or not payload[name]:
            raise ValueError(f"event field is invalid: {name}")
    if not SHA256.fullmatch(payload["action_sha256"]):
        raise ValueError("event action digest is invalid")
    if payload["decision"] not in VALID_DECISIONS:
        raise ValueError("event decision is invalid")
    sequence = payload["sequence"]
    if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence < 0:
        raise ValueError("event sequence is invalid")
    previous = producer_high_water.get(producer)
    if previous is not None and sequence != previous + 1:
        raise ValueError("event sequence is replayed or has a gap")
    observed = _timestamp(payload["observed_at"])
    age = (evaluated_at.astimezone(timezone.utc) - observed).total_seconds()
    if age > maximum_age_seconds or age < -maximum_future_skew_seconds:
        raise ValueError("event timestamp is stale or future-dated")
    canonical = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode()
    active_key.verify(
        base64.b64decode(envelope["signature_b64"], validate=True), canonical
    )
    producer_high_water[producer] = sequence
    return {
        "status": "VALID",
        "event": payload,
        "event_sha256": hashlib.sha256(canonical).hexdigest(),
    }

The schema must require event ID, producer, sequence, interaction ID, source and destination agent IDs, event type, decision, policy version, timestamps, and payload digest. Reject extra fields unless the signed schema explicitly defines them.

AID-D-011.002-G002

Reconcile the complete expected inter-agent security-event population with authoritative interaction ledgers and monitor producer, sequence, and delivery health.
Population reconciliation

Derive the expected population from independently read discovery requests, selected routes, accepted message-envelope IDs, delegation requests, critical-action proposals, and execution receipts. Join only on stable interaction and trace IDs. For each expected event type, compare expected, produced, accepted, duplicate, late, rejected, and missing counts by producer and contract version; a missing event cannot be inferred from the absence of a finding.

Approval telemetry is evidence, not enforcement

For policy-declared approval flows, preserve APPROVE, REJECT, ABSTAIN, MISSING, and ERROR, independently recompute the recorded quorum, and bind it to the exact action digest and execution receipt. Emit disagreement, bypass, substitution, or executed-without-quorum findings without making the authorization decision here.

Independent verification

Use a separate read-only identity to reconstruct the complete population and replay missing producer, partial outage, duplicate, late, restart, registry rollover, approval bypass, substituted action, and execution-without-proposal cases.

Executable population reconciliation
# File: a2a_events/reconcile.py
from __future__ import annotations


CELL_FIELDS = (
    "event_id", "producer_id", "event_type", "trace_id", "interaction_id",
    "source_agent_id", "destination_agent_id", "protocol_version",
    "schema_version", "registry_version", "route_version", "action_sha256",
    "decision", "reason_code", "sequence",
)


def _cells(rows: list[dict], label: str) -> dict[str, tuple]:
    result = {}
    for row in rows:
        if not isinstance(row, dict) or any(field not in row for field in CELL_FIELDS):
            raise ValueError(f"{label} event cell is incomplete")
        event_id = row["event_id"]
        if not isinstance(event_id, str) or not event_id or event_id in result:
            raise ValueError(f"{label} event ID is absent or duplicated")
        result[event_id] = tuple(row[field] for field in CELL_FIELDS)
    return result


def reconcile(expected: list[dict], observed: list[dict]) -> dict:
    expected_cells = _cells(expected, "expected")
    observed_cells = _cells(observed, "observed")
    missing = sorted(set(expected_cells) - set(observed_cells))
    extra = sorted(set(observed_cells) - set(expected_cells))
    substituted = sorted(
        event_id
        for event_id in set(expected_cells) & set(observed_cells)
        if observed_cells[event_id] != expected_cells[event_id]
    )
    by_producer: dict[str, list[int]] = {}
    for row in observed:
        sequence = row["sequence"]
        if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence < 0:
            raise ValueError("observed sequence is invalid")
        by_producer.setdefault(row["producer_id"], []).append(sequence)
    sequence_gaps = {
        producer: sorted(values)
        for producer, values in by_producer.items()
        if len(values) != len(set(values))
        or sorted(values) != list(range(min(values), max(values) + 1))
    }
    return {
        "expected_count": len(expected_cells),
        "observed_count": len(observed_cells),
        "missing": missing,
        "extra": extra,
        "substituted": substituted,
        "sequence_gap_producers": sorted(sequence_gaps),
        "finding": bool(missing or extra or substituted or sequence_gaps),
    }

Build expected from the authoritative interaction and delegation ledgers, not from the telemetry stream being tested. Reconcile each bounded window and retain producer health and delivery-lag evidence.

AID-D-011.002-G003

Correlate contract-valid discovery, route, message, delegation, and approval events for registry poisoning, identity or capability forgery, replay, and confused-deputy patterns.
Contract-bound correlation

Correlate only validated events on declared source, target, credential or key, route, registry version, delegation, task, action digest, and interaction ID dimensions. Preserve the canonical reason code and contract version. Emit single-event findings for invalid binding or replay and policy-calibrated window findings for repeated denials, route churn, capability inflation, verifier disagreement, or bypass attempts. Never reinterpret a missing or invalid event as success.

Independent verification

Revalidate referenced Agent Card, registry, route, key, message digest, delegation, policy, and execution evidence; then recompute correlations from the immutable event population. Replay forged card, revoked key, sender or receiver mismatch, capability or scope inflation, wrong route, replay, missing event, duplicate, late event, planned maintenance, and benign failover fixtures.

Executable A2A correlation rules
# File: a2a_events/correlate.py
from __future__ import annotations

from datetime import datetime, timezone


EXACT_BINDINGS = (
    "action_sha256", "capability_sha256", "route_version",
    "registry_version", "delegation_sha256", "policy_version", "nonce",
)


def _time(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("approval time lacks timezone")
    return parsed.astimezone(timezone.utc)


def correlate(events: list[dict], *, evaluated_at: datetime) -> list[dict]:
    findings = []
    seen_event_ids = set()
    approvals: dict[tuple[str, str], dict] = {}
    for event in sorted(events, key=lambda row: (row["observed_at"], row["event_id"])):
        event_id = event["event_id"]
        if event_id in seen_event_ids:
            findings.append({"event_id": event_id, "reason": "replay"})
            continue
        seen_event_ids.add(event_id)
        if (
            event.get("validation_status") != "VALID"
            or event.get("population_status") != "COMPLETE"
        ):
            findings.append({"event_id": event_id, "reason": "unvalidated_or_incomplete_event"})
            continue
        key = (event["interaction_id"], event["action_sha256"])
        if event["event_type"] == "approval":
            authorized = (
                event.get("decision") == "APPROVE"
                and event.get("signature_status") == "VALID"
                and event.get("authority_status") == "CURRENT"
                and event.get("quorum_status") == "SATISFIED"
                and _time(event["valid_until"]) >= evaluated_at.astimezone(timezone.utc)
            )
            if not authorized:
                findings.append({"event_id": event_id, "reason": "approval_not_authorized"})
                continue
            if key in approvals:
                findings.append({"event_id": event_id, "reason": "conflicting_approval"})
                continue
            approvals[key] = event
        elif event["event_type"] == "execution":
            approval = approvals.get(key)
            misbound = (
                approval is None
                or approval["destination_agent_id"] != event["source_agent_id"]
                or any(approval.get(name) != event.get(name) for name in EXACT_BINDINGS)
                or event.get("execution_receipt_sha256") is None
            )
            if misbound:
                findings.append({
                    "event_id": event_id,
                    "reason": "missing_or_misbound_approval",
                })
    return findings

Extend the signed rule pack with registry-revision, route, identity, capability, nonce, expiry, and delegation-chain bindings. The detector emits findings and coverage gaps; it does not authorize or terminate the interaction.

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.002: Authenticated Inter-Agent Security Event Contract Monitoring | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service