AID-H-004.003: Secure Agent-to-Agent Communication

Version 1.20261015 · Actionable sub-technique

Secure the live request path between agents in custom, MCP-adjacent, and standards-based Agent2Agent (A2A) deployments. Each receiver verifies the authenticated peer, protocol version, payload integrity, replay state, task/context ownership, valid task transition, stream order, push-callback authenticity, and per-request authorization before exposing task data or accepting work.

Pillars: app

Lifecycle: building, operation

Scope boundary

Owns receiver-side security for live agent-to-agent requests and streams, including peer authentication, protocol and payload integrity, replay state, task and context ownership, state transitions, stream order, callback authenticity, and per-request authorization.

AID-H-004.002: AID-H-004.003 enforces the A2A message and task lifecycle after peer identity is established; AID-H-004.002 authenticates the underlying workload, service, or API caller. Machine authentication does not validate message replay, task state, protocol binding, or delegation context.

AID-H-004.004: AID-H-004.003 secures the selected live peer communication path; AID-H-004.004 verifies discovery records, Agent Cards, endpoint selection, and advertised capability provenance before that path is chosen. Discovery trust precedes communication enforcement but does not replace it.

AID-H-017.006: AID-H-004.003 validates each receiver-side message and task transition; AID-H-017.006 constrains the end-to-end delegation graph, hop and loop bounds, and capability attenuation across multiple agents. Per-hop transport security and whole-chain delegation safety are separate controls.

AID-H-018.008: AID-H-004.003 rejects replayed or misbound agent messages; AID-H-018.008 prevents duplicate external side effects from retries, redelivery, resume, and uncertain outcomes. Message replay resistance does not guarantee that a provider effect executes at most once.

AID-H-004.005: AID-H-004.003 owns A2A peer, task and context ownership, state-transition, stream-order, callback, and per-request authorization semantics; AID-H-004.005 owns the generic end-to-end message envelope and replay primitive across applicable trust boundaries. The A2A control remains independently implementable and must not be reduced to a reference to the generic primitive.

Implementation guidance

AID-H-004.003-G001

Issue unique, cryptographically verifiable identities to each AI agent (e.g., using SPIFFE/SPIRE).
Concept:

Treat each autonomous agent like a microservice. It needs a strong, verifiable identity that doesn't depend on a shared secret or API key. The SPIFFE/SPIRE standard provides a platform-agnostic way to issue short-lived cryptographic identities (SVIDs, which are X.509 certificates) to workloads automatically.

Step 1: Configure SPIRE for Agent Attestation

A SPIRE agent running on the same host as the AI agent can 'attest' the AI agent's identity based on properties like its process ID, user ID, or a Kubernetes service account. The SPIRE server then issues a unique SPIFFE ID and a corresponding SVID (certificate).

# Example: Registering an agent entry in the SPIRE server
# This command tells the SPIRE server that any process running as user 'ai_agent_user'
# on a node with a specific AWS instance ID is allowed to have the identity 'spiffe://example.org/agent/payment'.

spire-server entry create \
    -spiffeID "spiffe://example.org/agent/payment" \
    -parentID "spiffe://example.org/node/aws/i-0123456789abcdef0" \
    -selector "unix:uid:1001"
Step 2: Fetch Identity in the Agent Code

The AI agent uses the SPIFFE Workload API (typically via a local Unix domain socket) to request its identity document (SVID) without needing any bootstrap credentials.

# Run as the attested workload UID; do not run this diagnostic as root.
spire-agent api fetch x509 \
  -socketPath /tmp/spire-agent/public/api.sock \
  -timeout 5s

# Staging-only readback when certificate files are needed for inspection.
workdir="$(mktemp -d)"
trap 'rm -rf "$workdir"' EXIT
spire-agent api fetch x509 -write "$workdir"
openssl x509 -in "$workdir/svid.0.pem" -noout -subject -issuer -dates
Verification and evidence

Run the fetch under the registered UID and require the expected SPIFFE ID and a currently valid SVID. Repeat under an unregistered UID and require PermissionDenied. Preserve the registration-entry digest, selector readback, SPIRE version, SPIFFE ID, certificate serial and validity window, and explicit PASS/FAIL/ERROR result. Production workloads should consume the streaming Workload API through a maintained SPIFFE library so SVID rotation is applied without writing private keys to disk.

Action: Deploy SPIRE to your AI agent infrastructure. Register each agent type with a unique SPIFFE ID based on verifiable selectors. Code your agents to use the Workload API to fetch their identities for use in mTLS communication.

Keep the action safely bounded

deny disclosure, transition, dispatch, or side effect on unverifiable identity, route, signature, replay state, sequence, ownership, authorization, or protocol contract.

AID-H-004.003-G002

Enforce A2A v1 task/context authorization, terminal-state immutability, specification-defined stream shape, and authenticated push delivery at the receiving boundary.
Before you begin

Apply AID-H-004.003-G002 only when one agent, agent gateway, protocol adapter, stream, or push receiver sends task data or requests to another agent.

Concept

Authorize before any lookup that could reveal task existence. Bind tasks and contexts to the authenticated principal, client, and tenant; accept only known A2A v1 states; reject changes after a terminal state; and validate the two allowed stream patterns. Agent Card verification belongs to H-004.004, message replay protection to the separate signed-message guidance below, webhook URL admission to AID-H-019.001, and duplicate side-effect safety to AID-H-018.008.

