feat: inter-service auth validation

CHANGELOG.md

···
 ### Added
 
 - `Atex.PLC` module for interacting with [a did:plc directory API](https://web.plc.directory/).
+- `Atex.ServiceAuth` module for validating [inter-service authentication tokens](https://atproto.com/specs/xrpc#inter-service-authentication-jwt).
 
 ### Fixed
 

README.md

···
 # atex
 
-An Elxir toolkit for the [AT Protocol](https://atproto.com).
+An Elixir toolkit for the [AT Protocol](https://atproto.com).
 
 ## Feature map
 
···
 - [x] XRPC client
   - With integration for generated Lexicon structs!
 - [ ] Repository reading and manipulation (MST & CAR)
-- [ ] Service auth
+- [x] Service auth
 - [x] PLC client
+- [ ] XRPC server router
 
 Looking to use a data subscription service like the Firehose, [Jetstream], or [Tap]? Check out [Drinkup].
 

examples/service_auth.ex

···
+defmodule ServiceAuthExample do
+  require Logger
+  use Plug.Router
+
+  plug :match
+  plug :dispatch
+
+  @did_doc JSON.encode!(%{
+             "@context" => [
+               "https://www.w3.org/ns/did/v1",
+               "https://w3id.org/security/multikey/v1"
+             ],
+             "id" => "did:web:setsuna.prawn-galaxy.ts.net",
+             "verificationMethod" => [
+               %{
+                 "id" => "did:web:setsuna.prawn-galaxy.ts.net#atproto",
+                 "type" => "Multikey",
+                 "controller" => "did:web:setsuna.prawn-galaxy.ts.net",
+                 "publicKeyMultibase" => "zDnaeRBG9swcjKP6GjjQF7kqxP6JaJaVbvjTjJ1YbXnKWWLna"
+               }
+             ],
+             "service" => [
+               %{
+                 "id" => "atex_test",
+                 "type" => "AtexTest",
+                 "serviceEndpoint" => "https://setsuna.prawn-galaxy.ts.net"
+               }
+             ]
+           })
+
+  get "/.well-known/did.json" do
+    Logger.info("got did json")
+
+    conn
+    |> put_resp_content_type("application/json")
+    |> send_resp(200, @did_doc)
+  end
+
+  get "/xrpc/com.ovyerus.example" do
+    IO.inspect(conn)
+
+    conn
+    |> send_resp(200, "")
+  end
+
+  match _ do
+    send_resp(conn, 404, "Not Found")
+  end
+end

lib/atex/application.ex

···
       Atex.IdentityResolver.Cache,
       Atex.OAuth.Cache,
       Atex.OAuth.SessionStore,
+      Atex.ServiceAuth.JTICache,
       {Mutex, name: Atex.SessionMutex}
     ]
 

lib/atex/crypto.ex

···
     _ -> {:error, :sign_failed}
   end
 
+  def generate_p256() do
+    JOSE.JWK.generate_key({:ec, "P-256"})
+  end
+
+  def generate_k256() do
+    JOSE.JWK.generate_key({:ec, "secp256k1"})
+  end
+
   # Private helpers
 
   @spec strip_did_key_prefix(String.t()) :: String.t()

lib/atex/identity_resolver/identity.ex

···
   @typedoc """
   The resolved DID document for an identity.
   """
-  @type document() :: Atex.IdentityResolver.DIDDocument.t()
+  @type document() :: Atex.PLC.DIDDocument.t()
 
   typedstruct do
     field :did, did(), enforce: true

lib/atex/plc/did_document.ex

···
     end
   end
 
+  @spec get_atproto_signing_key(t()) :: JOSE.JWK.t() | nil
+  def get_atproto_signing_key(%__MODULE__{} = doc) do
+    doc.verification_method
+    |> Enum.find(fn
+      %{id: id} -> String.ends_with?(id, "#atproto")
+    end)
+    |> case do
+      nil ->
+        nil
+
+      %{public_key_multibase: multibase} ->
+        {:ok, jwk} = Atex.Crypto.decode_did_key(multibase)
+        jwk
+
+      # TODO
+      _ ->
+        raise ArgumentError, message: "Legacy verification method keys are not yet supported"
+        # %{public_key_jwk: jwk} -> nil
+    end
+  end
+
   defp valid_pds_endpoint?(endpoint) do
     case URI.new(endpoint) do
       {:ok, uri} ->

lib/atex/service_auth.ex

···
+defmodule Atex.ServiceAuth do
+  @moduledoc """
+  Validating and working with inter-service authentication tokens.
+
+  Provides functions for validating [ATProto inter-service authentication JWTs](https://atproto.com/specs/xrpc#inter-service-authentication-jwt),
+  either from a raw token string or directly from an incoming `Plug.Conn`.
+
+  Validation covers:
+
+  - Token timing (`iat` not in the future, `exp` not in the past).
+  - Audience (`aud`) matching the caller-supplied expected value.
+  - Optional lexicon method (`lxm`) matching the caller-supplied expected value.
+  - Issuer DID resolution and signing-key verification via `Atex.IdentityResolver`.
+  - Replay prevention via `Atex.ServiceAuth.JTICache` - each `jti` nonce may
+    only be accepted once.
+
+  ## Configuration
+
+  The JTI cache implementation is pluggable. See `Atex.ServiceAuth.JTICache` for
+  details.
+  """
+
+  import Plug.Conn
+
+  @typedoc """
+  Options accepted by `validate_conn/2` and `validate_jwt/2`.
+
+  - `:aud` - **required**. The expected audience string. The token's `aud` claim
+    must equal this value exactly.
+  - `:lxm` - optional. When provided, the token's `lxm` claim must match. If
+    the token omits `lxm` the check is skipped; if the token carries `lxm` but
+    no expected value is configured, validation fails with `:lxm_not_configured`.
+  """
+  @type validate_option() :: {:lxm, String.t()} | {:aud, String.t()}
+
+  @doc """
+  Validate a service auth token from a `Plug.Conn` request.
+
+  Extracts the `Authorization: Bearer <jwt>` header and delegates to
+  `validate_jwt/2`. Returns `{:error, :missing_token}` when the header is
+  absent or malformed.
+
+  ## Options
+
+  See `t:validate_option/0`.
+
+  ## Examples
+
+      iex> Atex.ServiceAuth.validate_conn(conn, aud: "did:web:my-service.example")
+      {:ok, %JOSE.JWT{}}
+
+      iex> Atex.ServiceAuth.validate_conn(conn, aud: "did:web:my-service.example", lxm: "app.bsky.feed.getTimeline")
+      {:error, :lxm_mismatch}
+  """
+  @spec validate_conn(Plug.Conn.t(), list(validate_option())) ::
+          {:ok, jwt :: JOSE.JWT.t()} | {:error, reason :: atom()}
+  def validate_conn(conn, opts \\ []) do
+    case get_req_header(conn, "authorization") do
+      ["Bearer " <> jwt] -> validate_jwt(jwt, opts)
+      [_] -> :error
+      _ -> :error
+    end
+  end
+
+  @doc """
+  Validate a raw service auth JWT string.
+
+  Performs the full validation pipeline:
+
+  1. Decodes the JWT payload (without verifying the signature yet) to extract claims.
+  2. Validates `:aud` and `:lxm` against the provided options.
+  3. Validates token timing (`iat`, `exp`).
+  4. Resolves the issuer DID and retrieves the ATProto signing key from their
+     DID document.
+  5. Verifies the JWT signature with the resolved key.
+  6. Records the `jti` nonce in `Atex.ServiceAuth.JTICache` - returns
+     `{:error, :replayed_token}` if it has already been seen.
+
+  ## Options
+
+  See `t:validate_option/0`.
+
+  ## Error reasons
+
+  - `:aud_mismatch` - `aud` claim does not match the expected audience.
+  - `:lxm_mismatch` - `lxm` claim does not match the expected lexicon method.
+  - `:lxm_not_configured` - token carries an `lxm` claim but no expected value
+    was provided via `:lxm` opt.
+  - `:future_iat` - `iat` is in the future.
+  - `:expired` - `exp` is in the past.
+  - `:replayed_token` - `jti` has already been used.
+
+  ## Examples
+
+      iex> Atex.ServiceAuth.validate_jwt(jwt, aud: "did:web:my-service.example")
+      {:ok, %JOSE.JWT{}}
+
+      iex> Atex.ServiceAuth.validate_jwt(expired_jwt, aud: "did:web:my-service.example")
+      {:error, :expired}
+  """
+  @spec validate_jwt(String.t(), list(validate_option())) ::
+          {:ok, jwt :: JOSE.JWT.t()} | {:error, reason :: atom()}
+  def validate_jwt(jwt, opts \\ []) do
+    {expected_aud, expected_lxm} = options(opts)
+
+    %{
+      fields:
+        %{
+          "aud" => target_aud,
+          "iat" => iat,
+          "exp" => exp,
+          "iss" => issuing_did,
+          "jti" => nonce
+        } = fields
+    } = JOSE.JWT.peek(jwt)
+
+    target_lxm = Map.get(fields, "lxm")
+
+    with :ok <- validate_aud(expected_aud, target_aud),
+         :ok <- validate_lxm(expected_lxm, target_lxm),
+         :ok <- validate_token_times(iat, exp),
+         # Resolve JWT's issuer to: a) make sure it's a real identity, b) get
+         # the signing key from their DID document to verify the token
+         {:ok, identity} <- Atex.IdentityResolver.resolve(issuing_did),
+         user_jwk when not is_nil(user_jwk) <-
+           Atex.PLC.DIDDocument.get_atproto_signing_key(identity.document),
+         {true, %JOSE.JWT{} = jwt_struct, _jws} <- JOSE.JWT.verify(user_jwk, jwt),
+         # Record the nonce atomically after successful verification. insert_new
+         # is used under the hood so this returns :seen if the jti was already
+         # consumed, preventing replay attacks.
+         :ok <- Atex.ServiceAuth.JTICache.put(nonce, exp) do
+      {:ok, jwt_struct}
+    else
+      :seen -> {:error, :replayed_token}
+      err -> err
+    end
+  end
+
+  @spec validate_token_times(integer(), integer()) :: :ok | {:error, reason :: atom()}
+  defp validate_token_times(iat, exp) do
+    now = DateTime.utc_now()
+
+    with {:ok, iat} <- DateTime.from_unix(iat),
+         {:ok, exp} <- DateTime.from_unix(exp) do
+      cond do
+        DateTime.before?(now, iat) ->
+          {:error, :future_iat}
+
+        DateTime.after?(now, exp) ->
+          {:error, :expired}
+
+        true ->
+          :ok
+      end
+    end
+  end
+
+  @spec validate_aud(String.t(), String.t()) :: :ok | {:error, reason :: atom()}
+  defp validate_aud(expected, target) when expected == target, do: :ok
+  defp validate_aud(_expected, _target), do: {:error, :aud_mismatch}
+
+  @spec validate_lxm(String.t() | nil, String.t() | nil) :: :ok | {:error, reason :: atom()}
+  defp validate_lxm(expected, target) when expected == target, do: :ok
+  # `lxm` in JWTs is currently optional so we can do this.
+  # TODO: should have an option to force requirement (e.g. for security-sensitive operations)
+  defp validate_lxm(expected, nil) when is_binary(expected), do: :ok
+  defp validate_lxm(nil, target) when is_binary(target), do: {:error, :lxm_not_configured}
+  defp validate_lxm(expected, target) when expected != target, do: {:error, :lxm_mismatch}
+
+  @spec options(list(validate_option())) :: {aud :: String.t(), lxm :: String.t() | nil}
+  defp options(opts) do
+    opts = Keyword.validate!(opts, aud: nil, lxm: nil)
+    aud = Keyword.get(opts, :aud)
+    lxm = Keyword.get(opts, :lxm)
+
+    if !aud do
+      raise ArgumentError, "`:aud` option is required for service auth validation"
+    end
+
+    {aud, lxm}
+  end
+end

lib/atex/service_auth/jti_cache.ex

···
+defmodule Atex.ServiceAuth.JTICache do
+  @moduledoc """
+  Behaviour and compile-time dispatch for tracking used `jti` (JWT ID) nonces
+  from service auth tokens, preventing replay attacks.
+
+  Implementations are responsible for:
+
+  - Storing a `jti` alongside its expiry so that entries can be evicted once
+    the corresponding token has naturally expired (avoiding unbounded growth).
+  - Returning `:seen` when a `jti` has already been recorded, and `:ok` when it
+    is new (and recording it atomically).
+
+  ## Configuration
+
+  The active implementation is resolved at compile time:
+
+  ```elixir
+  config :atex, :jti_cache, Atex.ServiceAuth.JTICache.ETS
+  ```
+
+  Defaults to `Atex.ServiceAuth.JTICache.ETS` when not configured.
+  """
+
+  @cache Application.compile_env(:atex, :jti_cache, Atex.ServiceAuth.JTICache.ETS)
+
+  @doc """
+  Record a `jti` as seen. The implementation must store it until at least
+  `expires_at` (a Unix timestamp integer) so that expired tokens cannot be
+  replayed before the entry is evicted.
+
+  Returns `:ok` if this is the first time the `jti` has been seen, or `:seen`
+  if it was already present.
+  """
+  @callback put(jti :: String.t(), expires_at :: integer()) :: :ok | :seen
+
+  @doc """
+  Check whether a `jti` has already been seen without modifying the cache.
+
+  Returns `:ok` if unseen, `:seen` if already present.
+  """
+  @callback get(jti :: String.t()) :: :ok | :seen
+
+  @doc """
+  Get the child specification for starting the cache in a supervision tree.
+  """
+  @callback child_spec(any()) :: Supervisor.child_spec()
+
+  defdelegate put(jti, expires_at), to: @cache
+  defdelegate get(jti), to: @cache
+  @doc false
+  defdelegate child_spec(opts), to: @cache
+end

lib/atex/service_auth/jti_cache/ets.ex

···
+defmodule Atex.ServiceAuth.JTICache.ETS do
+  @moduledoc """
+  ConCache-backed implementation of `Atex.ServiceAuth.JTICache`.
+
+  Each `jti` is stored with a per-item TTL derived from the token's own `exp`
+  claim, so entries are evicted automatically once the corresponding token could
+  no longer be presented as valid. This keeps memory use proportional to the
+  number of currently-live tokens rather than growing without bound.
+
+  The TTL check interval defaults to 30 seconds and can be overridden:
+
+  ```elixir
+  config :atex, Atex.ServiceAuth.JTICache.ETS, ttl_check_interval: :timer.seconds(10)
+  ```
+  """
+
+  @behaviour Atex.ServiceAuth.JTICache
+  use Supervisor
+
+  @cache :atex_service_auth_jti_cache
+  @default_ttl_check_interval :timer.seconds(30)
+
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl Supervisor
+  def init(_opts) do
+    ttl_check_interval =
+      Application.get_env(:atex, __MODULE__, [])
+      |> Keyword.get(:ttl_check_interval, @default_ttl_check_interval)
+
+    children = [
+      {ConCache,
+       [
+         name: @cache,
+         ttl_check_interval: ttl_check_interval,
+         # No global TTL - each entry sets its own based on token expiry.
+         global_ttl: :infinity
+       ]}
+    ]
+
+    Supervisor.init(children, strategy: :one_for_one)
+  end
+
+  @impl Atex.ServiceAuth.JTICache
+  @spec put(String.t(), integer()) :: :ok | :seen
+  def put(jti, expires_at) do
+    now_unix = System.os_time(:second)
+    remaining_ms = max((expires_at - now_unix) * 1_000, 0)
+
+    result =
+      ConCache.insert_new(@cache, jti, %ConCache.Item{
+        value: true,
+        ttl: remaining_ms
+      })
+
+    case result do
+      :ok -> :ok
+      {:error, :already_exists} -> :seen
+    end
+  end
+
+  @impl Atex.ServiceAuth.JTICache
+  @spec get(String.t()) :: :ok | :seen
+  def get(jti) do
+    case ConCache.get(@cache, jti) do
+      nil -> :ok
+      _ -> :seen
+    end
+  end
+end

test/atex/crypto_test.exs

···
   # secp256k1 compressed public key as multikey
   @k256_multikey "zQ3shqwJEJyMBsBXCWyCBpUBMqxcon9oHB7mCvx4sSpMdLJwc"
 
+  # Spec example legacy K-256 key (uncompressed, no multicodec) from
+  # https://atproto.com/specs/did#legacy-representation
+  @legacy_k256_multibase "zQYEBzXeuTM9UR3rfvNag6L3RNAs5pQZyYPsomTsgQhsxLdEgCrPTLgFna8yqCnxPpNT7DBk6Ym3dgPKNu86vt9GR"
+  # The same key in the current Multikey (compressed) format
+  @multikey_k256_same "zQ3shXjHeiBuRCKmM36cuYnm7YEMzhGnCmCyW92sRJ9pribSF"
+
   # ---------------------------------------------------------------------------
   # decode_did_key/1
   # ---------------------------------------------------------------------------
···
       {_, orig_map} = JOSE.JWK.to_map(jwk)
       {_, decoded_map} = JOSE.JWK.to_map(jwk2)
       assert orig_map == decoded_map
+    end
+  end
+
+  # ---------------------------------------------------------------------------
+  # decode_legacy_multibase/2
+  # ---------------------------------------------------------------------------
+
+  describe "decode_legacy_multibase/2" do
+    test "decodes a legacy K-256 uncompressed multibase into a JOSE JWK" do
+      assert {:ok, jwk} =
+               Crypto.decode_legacy_multibase(
+                 "EcdsaSecp256k1VerificationKey2019",
+                 @legacy_k256_multibase
+               )
+
+      assert %JOSE.JWK{} = jwk
+      {_, map} = JOSE.JWK.to_map(jwk)
+      assert map["kty"] == "EC"
+      assert map["crv"] == "secp256k1"
+    end
+
+    test "legacy and Multikey encoding of the same K-256 key produce equal JWK coordinates" do
+      {:ok, jwk_legacy} =
+        Crypto.decode_legacy_multibase(
+          "EcdsaSecp256k1VerificationKey2019",
+          @legacy_k256_multibase
+        )
+
+      {:ok, jwk_current} = Crypto.decode_did_key(@multikey_k256_same)
+
+      {_, legacy_map} = JOSE.JWK.to_map(jwk_legacy)
+      {_, current_map} = JOSE.JWK.to_map(jwk_current)
+      assert legacy_map["x"] == current_map["x"]
+      assert legacy_map["y"] == current_map["y"]
+    end
+
+    test "decodes a generated P-256 key from its uncompressed form" do
+      priv = JOSE.JWK.generate_key({:ec, "P-256"})
+      {_, map} = priv |> JOSE.JWK.to_public() |> JOSE.JWK.to_map()
+      x = Base.url_decode64!(map["x"], padding: false)
+      y = Base.url_decode64!(map["y"], padding: false)
+      legacy_mb = Multiformats.Multibase.encode(<<0x04>> <> x <> y, :base58btc)
+
+      assert {:ok, jwk} =
+               Crypto.decode_legacy_multibase("EcdsaSecp256r1VerificationKey2019", legacy_mb)
+
+      {_, decoded_map} = JOSE.JWK.to_map(jwk)
+      assert decoded_map["x"] == map["x"]
+      assert decoded_map["y"] == map["y"]
+    end
+
+    test "returns {:error, :unsupported_curve} for an unknown type" do
+      assert {:error, :unsupported_curve} =
+               Crypto.decode_legacy_multibase("UnknownType", @legacy_k256_multibase)
+    end
+
+    test "returns {:error, :invalid_multikey} for a bad multibase string" do
+      assert {:error, :invalid_multikey} =
+               Crypto.decode_legacy_multibase("EcdsaSecp256k1VerificationKey2019", "not-base58")
+    end
+
+    test "returns {:error, :invalid_point} when bytes are not a valid uncompressed point" do
+      bad_bytes = <<0x04>> <> :crypto.strong_rand_bytes(10)
+      bad_mb = Multiformats.Multibase.encode(bad_bytes, :base58btc)
+
+      assert {:error, :invalid_point} =
+               Crypto.decode_legacy_multibase("EcdsaSecp256k1VerificationKey2019", bad_mb)
     end
   end