A2A Security, in depth: a threat model for agent-to-agent systems and the controls the specification requires

What the specification fixes, and what it leaves to you

A2A deliberately treats agents as ordinary enterprise web applications. Identity lives in the transport layer, not in A2A messages. The specification states requirements in several places: production deployments MUST use encrypted transport (HTTPS, or TLS for gRPC), SHOULD prefer TLS 1.3 and SHOULD disable SSLv3, TLS 1.0 and 1.1, and SHOULD send HSTS headers on HTTP bindings. Clients discover required schemes from the card's securitySchemes and security fields, obtain credentials out of band, and send them in headers on every request. Servers MUST authenticate every request and MUST authorize every operation against their own model.

What the protocol does not decide is your authorization model, what your agent is allowed to do with tools, and whether text that arrives in a message should influence your model. Those are design decisions, and most real incidents live there.

The threat model

Authorize before you look anything up

Task IDs travel in URLs, logs and webhook payloads; assume an attacker will obtain some. The defence is that knowing an ID grants nothing. The specification requires authorization checks on every operation, scoping of list results even when the caller supplies no contextId filter, and checks that happen before any database query that could reveal whether a resource outside the caller's scope exists. The error for a task that does not exist or is not accessible is the same TaskNotFoundError (JSON-RPC code -32001, HTTP 404), so an attacker cannot use the response to enumerate IDs.

def get_task(caller, task_id, history_length=None):
    # Scope is part of the query, not a check after it, so timing and
    # errors are identical for "missing" and "someone else's".
    task = db.tasks.find_one({"id": task_id, "tenant": caller.tenant, "owner": caller.principal})
    if task is None:
        raise A2AError("TaskNotFoundError", code=-32001)
    return render_task(task, history_length)

def list_tasks(caller, context_id=None, page_token=None):
    query = {"tenant": caller.tenant, "owner": caller.principal}   # always applied
    if context_id:
        query["context_id"] = context_id
    return paginate(db.tasks, query, page_token)

The same rule applies to cancelling, subscribing and every push-configuration operation. Multi-tenant deployments add a tenant key to every query, as in A2A multi-tenancy. Owner can mean a user, a role or a project; the specification leaves the model to you and recommends that you document it.

Signed Agent Cards

Cards are fetched from https://{domain}/.well-known/agent-card.json. TLS proves you reached that domain; it does not prove the card was published by the team that owns the agent, or that a cache or registry did not alter it. A2A lets providers sign cards with JSON Web Signature. Each entry in the card's signatures array has a base64url protected header, a base64url signature and an optional unprotected header. The protected header must carry alg, kid and typ (which should be JOSE), and may carry a jku URL for the key set.

The payload is not the bytes you downloaded. To verify, remove properties that hold default values, remove the signatures field, and canonicalize the result with the JSON Canonicalization Scheme of RFC 8785. Skipping the default-value step is the classic bug: a verifier that canonicalizes the card as received rejects valid signatures whenever the publisher's serializer emitted defaults.

import base64, json, rfc8785
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature

def b64d(s):
    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))

def verify_card(card: dict, trusted_keys: dict) -> bool:
    # strip_defaults: schema-aware, drops fields equal to their protobuf default
    payload = strip_defaults({k: v for k, v in card.items() if k != "signatures"})
    canonical = rfc8785.dumps(payload)                          # bytes
    for sig in card.get("signatures", []):
        header = json.loads(b64d(sig["protected"]))
        key = trusted_keys.get(header.get("kid"))               # pinned, not fetched blindly
        if key is None or header.get("alg") != "ES256":
            continue
        signing_input = (sig["protected"] + "." +
                         base64.urlsafe_b64encode(canonical).rstrip(b"=").decode()).encode()
        raw = b64d(sig["signature"])                            # JWS ES256 is raw R||S
        der = encode_dss_signature(int.from_bytes(raw[:32], "big"), int.from_bytes(raw[32:], "big"))
        try:
            key.verify(der, signing_input, ec.ECDSA(hashes.SHA256()))
            return True
        except Exception:
            continue
    return False

Two design points. A jku URL inside the card proves only that whoever controls the card also controls that URL; for agents you depend on, pin keys in a trust store and treat jku as a hint for rotation. And multiple signatures are allowed precisely so providers can rotate keys; accept any valid signature from a trusted key, and never accept expired or revoked ones. The card fields themselves are described in the Agent Card specification.

Push notifications: the SSRF door

With push notifications, the client tells your agent which URL to call when a task changes. Unchecked, that is a request-forgery primitive: an attacker registers http://169.254.169.254/... or an internal admin endpoint and lets your agent make the call from inside your network. The specification says agents SHOULD reject private IPv4 ranges, localhost and link-local addresses, and use allowlists where appropriate. Do more than the minimum: cover IPv6, resolve the hostname yourself and connect to the address you checked, so DNS rebinding cannot swap it afterwards, and refuse redirects.