# File: a2a/task_and_stream_boundary.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, Mapping


TERMINAL = {
    "TASK_STATE_COMPLETED", "TASK_STATE_FAILED",
    "TASK_STATE_CANCELED", "TASK_STATE_REJECTED",
}
KNOWN_STATES = TERMINAL | {
    "TASK_STATE_SUBMITTED", "TASK_STATE_WORKING",
    "TASK_STATE_INPUT_REQUIRED", "TASK_STATE_AUTH_REQUIRED",
}
TASK_OPERATIONS = {
    "SendMessage", "SendStreamingMessage", "GetTask", "ListTasks",
    "CancelTask", "SubscribeToTask",
    "CreateTaskPushNotificationConfig", "GetTaskPushNotificationConfig",
    "ListTaskPushNotificationConfigs", "DeleteTaskPushNotificationConfig",
}


@dataclass(frozen=True)
class Principal:
    subject: str
    client_id: str
    tenant_id: str
    scopes: frozenset[str]


def authorize_operation(
    principal: Principal,
    operation: str,
    required_scopes: Mapping[str, str],
) -> None:
    if operation not in TASK_OPERATIONS:
        raise PermissionError("unsupported_a2a_operation")
    scope = required_scopes.get(operation)
    if not scope or scope not in principal.scopes:
        raise PermissionError("a2a_operation_not_authorized")


def load_owned_task(repository, principal: Principal, task_id: str):
    task = repository.get(
        tenant_id=principal.tenant_id,
        subject=principal.subject,
        client_id=principal.client_id,
        task_id=task_id,
    )
    if task is None:
        raise LookupError("task_not_found_or_not_accessible")
    return task


def require_state_change(current: str, proposed: str) -> None:
    if proposed not in KNOWN_STATES:
        raise ValueError("unknown_or_unspecified_task_state")
    if current in TERMINAL and proposed != current:
        raise PermissionError("terminal_task_is_immutable")


def exactly_one_variant(response: dict) -> tuple[str, dict]:
    variants = [key for key in ("task", "message", "statusUpdate", "artifactUpdate")
                if key in response]
    if len(variants) != 1 or not isinstance(response[variants[0]], dict):
        raise ValueError("invalid_stream_response_variant")
    return variants[0], response[variants[0]]


class StreamGuard:
    def __init__(self) -> None:
        self.mode: str | None = None
        self.task_id: str | None = None
        self.context_id: str | None = None
        self.state: str | None = None
        self.closed = False

    def accept(self, response: dict) -> None:
        if self.closed:
            raise ValueError("event_after_stream_completion")
        variant, value = exactly_one_variant(response)

        if self.mode is None:
            if variant == "message":
                self.mode, self.closed = "message", True
                return
            if variant != "task":
                raise ValueError("task_stream_must_start_with_task")
            self.mode = "task"
            self.task_id = value.get("id")
            self.context_id = value.get("contextId")
            self.state = value.get("status", {}).get("state")
            if not self.task_id or self.state not in KNOWN_STATES:
                raise ValueError("invalid_initial_task")
            self.closed = self.state in TERMINAL
            return

        if variant not in {"statusUpdate", "artifactUpdate"}:
            raise ValueError("invalid_task_stream_followup")
        if value.get("taskId") != self.task_id or value.get("contextId") != self.context_id:
            raise PermissionError("cross_task_or_context_stream_event")
        if variant == "statusUpdate":
            proposed = value.get("status", {}).get("state")
            require_state_change(self.state, proposed)
            self.state = proposed
            self.closed = proposed in TERMINAL


def verify_push_delivery(
    *,
    authorization_header: str,
    expected_config_id: str,
    body: dict,
    authenticate_push: Callable[[str, str], Principal],
    repository,
):
    principal = authenticate_push(authorization_header, expected_config_id)
    variant, value = exactly_one_variant(body)
    task_id = value.get("taskId") or value.get("id")
    if not isinstance(task_id, str) or not task_id:
        raise ValueError("push_missing_task_id")
    task = load_owned_task(repository, principal, task_id)
    if task.push_config_id != expected_config_id:
        raise PermissionError("push_configuration_binding_mismatch")
    return principal, task, variant

Verification and evidence: replay every task and push-config operation across owner, client, tenant, scope, task, and context boundaries. Test both valid stream patterns; a non-Task first event, Message plus another event, update for the wrong task/context, unknown state, terminal mutation, event after closure, invalid push credential, and wrong push configuration must fail before data or side effects. On transport interruption, re-read or re-subscribe to the Task; do not claim gap-free delivery from a guessed sequence field not defined by A2A. Preserve protocol version, authenticated peer, operation/scope decision, task/context digests, state history, stream close reason, push-config/credential digests, and trace ID.

Keep the action safely bounded

deny disclosure, transition, dispatch, or side effect on unverifiable identity, route, signature, replay state, sequence, ownership, authorization, or protocol contract.

AID-H-004.003-G003

Encrypt all agent-to-agent communication channels (prefer mTLS).
Before you begin

Apply AID-H-004.003-G003 only when one agent, agent gateway, protocol adapter, stream, or push receiver sends task data or requests to another agent.

Concept:

All network communication between agents must be encrypted to prevent eavesdropping. Mutual TLS (mTLS) is preferred because it provides both encryption and strong, mutual authentication.

Use gRPC with TLS Credentials

