From 488245f9f9b0db9459f93d7f5a81ae28ec95b531 Mon Sep 17 00:00:00 2001 From: Johanna Larsson Date: Sun, 26 Jul 2026 20:40:35 +0100 Subject: [PATCH] Adds Latch.Store.ETS To make it easier to integrate without losing flexibility, this commit brings a pre-built Store implementation using ETS. The downside is that it's in-memory and any restart loses in-progress requests as well as any existing access tokens. In an app this would be experienced as getting logged out early. Not ideal, but also not the worst. For an implementation that survives restarts, take a look at annot.at's Store implementation here https://tangled.org/jola.dev/annot.at/blob/c42010e2b62972e445de4a3a9c89c99155fab4ff/lib/annot_at/latch_store.ex --- README.md | 13 +- lib/latch/config.ex | 4 +- lib/latch/store/ets.ex | 164 +++++++++++++++ test/latch/store/ets_test.exs | 368 ++++++++++++++++++++++++++++++++++ test/latch_test.exs | 12 +- test/support/ets_store.ex | 4 + 6 files changed, 554 insertions(+), 11 deletions(-) create mode 100644 lib/latch/store/ets.ex create mode 100644 test/latch/store/ets_test.exs create mode 100644 test/support/ets_store.ex diff --git a/README.md b/README.md index 40863e4..e99db4a 100644 --- a/README.md +++ b/README.md @@ -18,10 +18,17 @@ atproto OAuth and client library attempting to follow the specification strictly ### Setting it up +Using the built-in ETS `Store` implementation, create your module like: + + defmodule MyApp.LatchStore do + use Latch.Store.ETS + end + Add `Latch` to your supervision tree, giving it a unique name and a `Latch.Store` implementation: children = [ + {MyApp.LatchStore, []}, {Latch, name: MyApp.Latch, mode: :confidential, @@ -32,9 +39,9 @@ and a `Latch.Store` implementation: signing_key: System.fetch_env!("ATPROTO_CLIENT_PRIVATE_JWK")} ] -`signing_key` is the JWK from step 2 above. Optional keys: `:client_name`, `:client_uri` and `request_ttl`. +`signing_key` is the JWK from step 2 above. Optional keys: `:client_name`, `:client_uri` and `request_ttl`. Instead of using the built-in ETS Store implementation you can create your own, implementing the `Latch.Store` behavior. -Set up a route to serve the client metadata. +You need to set up a route to serve the client metadata. def client_metadata(conn, _params) do json(conn, Latch.client_metadata(MyApp.Latch)) @@ -104,7 +111,7 @@ be found at . - [x] DPoP nonce caching - [x] Public client - [x] Local client -- [ ] Built-in ETS LatchStore implementation +- [x] Built-in ETS LatchStore implementation - [ ] Getting started guide - [ ] Extensive tests diff --git a/lib/latch/config.ex b/lib/latch/config.ex index c7d133e..38fe631 100644 --- a/lib/latch/config.ex +++ b/lib/latch/config.ex @@ -38,7 +38,7 @@ defmodule Latch.Config do name: atom() | pid(), client_name: String.t() | nil, client_uri: String.t() | nil, - request_ttl: pos_integer() + request_ttl: non_neg_integer() } @modes [:confidential, :public, :localhost] @@ -52,7 +52,7 @@ defmodule Latch.Config do name: [type: {:or, [:atom, :pid]}, required: true], client_name: [type: :string, required: false], client_uri: [type: :string, required: false], - request_ttl: [type: :pos_integer, required: false, default: @default_request_ttl], + request_ttl: [type: :non_neg_integer, required: false, default: @default_request_ttl], mode: [type: {:in, @modes}, required: true] ] diff --git a/lib/latch/store/ets.ex b/lib/latch/store/ets.ex new file mode 100644 index 0000000..d5938d1 --- /dev/null +++ b/lib/latch/store/ets.ex @@ -0,0 +1,164 @@ +defmodule Latch.Store.ETS do + @moduledoc """ + Pre-built ETS Store for Latch. Does not maintain access tokens or in-progress requests + across restarts. For a more reliable approach, create your own implementation of + the Latch.Store behavior. + + TTL for requests is set on the Latch level, passed by the implementor as `request_ttl`, + defaulting to 600s. + """ + + require Logger + + defmacro __using__(_) do + quote do + use GenServer + + alias Latch.Store.ETS, as: ETSStore + + require Logger + + @behaviour Latch.Store + + @sweep_interval_ms :timer.hours(24) + + @table_options [ + :public, + :set, + :named_table, + {:read_concurrency, true} + ] + + def put_request(state, request, ttl) do + ETSStore.put_request(requests_table(), state, request, ttl) + end + + def take_request(state) do + ETSStore.take_request(requests_table(), state) + end + + def fetch_session(did) do + ETSStore.fetch_session(sessions_table(), did) + end + + def put_session(did, session) do + ETSStore.put_session(sessions_table(), did, session) + end + + def delete_session(did) do + ETSStore.delete_session(sessions_table(), did) + end + + def update_session(did, fun) do + GenServer.call(__MODULE__, {:update_session, did, fun}) + catch + :exit, reason -> + Logger.warning("Latch.Store.ETS implementation raised with: #{inspect(reason)}") + {:error, :backend_error} + end + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + @impl GenServer + def init(opts) do + sweep_disabled = Keyword.get(opts, :sweep_disabled, false) + sweep_interval_ms = Keyword.get(opts, :sweep_interval_ms, @sweep_interval_ms) + + requests = :ets.new(requests_table(), @table_options) + sessions = :ets.new(sessions_table(), @table_options) + + schedule_sweep(sweep_disabled, sweep_interval_ms) + + {:ok, + %{ + sweep_disabled: sweep_disabled, + sweep_interval_ms: sweep_interval_ms, + requests: requests, + sessions: sessions + }} + end + + @impl GenServer + def handle_call({:update_session, did, fun}, _from, state) do + result = ETSStore.update_session(sessions_table(), did, fun) + {:reply, result, state} + end + + @impl GenServer + def handle_info(:sweep, state) do + schedule_sweep(state.sweep_disabled, state.sweep_interval_ms) + + {:noreply, state} + end + + defp schedule_sweep(sweep_disabled, sweep_interval_ms) do + if not sweep_disabled do + now = System.monotonic_time(:second) + :ets.select_delete(requests_table(), [{{:_, :_, :"$1"}, [{:<, :"$1", now}], [true]}]) + + Process.send_after(self(), :sweep, sweep_interval_ms) + end + end + + defp requests_table do + :"#{__MODULE__}_requests" + end + + defp sessions_table do + :"#{__MODULE__}_sessions" + end + end + end + + def put_request(ref, state, request, ttl) do + sweep_after_timestamp = System.monotonic_time(:second) + ttl + :ets.insert(ref, {state, request, sweep_after_timestamp}) + :ok + end + + def take_request(ref, state) do + case :ets.take(ref, state) do + [] -> + {:error, :not_found} + + [{^state, request, sweep_after_timestamp} | _] -> + if sweep_after_timestamp > System.monotonic_time(:second) do + {:ok, request} + else + {:error, :not_found} + end + end + end + + def fetch_session(ref, did) do + case :ets.lookup(ref, did) do + [] -> {:error, :not_found} + [{^did, session} | _] -> {:ok, session} + end + end + + def put_session(ref, did, session) do + :ets.insert(ref, {did, session}) + :ok + end + + def delete_session(ref, did) do + :ets.delete(ref, did) + :ok + end + + def update_session(ref, did, fun) do + with {:ok, session} <- fetch_session(ref, did) do + with {:ok, session} <- fun.(session) do + :ok = put_session(ref, did, session) + {:ok, session} + end + end + rescue + error -> + Logger.warning("Latch.Store.ETS implementation raised with: #{inspect(error)}") + {:error, :backend_error} + end +end diff --git a/test/latch/store/ets_test.exs b/test/latch/store/ets_test.exs new file mode 100644 index 0000000..caffaad --- /dev/null +++ b/test/latch/store/ets_test.exs @@ -0,0 +1,368 @@ +defmodule Latch.Store.ETSTest do + use ExUnit.Case, async: true + use Mimic + + alias Latch.Discovery + alias Latch.DPoP + alias Latch.Flow + alias Latch.Identity + alias Latch.ServerMetadata + alias Latch.Session + alias Latch.XRPC + + @did "did:plc:bvraa6gajy4tfr3eh2sisdkr" + @handle "jola.dev" + @client_id "https://client.example.com/oauth-client-metadata.json" + @redirect_uri "https://client.example.com/oauth/callback" + @issuer "https://issuer.example.com" + @pds "https://pds.example.com" + @dpop_key DPoP.generate_key() + + describe "integration test suite" do + test "lifecycle" do + pid = start_latch() + + # AUTHORIZE SECTION + + identity = %Identity{did: @did, handle: @handle, pds_endpoint: @pds} + server = server() + request_uri = "urn:ietf:params:oauth:request_uri:request" + + expect(Identity, :resolve_handle, fn _handle -> {:ok, identity} end) + expect(Discovery, :discover, fn _pds, _opts -> {:ok, server} end) + + expect(Flow, :par, fn _config, _server, opts -> + send(self(), {:state, opts[:state]}) + {:ok, request_uri} + end) + + {:ok, _redirect_url} = Latch.authorize(pid, @handle) + + assert_receive {:state, state} + + # CALLBACK SECTION + + session = %Session{ + did: @did, + access_token: "access-token", + refresh_token: "refresh-token", + dpop_key: @dpop_key, + scope: "atproto", + issuer: @issuer, + pds_endpoint: @pds, + expires_at: DateTime.add(DateTime.utc_now(), 5, :minute) + } + + expect(Flow, :exchange_code, fn _config, _opts -> + {:ok, session} + end) + + assert {:ok, %{did: @did, handle: @handle}} = + Latch.callback( + pid, + %{ + "state" => state, + "iss" => @issuer, + "code" => "auth-code" + } + ) + + # CLIENT SECTION + + expect(XRPC, :query, fn _config, _session, _method, _opts -> + {:ok, %{"did" => @did}} + end) + + assert {:ok, %{"did" => @did}} = + Latch.query(pid, @did, "app.bsky.actor.getProfile", actor: @did) + + assert :ok = Latch.delete_session(pid, @did) + + reject(&XRPC.query/4) + + assert {:error, %Latch.Error.NoSession{}} = + Latch.query(pid, @did, "app.bsky.actor.getProfile", actor: @did) + end + + test "callback is single use" do + pid = start_latch() + + # AUTHORIZE SECTION + + identity = %Identity{did: @did, handle: @handle, pds_endpoint: @pds} + server = server() + request_uri = "urn:ietf:params:oauth:request_uri:request" + + expect(Identity, :resolve_handle, fn _handle -> {:ok, identity} end) + expect(Discovery, :discover, fn _pds, _opts -> {:ok, server} end) + + expect(Flow, :par, fn _config, _server, opts -> + send(self(), {:state, opts[:state]}) + {:ok, request_uri} + end) + + {:ok, _redirect_url} = Latch.authorize(pid, @handle) + + assert_receive {:state, state} + + # CALLBACK SECTION + + session = %Session{ + did: @did, + access_token: "access-token", + refresh_token: "refresh-token", + dpop_key: @dpop_key, + scope: "atproto", + issuer: @issuer, + pds_endpoint: @pds, + expires_at: DateTime.add(DateTime.utc_now(), 5, :minute) + } + + expect(Flow, :exchange_code, fn _config, _opts -> + {:ok, session} + end) + + assert {:ok, %{did: @did, handle: @handle}} = + Latch.callback( + pid, + %{ + "state" => state, + "iss" => @issuer, + "code" => "auth-code" + } + ) + + assert {:error, %Latch.Error.SecurityViolation{}} = + Latch.callback( + pid, + %{ + "state" => state, + "iss" => @issuer, + "code" => "auth-code" + } + ) + end + + test "expired request" do + pid = start_latch(request_ttl: 0) + + # AUTHORIZE SECTION + + identity = %Identity{did: @did, handle: @handle, pds_endpoint: @pds} + server = server() + request_uri = "urn:ietf:params:oauth:request_uri:request" + + expect(Identity, :resolve_handle, fn _handle -> {:ok, identity} end) + expect(Discovery, :discover, fn _pds, _opts -> {:ok, server} end) + + expect(Flow, :par, fn _config, _server, opts -> + send(self(), {:state, opts[:state]}) + {:ok, request_uri} + end) + + {:ok, _redirect_url} = Latch.authorize(pid, @handle) + + assert_receive {:state, state} + + assert {:error, %Latch.Error.SecurityViolation{reason: :state_mismatch}} = + Latch.callback( + pid, + %{ + "state" => state, + "iss" => @issuer, + "code" => "auth-code" + } + ) + end + + test "refresh token" do + pid = start_latch() + + # AUTHORIZE SECTION + + identity = %Identity{did: @did, handle: @handle, pds_endpoint: @pds} + server = server() + request_uri = "urn:ietf:params:oauth:request_uri:request" + + expect(Identity, :resolve_handle, fn _handle -> {:ok, identity} end) + expect(Discovery, :discover, fn _pds, _opts -> {:ok, server} end) + + expect(Flow, :par, fn _config, _server, opts -> + send(self(), {:state, opts[:state]}) + {:ok, request_uri} + end) + + {:ok, _redirect_url} = Latch.authorize(pid, @handle) + + assert_receive {:state, state} + + # CALLBACK SECTION + + session = %Session{ + did: @did, + access_token: "access-token", + refresh_token: "refresh-token", + dpop_key: @dpop_key, + scope: "atproto", + issuer: @issuer, + pds_endpoint: @pds, + expires_at: DateTime.utc_now(:second) + } + + expect(Flow, :exchange_code, fn _config, _opts -> + {:ok, session} + end) + + expect(Discovery, :discover, fn _pds, _opts -> {:ok, server} end) + + expect(Flow, :refresh, fn _config, _server, old_session, _opts -> + assert session == old_session + + {:ok, + %Session{ + did: @did, + access_token: "access-token", + refresh_token: "refresh-token", + dpop_key: @dpop_key, + scope: "atproto", + issuer: @issuer, + pds_endpoint: @pds, + expires_at: DateTime.add(DateTime.utc_now(), 5, :minute) + }} + end) + + assert {:ok, %{did: @did}} = + Latch.callback( + pid, + %{ + "state" => state, + "iss" => @issuer, + "code" => "auth-code" + } + ) + + expect(XRPC, :query, fn _config, _session, _method, _opts -> + {:ok, %{"did" => @did}} + end) + + assert {:ok, %{"did" => @did}} = + Latch.query(pid, @did, "app.bsky.actor.getProfile", actor: @did) + end + + test "refresh storm" do + pid = start_latch() + + # AUTHORIZE SECTION + + identity = %Identity{did: @did, handle: @handle, pds_endpoint: @pds} + server = server() + request_uri = "urn:ietf:params:oauth:request_uri:request" + + expect(Identity, :resolve_handle, fn _handle -> {:ok, identity} end) + expect(Discovery, :discover, fn _pds, _opts -> {:ok, server} end) + + expect(Flow, :par, fn _config, _server, opts -> + send(self(), {:state, opts[:state]}) + {:ok, request_uri} + end) + + {:ok, _redirect_url} = Latch.authorize(pid, @handle) + + assert_receive {:state, state} + + # CALLBACK SECTION + + session = %Session{ + did: @did, + access_token: "access-token", + refresh_token: "refresh-token", + dpop_key: @dpop_key, + scope: "atproto", + issuer: @issuer, + pds_endpoint: @pds, + expires_at: DateTime.utc_now(:second) + } + + expect(Flow, :exchange_code, fn _config, _opts -> + {:ok, session} + end) + + expect(Discovery, :discover, fn _pds, _opts -> {:ok, server} end) + + # Refresh only happens once even when we trigger refresh 10 times + # by using XPRC in a `for` below. + expect(Flow, :refresh, fn _config, _server, old_session, _opts -> + assert session == old_session + + {:ok, + %Session{ + did: @did, + access_token: "access-token-2", + refresh_token: "refresh-token", + dpop_key: @dpop_key, + scope: "atproto", + issuer: @issuer, + pds_endpoint: @pds, + expires_at: DateTime.add(DateTime.utc_now(), 5, :minute) + }} + end) + + assert {:ok, %{did: @did}} = + Latch.callback( + pid, + %{ + "state" => state, + "iss" => @issuer, + "code" => "auth-code" + } + ) + + expect(XRPC, :query, 10, fn _config, _session, _method, _opts -> + {:ok, %{"did" => @did}} + end) + + tasks = + for _ <- 1..10 do + Task.async(fn -> + assert {:ok, %{"did" => @did}} = + Latch.query(pid, @did, "app.bsky.actor.getProfile", actor: @did) + end) + end + + Task.await_many(tasks) + end + end + + defp start_latch(overrides \\ []) do + name = String.to_atom("latch_#{inspect(self())}") + store_pid = start_link_supervised!({Latch.ETSStore, []}) + Mimic.allow(Latch.Flow, self(), store_pid) + Mimic.allow(Latch.Discovery, self(), store_pid) + + opts = + Keyword.merge( + [ + name: name, + store: Latch.ETSStore, + client_id: @client_id, + redirect_uri: @redirect_uri, + scope: "atproto", + signing_key: Jason.encode!(@dpop_key), + mode: :confidential + ], + overrides + ) + + start_link_supervised!({Latch, opts}) + end + + defp server do + %ServerMetadata{ + issuer: @issuer, + authorization_endpoint: @issuer <> "/oauth/authorize", + token_endpoint: @issuer <> "/oauth/token", + par_endpoint: @issuer <> "/oauth/par", + scopes_supported: ["atproto"] + } + end +end diff --git a/test/latch_test.exs b/test/latch_test.exs index 6c340b1..ebd5679 100644 --- a/test/latch_test.exs +++ b/test/latch_test.exs @@ -29,7 +29,7 @@ defmodule LatchTest do client_id: @client_id, redirect_uri: @redirect_uri, scope: "atproto", - signing_key: Jason.encode!(Latch.DPoP.generate_key()), + signing_key: Jason.encode!(DPoP.generate_key()), mode: :confidential ) @@ -90,7 +90,7 @@ defmodule LatchTest do client_id: @client_id, redirect_uri: @redirect_uri, scope: "atproto", - signing_key: Jason.encode!(Latch.DPoP.generate_key()), + signing_key: Jason.encode!(DPoP.generate_key()), mode: :confidential ) @@ -152,7 +152,7 @@ defmodule LatchTest do client_id: @client_id, redirect_uri: @redirect_uri, scope: "atproto", - signing_key: Jason.encode!(Latch.DPoP.generate_key()), + signing_key: Jason.encode!(DPoP.generate_key()), mode: :confidential ) @@ -173,7 +173,7 @@ defmodule LatchTest do client_id: @client_id, redirect_uri: @redirect_uri, scope: "atproto", - signing_key: Jason.encode!(Latch.DPoP.generate_key()), + signing_key: Jason.encode!(DPoP.generate_key()), mode: :confidential ) @@ -200,7 +200,7 @@ defmodule LatchTest do client_id: @client_id, redirect_uri: @redirect_uri, scope: "atproto", - signing_key: Jason.encode!(Latch.DPoP.generate_key()), + signing_key: Jason.encode!(DPoP.generate_key()), mode: :confidential ) @@ -220,7 +220,7 @@ defmodule LatchTest do client_id: @client_id, redirect_uri: @redirect_uri, scope: "atproto", - signing_key: Jason.encode!(Latch.DPoP.generate_key()), + signing_key: Jason.encode!(DPoP.generate_key()), mode: :confidential ) diff --git a/test/support/ets_store.ex b/test/support/ets_store.ex new file mode 100644 index 0000000..adebefa --- /dev/null +++ b/test/support/ets_store.ex @@ -0,0 +1,4 @@ +defmodule Latch.ETSStore do + @moduledoc false + use Latch.Store.ETS +end -- 2.51.2