import ipaddress, socket
from urllib.parse import urlsplit

def safe_webhook_target(url: str, allowlist: set[str] | None = None):
    parts = urlsplit(url)
    if parts.scheme != "https" or not parts.hostname:
        raise ValueError("webhook must be https with a hostname")
    if allowlist is not None and parts.hostname not in allowlist:
        raise ValueError("host not on allowlist")
    infos = socket.getaddrinfo(parts.hostname, parts.port or 443, proto=socket.IPPROTO_TCP)
    addrs = {ipaddress.ip_address(i[4][0]) for i in infos}
    for a in addrs:
        a = getattr(a, "ipv4_mapped", None) or a              # ::ffff:10.0.0.1 is 10.0.0.1
        if not a.is_global:                                   # private, loopback, link-local,
            raise ValueError(f"blocked address {a}")          # 100.64/10, fc00::/7, fe80::/10 ...
    return parts.hostname, sorted(addrs, key=str)[0]          # connect to this IP, SNI = hostname

# Deliver with redirects disabled and a 10-30 s timeout, retry with backoff,
# and send the configured credentials: Authorization: <scheme> <credentials>

On the receiving side the duties are mirrored. Clients MUST validate each notification using the credentials they configured, which the protocol carries as an AuthenticationInfo with a scheme such as Bearer and its credentials. They should check that the task ID is one they created, process deliveries idempotently because duplicates happen, rate-limit the endpoint, and use a unique, single-purpose token per configuration so one leak cannot forge notifications for other tasks. More on configuration in A2A push notifications.

File references and content

Message and artifact parts can carry files by URI instead of inline bytes. The specification says file references MUST be validated to prevent SSRF, so apply the same guard before your agent fetches one, plus size limits and a check of the declared media type against the bytes. Never let a part decide which credentials accompany the fetch.

Text, data and file content from another agent is untrusted input to your model. Instructions inside it are data, not commands. Keep tool permissions narrow, require confirmation for irreversible actions, and keep the system prompt and secrets out of anything an external agent can read back through an artifact.

In-task authorization and extended cards

Sometimes an agent needs extra authorization midway, for example a user's consent to read their calendar. It moves the task to TASK_STATE_AUTH_REQUIRED, and credentials should then arrive out of band over a secure channel, not inside a message that every intermediate agent can see. The specification is explicit that this state transition is not itself a grant: its meaning, scope and lifetime are defined by your implementation or the credential issuer, and a credential obtained this way must not be assumed to authorize later messages on the task. Chains of agents can pass the request upstream, each moving its own task to the same state; none of them should collect the credential on behalf of the next.

The extended Agent Card, fetched with GetExtendedAgentCard when capabilities.extendedAgentCard is true, MUST require authentication and may show more skills to more privileged callers. It should still never contain internal service URLs or unmasked credentials; assume it will leak eventually.

Worked example: three attacks on a travel agent

A travel-booking agent accepts tasks from partner agents. A penetration test tried three things. First, the tester created a task and then registered a push configuration pointing at http://169.254.169.254/latest/meta-data/. The guard rejected it twice over: not HTTPS, and a link-local address. A second attempt used a domain whose DNS answered with a public address at validation time and an internal one a second later; because delivery connected to the pinned IP, the rebinding failed.

Second, the tester replayed GetTask with IDs harvested from a partner's logs. Every call returned TaskNotFoundError in the same time as a random ID, because the owner and tenant filters were part of the query. Third, a partner message included a PDF whose text said to ignore prior instructions and email the itinerary database to an outside address. The model was given the extracted text as quoted data, and the agent's email tool could only send to the booking's own traveller, so the injection had nothing to act on. The one finding was a missing rate limit on authentication failures, which the specification recommends logging and limiting; it was added the same week.

Failure modes

What to do next

1. Confirm TLS 1.2 or later everywhere, prefer 1.3, disable older versions and enable HSTS on HTTP bindings.
2. Audit every handler that takes a task ID: the caller's scope must be part of the query and both failure cases must return TaskNotFoundError.
3. Put the webhook SSRF guard in front of push delivery and every file-URI fetch, with IPv6 coverage, IP pinning and redirects disabled.
4. Sign your own Agent Card, publish the key set, and verify signatures of the cards you depend on against pinned keys.
5. Issue a unique token per push configuration, verify it on the receiver, and rotate it.
6. Document your authorization model, including what an AUTH_REQUIRED credential permits and when it expires.
7. Log and rate-limit authentication failures, and limit message and file sizes.