When using a framework like gRPC for agent communication, you can configure it to use the SVIDs obtained from SPIFFE to establish a secure mTLS channel.

// File: identity/agent_client.go
package identity

import (
    "context"

    "github.com/spiffe/go-spiffe/v2/spiffeid"
    "github.com/spiffe/go-spiffe/v2/spiffetls/tlsconfig"
    "github.com/spiffe/go-spiffe/v2/workloadapi"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
)

func DialAgent(ctx context.Context, address string) (*grpc.ClientConn, func(), error) {
    source, err := workloadapi.NewX509Source(ctx)
    if err != nil {
        return nil, nil, err
    }
    // Keep source alive for the connection lifetime so SVID rotations are observed.
    serverID := spiffeid.RequireFromString("spiffe://example.org/agent/peer")
    tlsConfig := tlsconfig.MTLSClientConfig(
        source, source, tlsconfig.AuthorizeID(serverID),
    )
    conn, err := grpc.NewClient(
        address, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)),
    )
    if err != nil {
        source.Close()
        return nil, nil, err
    }
    cleanup := func() {
        conn.Close()
        source.Close()
    }
    return conn, cleanup, nil
}

Use the reciprocal tlsconfig.MTLSServerConfig on the server and authorize an explicit set of client SPIFFE IDs or a reviewed trust-domain policy. Bind the X509Source lifecycle to the server or connection lifecycle so certificate rotation continues.

Verification and evidence

Replay an allowed peer, an unexpected SPIFFE ID, an expired SVID, an untrusted bundle, and certificate rotation during a long-lived connection. Preserve both peer IDs, trust-bundle digest, SVID serials, negotiated TLS version/cipher, authorization decision, rotation readback, and explicit PASS/FAIL/ERROR verdict.

Action: Enforce mTLS for every hop (agent—agent, agent—adapter, adapter—service). Prefer short-lived certs and automated rotation.

Keep the action safely bounded

deny disclosure, transition, dispatch, or side effect on unverifiable identity, route, signature, replay state, sequence, ownership, authorization, or protocol contract.

AID-H-004.003-G004

Digitally sign each canonical inter-agent envelope for end-to-end integrity, auditable attribution, and replay-resistant receiver verification.
Before you begin

Apply AID-H-004.003-G004 only when one agent, agent gateway, protocol adapter, stream, or push receiver sends task data or requests to another agent.

Sign one exact, canonical envelope at the authenticated sender boundary and verify it before deserialization into any agent task, tool, or memory path. The signature binds protocol and schema versions, sender, audience, task/context, unique message and nonce, validity window, key and policy versions, message type, and the exact payload digest. The receiver resolves the verification key from its trusted identity registry; it never trusts a key or sender identity carried only by the message.

# File: agent_comms/signed_envelope.py
from __future__ import annotations

import base64
import hashlib
from datetime import datetime, timedelta, timezone
from typing import Callable

import rfc8785
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)

SIGNED_FIELDS = {
    "schema_version", "protocol", "protocol_version", "message_type",
    "sender_id", "audience_id", "task_id", "context_id", "message_id",
    "nonce", "issued_at", "expires_at", "key_id", "policy_version",
    "payload_sha256",
}
WIRE_FIELDS = SIGNED_FIELDS | {"payload", "signature_b64"}


def canonical(value: dict) -> bytes:
    return rfc8785.dumps(value)


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


def issue(
    *,
    payload: dict,
    sender_id: str,
    audience_id: str,
    task_id: str,
    context_id: str,
    message_id: str,
    nonce: str,
    protocol: str,
    protocol_version: str,
    message_type: str,
    key_id: str,
    policy_version: str,
    private_key: Ed25519PrivateKey,
    validity_seconds: int,
    maximum_validity_seconds: int,
) -> dict:
    if (
        isinstance(maximum_validity_seconds, bool)
        or not isinstance(maximum_validity_seconds, int)
        or maximum_validity_seconds < 1
        or isinstance(validity_seconds, bool)
        or not isinstance(validity_seconds, int)
        or not 1 <= validity_seconds <= maximum_validity_seconds
    ):
        raise ValueError("validity_outside_signed_policy_ceiling")
    payload_bytes = canonical(payload)
    now = datetime.now(timezone.utc)
    signed = {
        "schema_version": "aidefend.agent-message.v2",
        "protocol": protocol,
        "protocol_version": protocol_version,
        "message_type": message_type,
        "sender_id": sender_id,
        "audience_id": audience_id,
        "task_id": task_id,
        "context_id": context_id,
        "message_id": message_id,
        "nonce": nonce,
        "issued_at": now.isoformat(),
        "expires_at": (now + timedelta(seconds=validity_seconds)).isoformat(),
        "key_id": key_id,
        "policy_version": policy_version,
        "payload_sha256": hashlib.sha256(payload_bytes).hexdigest(),
    }
    return {
        **signed,
        "payload": payload,
        "signature_b64": base64.b64encode(
            private_key.sign(canonical(signed))
        ).decode("ascii"),
    }


