A2A.Plug.JWTVerifier (A2A v0.3.0)

JWT verification utilities for A2A principal authentication.

Provides JWT verification for authenticating principals (users, agents, or services) accessing A2A endpoints. Signature verification is delegated to Joken/JOSE rather than hand-rolled crypto.

Features

JWT signature verification via Joken (HS256)
Expiration (exp) and not-before (nbf) validation with clock-skew tolerance
Issuer and audience verification
Configurable required claims (default: ["sub"])

Usage with HMAC (HS256)

# Configure HMAC-based JWT verification
verifier = A2A.Plug.JWTVerifier.new(
  secret: "your-secret-key",
  algorithm: "HS256",
  issuer: "https://auth.example.com",
  audience: "a2a-api"
)

# Use with A2A.Plug.Auth
plug A2A.Plug.Auth,
  schemes: %{
    "jwt_auth" => %A2A.SecurityScheme.HTTPAuth{scheme: "bearer"}
  },
  verify: fn _name, token, _conn ->
    A2A.Plug.JWTVerifier.verify(verifier, token)
  end

Configuration Options

:secret — HMAC secret key (for HS256)
:algorithm — Signature algorithm: "HS256" (default: "HS256")
:issuer — Expected issuer claim (optional)
:audience — Expected audience claim (optional)
:required_claims — List of claim names that must be present (default: ["sub"])
:clock_skew — Allowed clock skew in seconds (default: 60)

This module is only compiled when both :plug and :joken are available.

Summary

Types

claim_map()

verifier()

Functions

new(opts)

Creates a new JWT verifier configuration.

verify(config, token)

Verifies a JWT token and returns the claims.

Types

claim_map()

@type claim_map() :: %{required(String.t()) => any()}

verifier()

@type verifier() :: %{
  secret: String.t() | nil,
  algorithm: String.t(),
  issuer: String.t() | nil,
  audience: String.t() | nil,
  required_claims: [String.t()],
  clock_skew: non_neg_integer()
}

Functions

new(opts)

@spec new(keyword()) :: verifier()

Creates a new JWT verifier configuration.

Accepts a keyword list of options. See the module documentation for the full list of supported keys.

Examples

iex> v = A2A.Plug.JWTVerifier.new(secret: "s3cret")
iex> v.algorithm
"HS256"
iex> v.required_claims
["sub"]

iex> v = A2A.Plug.JWTVerifier.new(secret: "s", issuer: "iss", clock_skew: 120)
iex> {v.issuer, v.clock_skew}
{"iss", 120}

verify(config, token)

@spec verify(verifier(), String.t()) :: {:ok, claim_map()} | {:error, String.t()}

Verifies a JWT token and returns the claims.

Performs the following checks in order:

Decodes and validates the JWT header (algorithm match)
Verifies the cryptographic signature via Joken
Validates required claims, issuer, audience, expiration, and not-before

Returns {:ok, claims} on success or {:error, reason} with a human-readable description of what failed.

Examples

verifier = A2A.Plug.JWTVerifier.new(secret: "my-secret")

case A2A.Plug.JWTVerifier.verify(verifier, token) do
  {:ok, claims} -> IO.inspect(claims["sub"])
  {:error, reason} -> IO.puts("Auth failed: #{reason}")
end