annot.at is a service for syncing static websites, such as blogs, to atproto and standard.site
annot.at
elixir
atproto
standardsite
Authenticated XRPC queries
Johanna Larsson
(Jun 16, 2026, 7:40 AM +0100)
3aa18043
59e616a6
+295
-14
+18
lib/annot_at/accounts.ex
+18
lib/annot_at/accounts.ex
···
108
108
{count, _} = Repo.delete_all(from(r in OAuthLoginRequest, where: r.inserted_at < ^cutoff))
109
109
count
110
110
end
111
+
112
+
@doc """
113
+
Runs fun with a user and their atproto session under a `FOR UPDATE` row
114
+
lock, inside a transaction. How we do token refresh.
115
+
"""
116
+
@spec with_locked_session(integer(), (User.t(), AtprotoSession.t() ->
117
+
{:ok, term()} | {:error, term()})) :: term()
118
+
def with_locked_session(user_id, fun) do
119
+
Repo.transact(fn ->
120
+
query =
121
+
from(s in AtprotoSession, where: s.user_id == ^user_id, lock: "FOR UPDATE")
122
+
123
+
case Repo.one(query) do
124
+
nil -> {:error, :no_session}
125
+
atproto_session -> fun.(get_user!(user_id), atproto_session)
126
+
end
127
+
end)
128
+
end
111
129
end
+128
lib/annot_at/atproto/oauth/client.ex
+128
lib/annot_at/atproto/oauth/client.ex
···
1
+
defmodule AnnotAt.Atproto.OAuth.Client do
2
+
@moduledoc """
3
+
Authenticated XRPC for a logged in user with transparent token refresh.
4
+
5
+
Loads the user's stored session, ensures the access token is current
6
+
(refreshing if not), and delegates to `XRPC`.
7
+
"""
8
+
9
+
alias AnnotAt.Accounts
10
+
alias AnnotAt.Atproto.OAuth.Config
11
+
alias AnnotAt.Atproto.OAuth.Discovery
12
+
alias AnnotAt.Atproto.OAuth.DPoP
13
+
alias AnnotAt.Atproto.OAuth.Flow
14
+
alias AnnotAt.Atproto.OAuth.Session
15
+
alias AnnotAt.Atproto.XRPC
16
+
17
+
require Logger
18
+
19
+
@refresh_buffer_seconds 60
20
+
21
+
@doc """
22
+
Performs an authenticated XRPC query for the user, refreshing if needed.
23
+
"""
24
+
@spec query(integer(), String.t(), keyword()) ::
25
+
{:ok, map()}
26
+
| {:error,
27
+
:no_session
28
+
| :refresh_failed
29
+
| :missing_dpop_nonce
30
+
| :invalid_json
31
+
| {:transport, term()}
32
+
| {:xrpc_error, pos_integer(), map()}}
33
+
def query(user_id, method, params \\ []) do
34
+
call(user_id, fn session -> XRPC.query(session, method, params) end)
35
+
end
36
+
37
+
defp call(user_id, fun) do
38
+
with {:ok, session} <- fresh_session(user_id) do
39
+
with {:error, {:xrpc_error, 401, _}} <- fun.(session) do
40
+
with {:ok, refreshed} <- refresh(user_id, session.access_token) do
41
+
fun.(refreshed)
42
+
end
43
+
end
44
+
end
45
+
end
46
+
47
+
defp fresh_session(user_id) do
48
+
case load(user_id) do
49
+
nil ->
50
+
{:error, :no_session}
51
+
52
+
session ->
53
+
if expired?(session) do
54
+
refresh(user_id, session.access_token)
55
+
else
56
+
{:ok, session}
57
+
end
58
+
end
59
+
end
60
+
61
+
defp refresh(user_id, stale_token) do
62
+
Accounts.with_locked_session(user_id, fn user, atproto_session ->
63
+
session = to_session(user, atproto_session)
64
+
65
+
if session.access_token == stale_token do
66
+
do_refresh(user_id, session)
67
+
else
68
+
{:ok, session}
69
+
end
70
+
end)
71
+
end
72
+
73
+
defp do_refresh(user_id, session) do
74
+
result =
75
+
with {:ok, server} <- Discovery.discover(session.pds_endpoint),
76
+
{:ok, new_session} <-
77
+
Flow.refresh(server, session,
78
+
client_id: Config.client_id(),
79
+
client_jwk: Config.signing_key()
80
+
),
81
+
{:ok, _} <- persist(user_id, new_session) do
82
+
{:ok, new_session}
83
+
end
84
+
85
+
with {:error, reason} <- result do
86
+
Logger.warning("token refresh failed for user #{user_id}: #{inspect(reason)}")
87
+
{:error, :refresh_failed}
88
+
end
89
+
end
90
+
91
+
defp persist(user_id, session) do
92
+
Accounts.upsert_session(user_id, %{
93
+
auth_server_issuer: session.issuer,
94
+
granted_scopes: session.scope,
95
+
access_token: session.access_token,
96
+
refresh_token: session.refresh_token,
97
+
dpop_private_jwk: DPoP.dump(session.dpop_key),
98
+
expires_at: session.expires_at
99
+
})
100
+
end
101
+
102
+
defp load(user_id) do
103
+
user = Accounts.get_user(user_id)
104
+
session = Accounts.get_atproto_session(user_id)
105
+
106
+
if user && session do
107
+
to_session(user, session)
108
+
end
109
+
end
110
+
111
+
defp to_session(user, session) do
112
+
%Session{
113
+
did: user.did,
114
+
access_token: session.access_token,
115
+
refresh_token: session.refresh_token,
116
+
dpop_key: DPoP.load(session.dpop_private_jwk),
117
+
scope: session.granted_scopes,
118
+
issuer: session.auth_server_issuer,
119
+
pds_endpoint: user.pds_host,
120
+
expires_at: session.expires_at
121
+
}
122
+
end
123
+
124
+
defp expired?(session) do
125
+
threshold = DateTime.add(DateTime.utc_now(), @refresh_buffer_seconds, :second)
126
+
DateTime.compare(session.expires_at, threshold) != :gt
127
+
end
128
+
end
+19
lib/annot_at/atproto/oauth/dpop.ex
+19
lib/annot_at/atproto/oauth/dpop.ex
···
69
69
PKCE.challenge(access_token)
70
70
end
71
71
72
+
@doc """
73
+
Serializes a JWK to a JSON string for storage and whatnot.
74
+
"""
75
+
@spec dump(JOSE.JWK.t()) :: String.t()
76
+
def dump(jwk) do
77
+
{_, map} = JOSE.JWK.to_map(jwk)
78
+
Jason.encode!(map)
79
+
end
80
+
81
+
@doc """
82
+
Deserializes a JWK from a JSON string.
83
+
"""
84
+
@spec load(String.t()) :: JOSE.JWK.t()
85
+
def load(json) do
86
+
json
87
+
|> Jason.decode!()
88
+
|> JOSE.JWK.from()
89
+
end
90
+
72
91
defp htu(url) do
73
92
url
74
93
|> URI.parse()
+3
-14
lib/annot_at/atproto/oauth/login.ex
+3
-14
lib/annot_at/atproto/oauth/login.ex
···
114
114
pds_host: identity.pds_endpoint,
115
115
auth_server_issuer: server.issuer,
116
116
pkce_verifier: verifier,
117
-
dpop_private_jwk: dump_jwk(dpop_key)
117
+
dpop_private_jwk: DPoP.dump(dpop_key)
118
118
})
119
119
end
120
120
···
145
145
redirect_uri: Config.redirect_uri(),
146
146
code: code,
147
147
code_verifier: request.pkce_verifier,
148
-
dpop_key: load_jwk(request.dpop_private_jwk),
148
+
dpop_key: DPoP.load(request.dpop_private_jwk),
149
149
expected_did: request.did,
150
150
pds_endpoint: request.pds_host
151
151
)
···
168
168
granted_scopes: session.scope,
169
169
access_token: session.access_token,
170
170
refresh_token: session.refresh_token,
171
-
dpop_private_jwk: dump_jwk(session.dpop_key),
171
+
dpop_private_jwk: DPoP.dump(session.dpop_key),
172
172
expires_at: session.expires_at
173
173
}
174
174
)
175
-
end
176
-
177
-
defp dump_jwk(jwk) do
178
-
{_, map} = JOSE.JWK.to_map(jwk)
179
-
Jason.encode!(map)
180
-
end
181
-
182
-
defp load_jwk(json) do
183
-
json
184
-
|> Jason.decode!()
185
-
|> JOSE.JWK.from()
186
175
end
187
176
188
177
defp random_state do
+126
test/annot_at/atproto/oauth/client_test.exs
+126
test/annot_at/atproto/oauth/client_test.exs
···
1
+
defmodule AnnotAt.Atproto.OAuth.ClientTest do
2
+
use AnnotAt.DataCase, async: true
3
+
use Mimic
4
+
5
+
alias AnnotAt.Accounts
6
+
alias AnnotAt.Atproto.OAuth.Client
7
+
alias AnnotAt.Atproto.OAuth.Discovery
8
+
alias AnnotAt.Atproto.OAuth.Flow
9
+
alias AnnotAt.Atproto.OAuth.ServerMetadata
10
+
alias AnnotAt.Atproto.OAuth.Session
11
+
alias AnnotAt.Atproto.XRPC
12
+
13
+
@did "did:plc:ewvi7nxzyoun6zhxrhs64oiz"
14
+
@pds "https://shaggymane.us-west.host.bsky.network"
15
+
@issuer "https://bsky.social"
16
+
17
+
@dpop_jwk_json "../../../support/fixtures/atproto/es256_jwk.json"
18
+
|> Path.expand(__DIR__)
19
+
|> File.read!()
20
+
21
+
test "query/3 uses the current token when it is not expired" do
22
+
user = create_user(future())
23
+
reject(&Flow.refresh/3)
24
+
25
+
expect(XRPC, :query, fn %Session{access_token: "access-old"},
26
+
"app.bsky.actor.getProfile",
27
+
_ ->
28
+
{:ok, %{"handle" => "jola.dev"}}
29
+
end)
30
+
31
+
assert {:ok, %{"handle" => "jola.dev"}} =
32
+
Client.query(user.id, "app.bsky.actor.getProfile", actor: @did)
33
+
end
34
+
35
+
test "query/3 refreshes an expired token before calling, and persists it" do
36
+
user = create_user(~U[2020-01-01 00:00:00Z])
37
+
server = server()
38
+
39
+
expect(Discovery, :discover, fn @pds -> {:ok, server} end)
40
+
41
+
expect(Flow, :refresh, fn ^server, %Session{refresh_token: "refresh-old"}, _opts ->
42
+
{:ok, refreshed_session()}
43
+
end)
44
+
45
+
expect(XRPC, :query, fn %Session{access_token: "access-new"}, _method, _ ->
46
+
{:ok, %{"ok" => true}}
47
+
end)
48
+
49
+
assert {:ok, %{"ok" => true}} = Client.query(user.id, "app.bsky.actor.getProfile")
50
+
assert "access-new" == Accounts.get_atproto_session(user.id).access_token
51
+
end
52
+
53
+
test "query/3 refreshes and retries on a 401" do
54
+
user = create_user(future())
55
+
server = server()
56
+
57
+
expect(XRPC, :query, fn %Session{access_token: "access-old"}, _, _ ->
58
+
{:error, {:xrpc_error, 401, %{}}}
59
+
end)
60
+
61
+
expect(Discovery, :discover, fn @pds -> {:ok, server} end)
62
+
expect(Flow, :refresh, fn ^server, _session, _opts -> {:ok, refreshed_session()} end)
63
+
64
+
expect(XRPC, :query, fn %Session{access_token: "access-new"}, _, _ ->
65
+
{:ok, %{"retried" => true}}
66
+
end)
67
+
68
+
assert {:ok, %{"retried" => true}} = Client.query(user.id, "app.bsky.actor.getProfile")
69
+
end
70
+
71
+
test "query/3 returns :no_session for a user without a session" do
72
+
{:ok, user} = Accounts.upsert_user(%{did: @did, handle: "jola.dev", pds_host: @pds})
73
+
reject(&XRPC.query/3)
74
+
75
+
assert {:error, :no_session} = Client.query(user.id, "app.bsky.actor.getProfile")
76
+
end
77
+
78
+
defp create_user(expires_at) do
79
+
{:ok, user} =
80
+
Accounts.upsert_login(
81
+
%{did: @did, handle: "jola.dev", pds_host: @pds},
82
+
%{
83
+
auth_server_issuer: @issuer,
84
+
granted_scopes: "atproto",
85
+
access_token: "access-old",
86
+
refresh_token: "refresh-old",
87
+
dpop_private_jwk: @dpop_jwk_json,
88
+
expires_at: expires_at
89
+
}
90
+
)
91
+
92
+
user
93
+
end
94
+
95
+
defp future do
96
+
DateTime.utc_now()
97
+
|> DateTime.add(3600, :second)
98
+
|> DateTime.truncate(:second)
99
+
end
100
+
101
+
defp server do
102
+
%ServerMetadata{
103
+
issuer: @issuer,
104
+
authorization_endpoint: "#{@issuer}/oauth/authorize",
105
+
token_endpoint: "#{@issuer}/oauth/token",
106
+
par_endpoint: "#{@issuer}/oauth/par",
107
+
scopes_supported: ["atproto"]
108
+
}
109
+
end
110
+
111
+
defp refreshed_session do
112
+
%Session{
113
+
did: @did,
114
+
access_token: "access-new",
115
+
refresh_token: "refresh-new",
116
+
dpop_key:
117
+
@dpop_jwk_json
118
+
|> Jason.decode!()
119
+
|> JOSE.JWK.from(),
120
+
scope: "atproto",
121
+
issuer: @issuer,
122
+
pds_endpoint: @pds,
123
+
expires_at: future()
124
+
}
125
+
end
126
+
end
+1
test/test_helper.exs
+1
test/test_helper.exs