def verify(
    wire: dict,
    *,
    authenticated_peer_id: str,
    expected_audience_id: str,
    expected_protocol: str,
    accepted_protocol_versions: set[str],
    expected_message_type: str,
    expected_task_id: str,
    expected_context_id: str,
    expected_policy_version: str,
    resolve_key: Callable[[str, str, datetime], Ed25519PublicKey],
    replay_store,
    maximum_clock_skew_seconds: int,
    maximum_validity_seconds: int,
) -> dict:
    if not isinstance(wire, dict) or set(wire) != WIRE_FIELDS:
        raise PermissionError("message_schema_mismatch")
    signed = {key: wire[key] for key in SIGNED_FIELDS}
    if (
        wire["schema_version"] != "aidefend.agent-message.v2"
        or wire["sender_id"] != authenticated_peer_id
        or wire["audience_id"] != expected_audience_id
        or wire["protocol"] != expected_protocol
        or wire["protocol_version"] not in accepted_protocol_versions
        or wire["message_type"] != expected_message_type
        or wire["task_id"] != expected_task_id
        or wire["context_id"] != expected_context_id
        or wire["policy_version"] != expected_policy_version
    ):
        raise PermissionError("message_context_binding_failed")

    now = datetime.now(timezone.utc)
    issued = utc(wire["issued_at"])
    expires = utc(wire["expires_at"])
    if (
        isinstance(maximum_validity_seconds, bool)
        or not isinstance(maximum_validity_seconds, int)
        or maximum_validity_seconds < 1
        or expires - issued > timedelta(seconds=maximum_validity_seconds)
    ):
        raise PermissionError("message_validity_exceeds_policy")
    skew = timedelta(seconds=maximum_clock_skew_seconds)
    if expires <= issued or now < issued - skew or now >= expires + skew:
        raise PermissionError("message_outside_validity_window")
    payload_bytes = canonical(wire["payload"])
    if hashlib.sha256(payload_bytes).hexdigest() != wire["payload_sha256"]:
        raise PermissionError("payload_digest_mismatch")

    key = resolve_key(wire["sender_id"], wire["key_id"], issued)
    try:
        key.verify(base64.b64decode(wire["signature_b64"], validate=True), canonical(signed))
    except (InvalidSignature, ValueError) as exc:
        raise PermissionError("message_signature_invalid") from exc

    replay_keys = (
        f"agent-message:nonce:{wire['sender_id']}:{wire['audience_id']}:{wire['nonce']}",
        f"agent-message:id:{wire['sender_id']}:{wire['audience_id']}:{wire['message_id']}",
    )
    ttl = max(1, int((expires + skew - now).total_seconds()))
    reserve_script = """
    for _, key in ipairs(KEYS) do
      if redis.call('EXISTS', key) == 1 then return 0 end
    end
    for _, key in ipairs(KEYS) do
      redis.call('SET', key, ARGV[1], 'EX', ARGV[2])
    end
    return 1
    """
    reserved = replay_store.eval(
        reserve_script, len(replay_keys), *replay_keys,
        wire["payload_sha256"], ttl,
    )
    if int(reserved) != 1:
        raise PermissionError("message_replay_detected")
    return {
        "verified_metadata": signed,
        "payload": wire["payload"],
    }
Verification, outcomes, and evidence

Evidence: retain the canonical signed metadata and its digest, payload digest rather than sensitive payload where possible, signature/key ID, authenticated peer, receiver identity, protocol/schema/policy versions, both replay keys and their atomic reservation result, verifier build digest, trace ID, and explicit outcome. An independent verifier resolves the historical key again and reproduces the signature and replay decision over the complete message population.

AID-H-004.003-G005

Bind every signed inter-agent message to a unique message ID/nonce, audience, and policy-derived validity window, then reject replay atomically.
Concept

Payload signatures prove integrity but do not stop reuse. After signature, issuer, audience, and key-status verification, validate an aware UTC time window and atomically reserve the sender/message/nonce tuple in a shared replay store before dispatch. Do not use a process-local last_seen_sequence for a horizontally scaled receiver.

# File: a2a/replay_gate.py
from __future__ import annotations

import hashlib
from datetime import datetime, timedelta, timezone

from redis.asyncio import Redis


class ReplayRejected(PermissionError):
    pass


def parse_utc(value: str) -> datetime:
    if not isinstance(value, str):
        raise ReplayRejected("timestamp_not_string")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise ReplayRejected("timestamp_invalid") from exc
    if parsed.tzinfo is None:
        raise ReplayRejected("timestamp_missing_timezone")
    return parsed.astimezone(timezone.utc)


async def reserve_message_once(
    redis: Redis,
    *,
    verified_sender_id: str,
    expected_audience: str,
    envelope: dict,
    maximum_lifetime_seconds: int,
) -> None:
    if maximum_lifetime_seconds < 1:
        raise ValueError("invalid replay policy")
    if envelope.get("audience") != expected_audience:
        raise ReplayRejected("message_audience_mismatch")
    message_id = envelope.get("message_id")
    nonce = envelope.get("nonce")
    if not all(isinstance(value, str) and value for value in (message_id, nonce)):
        raise ReplayRejected("missing_message_id_or_nonce")

    now = datetime.now(timezone.utc)
    issued_at = parse_utc(envelope.get("issued_at"))
    expires_at = parse_utc(envelope.get("expires_at"))
    lifetime = int((expires_at - issued_at).total_seconds())
    if issued_at > now or expires_at <= now or not 1 <= lifetime <= maximum_lifetime_seconds:
        raise ReplayRejected("message_time_window_invalid")

    fingerprint = hashlib.sha256(
        (verified_sender_id + "|" + message_id + "|" + nonce).encode("utf-8")
    ).hexdigest()
    ttl = max(1, int((expires_at - now).total_seconds()))
    inserted = await redis.set("a2a-replay:" + fingerprint, "1", ex=ttl, nx=True)
    if not inserted:
        raise ReplayRejected("message_replayed")

Action and evidence: call this gate only after cryptographic identity/signature verification and before task mutation or side effects. Load the maximum lifetime from signed deployment policy. Test exact replay, cross-audience reuse, future/expired timestamps, excessive lifetime, nonce reuse with a different message ID, Redis failure, and concurrent duplicates across replicas. Fail closed on replay-store failure for privileged operations and preserve sender/key ID, audience, message/nonce digests, policy version, decision, and trace ID.

Before you begin

Apply AID-H-004.003-G005 only when one agent, agent gateway, protocol adapter, stream, or push receiver sends task data or requests to another agent.

AID-H-004.003-G006

Implement identity-preserving protocol translation (anti-scrubbing) for adapters/gateways.
Before you begin

Apply AID-H-004.003-G006 only when one agent, agent gateway, protocol adapter, stream, or push receiver sends task data or requests to another agent.

The adapter first verifies a signed origin-verification receipt produced from the authenticated upstream transport, then translates the already-verified payload and signs a transformation attestation. The downstream service authorizes the original sender and adapter jointly; it never accepts a caller-supplied original_sender or a signature over only the pre-translation body.

# File: adapter/verified_translation.py
from __future__ import annotations

import base64
import hashlib
from datetime import datetime, timedelta, timezone
from typing import Callable

import rfc8785
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)

ORIGIN_RECEIPT_FIELDS = {
    "schema_version", "outcome", "authenticated_sender_id", "audience_id",
    "origin_protocol", "origin_protocol_version", "origin_envelope_sha256",
    "payload_sha256", "message_id", "task_id", "context_id", "policy_version",
    "issued_at", "expires_at", "verifier_key_id", "signature_b64",
}
ATTESTATION_FIELDS = {
    "schema_version", "original_sender_id", "adapter_id",
    "downstream_audience_id", "origin_protocol", "origin_protocol_version",
    "downstream_protocol", "downstream_protocol_version",
    "origin_envelope_sha256", "origin_receipt_sha256",
    "translated_payload_sha256", "message_id", "task_id", "context_id",
    "policy_version", "issued_at", "expires_at", "adapter_key_id",
    "transformation_build_sha256", "transformation_ruleset_sha256",
}


def canonical(value: dict) -> bytes:
    return rfc8785.dumps(value)


def without_signature(value: dict) -> dict:
    return {key: value[key] for key in value if key != "signature_b64"}


def verify_origin_receipt(
    receipt: dict,
    *,
    authenticated_peer_id: str,
    adapter_id: str,
    expected_origin_protocol: str,
    expected_origin_protocol_version: str,
    expected_policy_version: str,
    maximum_validity_seconds: int,
    resolve_verifier_key: Callable[[str], Ed25519PublicKey],
) -> None:
    if not isinstance(receipt, dict) or set(receipt) != ORIGIN_RECEIPT_FIELDS:
        raise PermissionError("origin_receipt_schema_mismatch")
    if (
        receipt["schema_version"] != "aidefend.origin-verification.v1"
        or receipt["outcome"] != "PASS"
        or receipt["authenticated_sender_id"] != authenticated_peer_id
        or receipt["audience_id"] != adapter_id
        or receipt["origin_protocol"] != expected_origin_protocol
        or receipt["origin_protocol_version"] != expected_origin_protocol_version
        or receipt["policy_version"] != expected_policy_version
    ):
        raise PermissionError("origin_receipt_binding_failed")
    now = datetime.now(timezone.utc)
    issued = datetime.fromisoformat(receipt["issued_at"].replace("Z", "+00:00"))
    expires = datetime.fromisoformat(receipt["expires_at"].replace("Z", "+00:00"))
    if (
        issued.utcoffset() is None
        or expires.utcoffset() is None
        or not issued.astimezone(timezone.utc) <= now < expires.astimezone(timezone.utc)
        or isinstance(maximum_validity_seconds, bool)
        or not isinstance(maximum_validity_seconds, int)
        or maximum_validity_seconds < 1
        or expires - issued > timedelta(seconds=maximum_validity_seconds)
    ):
        raise PermissionError("origin_receipt_expired")
    key = resolve_verifier_key(receipt["verifier_key_id"])
    try:
        key.verify(
            base64.b64decode(receipt["signature_b64"], validate=True),
            canonical(without_signature(receipt)),
        )
    except (InvalidSignature, ValueError) as exc:
        raise PermissionError("origin_receipt_signature_invalid") from exc


def translate_verified(
    *,
    origin_envelope: dict,
    origin_receipt: dict,
    authenticated_peer_id: str,
    adapter_id: str,
    downstream_audience_id: str,
    downstream_protocol: str,
    downstream_protocol_version: str,
    expected_origin_protocol: str,
    expected_origin_protocol_version: str,
    expected_policy_version: str,
    maximum_validity_seconds: int,
    transformation_build_sha256: str,
    transformation_ruleset_sha256: str,
    adapter_key_id: str,
    adapter_private_key: Ed25519PrivateKey,
    resolve_verifier_key,
    translate_payload,
) -> dict:
    for digest in (transformation_build_sha256, transformation_ruleset_sha256):
        if not isinstance(digest, str) or len(digest) != 64:
            raise ValueError("transformation_digest_invalid")
        int(digest, 16)
    verify_origin_receipt(
        origin_receipt,
        authenticated_peer_id=authenticated_peer_id,
        adapter_id=adapter_id,
        expected_origin_protocol=expected_origin_protocol,
        expected_origin_protocol_version=expected_origin_protocol_version,
        expected_policy_version=expected_policy_version,
        maximum_validity_seconds=maximum_validity_seconds,
        resolve_verifier_key=resolve_verifier_key,
    )
    origin_digest = hashlib.sha256(canonical(origin_envelope)).hexdigest()
    payload_digest = hashlib.sha256(canonical(origin_envelope["payload"])).hexdigest()
    if (
        origin_digest != origin_receipt["origin_envelope_sha256"]
        or payload_digest != origin_receipt["payload_sha256"]
    ):
        raise PermissionError("origin_envelope_or_payload_digest_mismatch")

    translated = translate_payload(origin_envelope["payload"])
    translated_digest = hashlib.sha256(canonical(translated)).hexdigest()
    attestation = {
        "schema_version": "aidefend.protocol-translation.v1",
        "original_sender_id": origin_receipt["authenticated_sender_id"],
        "adapter_id": adapter_id,
        "downstream_audience_id": downstream_audience_id,
        "origin_protocol": origin_receipt["origin_protocol"],
        "origin_protocol_version": origin_receipt["origin_protocol_version"],
        "downstream_protocol": downstream_protocol,
        "downstream_protocol_version": downstream_protocol_version,
        "origin_envelope_sha256": origin_digest,
        "origin_receipt_sha256": hashlib.sha256(
            canonical(origin_receipt)
        ).hexdigest(),
        "translated_payload_sha256": translated_digest,
        "message_id": origin_receipt["message_id"],
        "task_id": origin_receipt["task_id"],
        "context_id": origin_receipt["context_id"],
        "policy_version": origin_receipt["policy_version"],
        "issued_at": origin_receipt["issued_at"],
        "expires_at": origin_receipt["expires_at"],
        "adapter_key_id": adapter_key_id,
        "transformation_build_sha256": transformation_build_sha256,
        "transformation_ruleset_sha256": transformation_ruleset_sha256,
    }
    return {
        "payload": translated,
        "attestation": attestation,
        "adapter_signature_b64": base64.b64encode(
            adapter_private_key.sign(canonical(attestation))
        ).decode("ascii"),
    }


def verify_translated(
    wire: dict,
    *,
    expected_original_sender_id: str,
    expected_adapter_id: str,
    expected_downstream_audience_id: str,
    expected_origin_protocol: str,
    expected_origin_protocol_version: str,
    expected_downstream_protocol: str,
    expected_downstream_protocol_version: str,
    expected_task_id: str,
    expected_context_id: str,
    expected_policy_version: str,
    expected_transformation_build_sha256: str,
    expected_transformation_ruleset_sha256: str,
    maximum_validity_seconds: int,
    resolve_adapter_key,
    resolve_origin_key,
    load_origin_receipt,
    downstream_private_key: Ed25519PrivateKey,
) -> dict:
    if not isinstance(wire, dict) or set(wire) != {
        "payload", "attestation", "adapter_signature_b64",
    }:
        raise PermissionError("translated_wire_schema_mismatch")
    attestation = wire["attestation"]
    if not isinstance(attestation, dict) or set(attestation) != ATTESTATION_FIELDS:
        raise PermissionError("translation_attestation_schema_mismatch")
    expected = {
        "schema_version": "aidefend.protocol-translation.v1",
        "original_sender_id": expected_original_sender_id,
        "adapter_id": expected_adapter_id,
        "downstream_audience_id": expected_downstream_audience_id,
        "origin_protocol": expected_origin_protocol,
        "origin_protocol_version": expected_origin_protocol_version,
        "downstream_protocol": expected_downstream_protocol,
        "downstream_protocol_version": expected_downstream_protocol_version,
        "task_id": expected_task_id,
        "context_id": expected_context_id,
        "policy_version": expected_policy_version,
        "transformation_build_sha256": expected_transformation_build_sha256,
        "transformation_ruleset_sha256": expected_transformation_ruleset_sha256,
    }
    if any(attestation.get(field) != value for field, value in expected.items()):
        raise PermissionError("translation_attestation_binding_mismatch")
    for field in (
        "origin_envelope_sha256", "origin_receipt_sha256",
        "translated_payload_sha256", "transformation_build_sha256",
        "transformation_ruleset_sha256",
    ):
        value = attestation[field]
        if not isinstance(value, str) or len(value) != 64:
            raise PermissionError("translation_digest_invalid")
        int(value, 16)

    now = datetime.now(timezone.utc)
    issued = datetime.fromisoformat(attestation["issued_at"].replace("Z", "+00:00"))
    expires = datetime.fromisoformat(attestation["expires_at"].replace("Z", "+00:00"))
    if (
        issued.utcoffset() is None
        or expires.utcoffset() is None
        or not issued.astimezone(timezone.utc) <= now < expires.astimezone(timezone.utc)
        or isinstance(maximum_validity_seconds, bool)
        or not isinstance(maximum_validity_seconds, int)
        or maximum_validity_seconds < 1
        or expires - issued > timedelta(seconds=maximum_validity_seconds)
    ):
        raise PermissionError("translation_attestation_outside_validity_window")
    adapter_key = resolve_adapter_key(
        expected_adapter_id, attestation["adapter_key_id"], issued
    )
    try:
        adapter_key.verify(
            base64.b64decode(wire["adapter_signature_b64"], validate=True),
            canonical(attestation),
        )
    except (InvalidSignature, ValueError) as exc:
        raise PermissionError("adapter_signature_invalid") from exc
    translated_sha256 = hashlib.sha256(canonical(wire["payload"])).hexdigest()
    if translated_sha256 != attestation["translated_payload_sha256"]:
        raise PermissionError("translated_payload_digest_mismatch")

    origin_receipt = load_origin_receipt(attestation["origin_receipt_sha256"])
    if hashlib.sha256(canonical(origin_receipt)).hexdigest() != attestation["origin_receipt_sha256"]:
        raise PermissionError("origin_receipt_readback_digest_mismatch")
    verify_origin_receipt(
        origin_receipt,
        authenticated_peer_id=expected_original_sender_id,
        adapter_id=expected_adapter_id,
        expected_origin_protocol=expected_origin_protocol,
        expected_origin_protocol_version=expected_origin_protocol_version,
        expected_policy_version=expected_policy_version,
        maximum_validity_seconds=maximum_validity_seconds,
        resolve_verifier_key=resolve_origin_key,
    )
    cross_bindings = {
        "origin_envelope_sha256": attestation["origin_envelope_sha256"],
        "message_id": attestation["message_id"],
        "task_id": expected_task_id,
        "context_id": expected_context_id,
        "policy_version": expected_policy_version,
    }
    if any(origin_receipt.get(field) != value for field, value in cross_bindings.items()):
        raise PermissionError("origin_and_translation_binding_mismatch")

    decision = {
        "schema_version": "aidefend.protocol-translation-decision.v1",
        "control_id": "AID-H-004.003",
        "outcome": "PASS",
        "original_sender_id": expected_original_sender_id,
        "adapter_id": expected_adapter_id,
        "downstream_audience_id": expected_downstream_audience_id,
        "task_id": expected_task_id,
        "context_id": expected_context_id,
        "message_id": attestation["message_id"],
        "policy_version": expected_policy_version,
        "origin_receipt_sha256": attestation["origin_receipt_sha256"],
        "translated_payload_sha256": translated_sha256,
        "transformation_build_sha256": expected_transformation_build_sha256,
        "transformation_ruleset_sha256": expected_transformation_ruleset_sha256,
        "verified_at": now.isoformat(),
    }
    return {
        "payload": wire["payload"],
        "verified_context": decision,
        "downstream_signature_b64": base64.b64encode(
            downstream_private_key.sign(canonical(decision))
        ).decode("ascii"),
    }
Downstream verification and evidence

Evidence: retain both canonical digests, both signatures and key IDs, authenticated upstream peer, original sender, adapter and downstream identities, protocol/schema versions, transformation build/image digest, policy version, verifier decisions, trace ID, and independent replay result for the complete translated-message population.

AID-H-004.003-G007

Bind every cross-agent request to delegation context and authorize it at the receiving agent before any side effect.
Concept:

Cross-agent attacks often succeed when Agent B trusts that Agent A already checked the user's intent, scope, or task boundary. The receiving agent must make its own authorization decision using explicit request facts: root principal, requesting agent, target agent, delegation ID, task ID, requested capability, allowed scopes, expiry, and risk tier. This blocks confused-deputy and control-flow hijacking paths where a low-privilege or manipulated agent tries to make a more privileged agent act outside the original task.

Step 1: Require an inter-agent authorization envelope
# File: policy/inter_agent_request.json
{
  "root_principal": "user:12345",
  "requesting_agent": "spiffe://company.com/agent/support/prod",
  "target_agent": "spiffe://company.com/agent/billing/prod",
  "delegation_id": "dlg-7f1d4f59",
  "task_id": "ticket-88214",
  "requested_capability": "get_invoice_status",
  "allowed_capabilities": ["get_invoice_status"],
  "scope": ["read:invoice_status"],
  "issued_at": "2026-04-08T18:52:11Z",
  "expires_at": "2026-04-08T18:57:11Z",
  "risk_tier": "low",
  "nonce": "jti-5eeb7b7d"
}
Step 2: Enforce fail-closed receiver policy
# File: policy/inter_agent_authorization.rego
package aidefend.inter_agent

import future.keywords.if
import future.keywords.in

default decision := {
  "allow": false,
  "reason": "default_deny"
}

required_fields := {
  "root_principal",
  "requesting_agent",
  "target_agent",
  "delegation_id",
  "task_id",
  "requested_capability",
  "allowed_capabilities",
  "scope",
  "issued_at",
  "expires_at",
  "risk_tier",
  "nonce"
}

missing_required[field] if {
  field := required_fields[_]
  not input[field]
}

decision := {"allow": false, "reason": "missing_required_field"} if {
  count(missing_required) > 0
}

else := {"allow": false, "reason": "target_agent_mismatch"} if {
  input.target_agent != input.receiver_spiffe_id
}

else := {"allow": false, "reason": "capability_not_delegated"} if {
  not input.requested_capability in input.allowed_capabilities
}

else := {"allow": false, "reason": "scope_not_sufficient"} if {
  input.requested_capability == "get_invoice_status"
  not "read:invoice_status" in input.scope
}

else := {"allow": false, "reason": "expired_delegation"} if {
  time.parse_rfc3339_ns(input.expires_at) <= time.now_ns()
}

else := {"allow": false, "reason": "invalid_or_high_risk_tier"} if {
  not input.risk_tier in {"low", "medium"}
}

else := {"allow": true, "reason": "delegated_scope_valid"} if {
  count(missing_required) == 0
  input.target_agent == input.receiver_spiffe_id
  input.requested_capability in input.allowed_capabilities
  "read:invoice_status" in input.scope
  input.risk_tier in {"low", "medium"}
  time.parse_rfc3339_ns(input.issued_at) <= time.now_ns()
  time.now_ns() < time.parse_rfc3339_ns(input.expires_at)
}
Step 3: Treat payment or legal-commitment workflows as mandate-bound

For payment, procurement, contract, or other legally binding workflows, use a signed mandate or verifiable credential pattern where the business domain already requires it. This low/medium-risk receiver example deliberately denies all high, destructive, and monetary tiers; route those plans through the independent AID-H-018.003 approval owner before re-evaluating the exact envelope. Do not require a generic human-origin-intent token for every A2A request; that would overfit the framework to a control that is not yet broadly deployed. The production-grade invariant is narrower and enforceable: every receiver must see an auditable delegation envelope and must deny side effects outside the delegated task, scope, target, and approval state.

Action: Add an inter-agent authorization middleware before the tool or action dispatcher. The middleware, not the peer payload, injects the authenticated receiver identity and trusted current time, enforces single-use nonce consumption, and bounds the maximum delegation lifetime. Reject missing delegation context, target-agent mismatch, non-delegated capabilities, malformed/future/expired envelopes, every high-risk tier at this low-risk gate, and any request where the receiver cannot reconstruct who asked, on whose authority, for which task.

AID-H-004.003-G008

Pin the A2A v1.0 service parameter and validate each HTTP+JSON request against a digest-pinned binding contract before dispatch.
Concept

A2A v1 clients send the A2A-Version service parameter on each request. Pin the deployed profile to 1.0, reject missing/unsupported versions rather than assuming the older 0.3 profile, require the HTTP+JSON binding's media type, and validate the normalized HTTP request with schemas generated from the pinned lf.a2a.v1 specification. Do not invent a generic version JSON field.

# File: a2a/version_and_schema_gate.py
from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

from jsonschema import Draft202012Validator, FormatChecker


PROFILE = "1.0"
BINDING = "http-json"
MEDIA_TYPE = "application/a2a+json"
SCHEMA_DIR = Path("schemas/a2a-1.0")
EXPECTED_DIGESTS: dict[str, str] = json.loads(
    (SCHEMA_DIR / "sha256-manifest.json").read_text(encoding="utf-8")
)


class A2ARequestRejected(ValueError):
    pass


def build_validators() -> dict[str, Draft202012Validator]:
    validators: dict[str, Draft202012Validator] = {}
    for filename, expected_digest in EXPECTED_DIGESTS.items():
        if Path(filename).name != filename or not filename.endswith(".schema.json"):
            raise RuntimeError("invalid A2A schema manifest entry")
        raw = (SCHEMA_DIR / filename).read_bytes()
        if hashlib.sha256(raw).hexdigest() != expected_digest:
            raise RuntimeError("A2A schema digest mismatch: " + filename)
        schema = json.loads(raw)
        Draft202012Validator.check_schema(schema)
        validators[filename] = Draft202012Validator(
            schema, format_checker=FormatChecker()
        )
    return validators


VALIDATORS = build_validators()


def validate_a2a_request(
    *,
    a2a_version: str | None,
    binding: str,
    content_type: str | None,
    schema_filename: str,
    normalized_payload: Any,
) -> None:
    if a2a_version != PROFILE:
        raise A2ARequestRejected("A2A version not supported")
    if binding != BINDING:
        raise A2ARequestRejected("unsupported A2A binding for this adapter")
    media_type = (content_type or "").split(";", 1)[0].strip().lower()
    if media_type != MEDIA_TYPE:
        raise A2ARequestRejected("application/a2a+json required")
    validator = VALIDATORS.get(schema_filename)
    if validator is None:
        raise A2ARequestRejected("unapproved A2A operation schema")
    errors = sorted(validator.iter_errors(normalized_payload),
                    key=lambda item: [str(part) for part in item.path])
    if errors:
        raise A2ARequestRejected("A2A request schema mismatch")

Integration: use generated A2A v1.0 HTTP routes to normalize body, path, and query parameters into the corresponding request message, then pass that message and its fixed server-selected schema filename to this gate. Never accept a caller-selected schema name. Apply bounded body/header/query limits before parsing and authorize after schema validation but before task lookup.

Action and evidence: generate schemas from the exact reviewed A2A v1.0 protocol commit and verify all digests at startup. Test missing/0.3/future versions, wrong media type, unknown operation, malformed identifiers, extra fields, invalid enums, oversized inputs, and a caller attempt to select another schema. Preserve A2A version, spec/schema commit and digests, operation/binding, validation result, server build, and trace ID. Implement JSON-RPC, gRPC, or any future binding in a separate adapter with its own official contract and media/framing checks; this HTTP+JSON example must not be claimed as coverage for those bindings.

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-H-004.003: Secure Agent-to-Agent Communication | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service