annot.at is a service for syncing static websites, such as blogs, to atproto and standard.site
annot.at
elixir
atproto
standardsite
Add atproto OAuth crypto primitives
First batch of atproto related code. Roughly inspired by `atex` and lots of digging. Going to want to go over this code again properly cause it's all scary crypto stuff. Wouldn't normally want that kind of code "handrolled" but it's all fairly standard and eg assent doesn't support atproto, so here we are.
Johanna Larsson
(Jun 11, 2026, 10:08 PM +0100)
354d7d8e
af4a3fd6
+488
+81
lib/annot_at/atproto/oauth/client_assertion.ex
+81
lib/annot_at/atproto/oauth/client_assertion.ex
···
1
+
defmodule AnnotAt.Atproto.OAuth.ClientAssertion do
2
+
@moduledoc """
3
+
Client assertion JWTs (RFC 7523 `private_key_jwt`) for atproto OAuth.
4
+
5
+
Confidential clients authenticate to the authorization server by signing
6
+
a fresh assertion for every PAR and token request.
7
+
"""
8
+
9
+
@algorithm "ES256"
10
+
@assertion_type "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
11
+
@lifetime_seconds 60
12
+
13
+
@doc """
14
+
The `client_assertion_type` request parameter value.
15
+
"""
16
+
@spec assertion_type() :: String.t()
17
+
def assertion_type do
18
+
@assertion_type
19
+
end
20
+
21
+
@doc """
22
+
Signs a client assertion JWT.
23
+
24
+
## Arguments
25
+
- `jwk` - the client's private JOSE JWK
26
+
- `client_id` - used as both `iss` and `sub`
27
+
- `audience` - the authorization server's `issuer` URL
28
+
29
+
## Options
30
+
- `:jti` - override `jti` (tests)
31
+
- `:iat` - override `iat` (tests)
32
+
33
+
`exp` is `iat` + #{@lifetime_seconds}s: atproto does not require it,
34
+
but RFC 7523 does, and servers expect assertions younger than a minute.
35
+
"""
36
+
37
+
@spec sign(JOSE.JWK.t(), String.t(), String.t(), keyword()) :: String.t()
38
+
def sign(jwk, client_id, audience, opts \\ []) do
39
+
jti = Keyword.get(opts, :jti, random_b64(20))
40
+
iat = Keyword.get(opts, :iat, System.os_time(:second))
41
+
42
+
jws = %{
43
+
"alg" => @algorithm,
44
+
"typ" => "JWT",
45
+
"kid" => kid(jwk)
46
+
}
47
+
48
+
claims = %{
49
+
"iss" => client_id,
50
+
"sub" => client_id,
51
+
"aud" => audience,
52
+
"jti" => jti,
53
+
"iat" => iat,
54
+
"exp" => iat + @lifetime_seconds
55
+
}
56
+
57
+
jwk
58
+
|> JOSE.JWT.sign(jws, claims)
59
+
|> JOSE.JWS.compact()
60
+
|> elem(1)
61
+
end
62
+
63
+
@doc """
64
+
Key ID for the client signing key: the JWK's own `kid` if set, otherwise
65
+
its RFC 7638 thumbprint.
66
+
67
+
The published client metadata JWKs must use the same value so the
68
+
authorization server can match assertion headers to a key.
69
+
"""
70
+
@spec kid(JOSE.JWK.t()) :: String.t()
71
+
def kid(jwk) do
72
+
{_, map} = JOSE.JWK.to_map(jwk)
73
+
Map.get_lazy(map, "kid", fn -> JOSE.JWK.thumbprint(jwk) end)
74
+
end
75
+
76
+
defp random_b64(bytes) do
77
+
bytes
78
+
|> :crypto.strong_rand_bytes()
79
+
|> Base.url_encode64(padding: false)
80
+
end
81
+
end
+59
lib/annot_at/atproto/oauth/client_metadata.ex
+59
lib/annot_at/atproto/oauth/client_metadata.ex
···
1
+
defmodule AnnotAt.Atproto.OAuth.ClientMetadata do
2
+
@moduledoc """
3
+
Client metadata document for atproto OAuth.
4
+
5
+
Served at the `client_id` URL. Authorization servers fetch it to discover
6
+
redirect URIs, scopes, and the client's public signing key.
7
+
"""
8
+
9
+
alias AnnotAt.Atproto.OAuth.ClientAssertion
10
+
11
+
@doc """
12
+
Builds the client metadata document as a JSON-serializable map.
13
+
14
+
## Options
15
+
- `:client_id` (required) - full URL of the metadata document itself
16
+
- `:redirect_uris` (required) - list of callback URLs
17
+
- `:scope` (required) - space-separated scopes, must include `atproto`
18
+
- `:jwk` (required) - client signing key, only the public part is published
19
+
- `:client_name` (optional)
20
+
- `:client_uri` (optional) - must share the `client_id` hostname
21
+
"""
22
+
@spec build(keyword()) :: map()
23
+
def build(opts) do
24
+
scope = Keyword.fetch!(opts, :scope)
25
+
26
+
if "atproto" not in String.split(scope, " ") do
27
+
raise ArgumentError, "scope must include atproto, got #{inspect(scope)}"
28
+
end
29
+
30
+
client_id = Keyword.fetch!(opts, :client_id)
31
+
redirect_uris = Keyword.fetch!(opts, :redirect_uris)
32
+
client_name = Keyword.get(opts, :client_name)
33
+
client_uri = Keyword.get(opts, :client_uri)
34
+
jwk = Keyword.fetch!(opts, :jwk)
35
+
36
+
%{
37
+
"client_id" => client_id,
38
+
"application_type" => "web",
39
+
"grant_types" => ["authorization_code", "refresh_token"],
40
+
"response_types" => ["code"],
41
+
"redirect_uris" => redirect_uris,
42
+
"scope" => scope,
43
+
"dpop_bound_access_tokens" => true,
44
+
"token_endpoint_auth_method" => "private_key_jwt",
45
+
"token_endpoint_auth_signing_alg" => "ES256",
46
+
"jwks" => %{"keys" => [public_jwk(jwk)]}
47
+
}
48
+
|> maybe_put("client_name", client_name)
49
+
|> maybe_put("client_uri", client_uri)
50
+
end
51
+
52
+
defp public_jwk(jwk) do
53
+
{_, public} = JOSE.JWK.to_public_map(jwk)
54
+
Map.put(public, "kid", ClientAssertion.kid(jwk))
55
+
end
56
+
57
+
defp maybe_put(map, _key, nil), do: map
58
+
defp maybe_put(map, key, value), do: Map.put(map, key, value)
59
+
end
+87
lib/annot_at/atproto/oauth/dpop.ex
+87
lib/annot_at/atproto/oauth/dpop.ex
···
1
+
defmodule AnnotAt.Atproto.OAuth.DPoP do
2
+
@moduledoc """
3
+
DPoP (RFC 9449) proof JWTs for atproto OAuth.
4
+
"""
5
+
6
+
alias AnnotAt.Atproto.OAuth.PKCE
7
+
8
+
@curve "P-256"
9
+
@algorithm "ES256"
10
+
@jwt_type "dpop+jwt"
11
+
12
+
@doc """
13
+
Generates a new ES256 (P-256) key pair as a JOSE JWK.
14
+
"""
15
+
@spec generate_key() :: JOSE.JWK.t()
16
+
def generate_key do
17
+
JOSE.JWK.generate_key({:ec, @curve})
18
+
end
19
+
20
+
@doc """
21
+
Signs a DPoP proof JWT for an HTTP request.
22
+
## Arguments
23
+
- `jwk` — private JOSE JWK for this OAuth session
24
+
- `method` — HTTP method (e.g. `"POST"`)
25
+
- `url` — request URL; query string is stripped for `htu` per atproto
26
+
## Options
27
+
- `:nonce` — server DPoP nonce (omit when unknown)
28
+
- `:access_token` — adds `ath` (S256 hash) for PDS/resource requests
29
+
- `:jti` — override `jti` (tests)
30
+
- `:iat` — override `iat` (tests)
31
+
Note: atproto currently says **do not** include `iss` on PDS-bound proofs.
32
+
"""
33
+
@spec proof(JOSE.JWK.t(), String.t(), String.t(), keyword()) :: String.t()
34
+
def proof(jwk, method, url, opts \\ []) do
35
+
jti = Keyword.get(opts, :jti, random_b64(20))
36
+
iat = Keyword.get(opts, :iat, System.os_time(:second))
37
+
nonce = Keyword.get(opts, :nonce)
38
+
access_token = Keyword.get(opts, :access_token)
39
+
40
+
{_, public_jwk} = JOSE.JWK.to_public_map(jwk)
41
+
42
+
jws = %{
43
+
"alg" => @algorithm,
44
+
"typ" => @jwt_type,
45
+
"jwk" => public_jwk
46
+
}
47
+
48
+
claims =
49
+
%{
50
+
"jti" => jti,
51
+
"htm" => String.upcase(method),
52
+
"htu" => htu(url),
53
+
"iat" => iat
54
+
}
55
+
|> maybe_put("nonce", nonce)
56
+
|> maybe_put("ath", access_token && access_token_hash(access_token))
57
+
58
+
jwk
59
+
|> JOSE.JWT.sign(jws, claims)
60
+
|> JOSE.JWS.compact()
61
+
|> elem(1)
62
+
end
63
+
64
+
@doc """
65
+
S256 hash of an access token for the `ath` claim (same as PKCE S256)
66
+
"""
67
+
@spec access_token_hash(String.t()) :: String.t()
68
+
def access_token_hash(access_token) when is_binary(access_token) do
69
+
PKCE.challenge(access_token)
70
+
end
71
+
72
+
defp htu(url) do
73
+
url
74
+
|> URI.parse()
75
+
|> Map.put(:query, nil)
76
+
|> URI.to_string()
77
+
end
78
+
79
+
defp maybe_put(map, _key, nil), do: map
80
+
defp maybe_put(map, key, value), do: Map.put(map, key, value)
81
+
82
+
defp random_b64(bytes) do
83
+
bytes
84
+
|> :crypto.strong_rand_bytes()
85
+
|> Base.url_encode64(padding: false)
86
+
end
87
+
end
+29
lib/annot_at/atproto/oauth/pkce.ex
+29
lib/annot_at/atproto/oauth/pkce.ex
···
1
+
defmodule AnnotAt.Atproto.OAuth.PKCE do
2
+
@moduledoc """
3
+
PKCE (RFC 7636) helpers for atproto OAuth.
4
+
"""
5
+
6
+
@verifier_min_bytes 32
7
+
8
+
@doc """
9
+
Generates a PKCE code verifier (base64url-encoded random bytes).
10
+
11
+
Length is 43-128 characters RFC 7636 / atproto OAuth.
12
+
"""
13
+
@spec generate_verifier() :: String.t()
14
+
def generate_verifier do
15
+
@verifier_min_bytes
16
+
|> :crypto.strong_rand_bytes()
17
+
|> Base.url_encode64(padding: false)
18
+
end
19
+
20
+
@doc """
21
+
Derives S256 code challenge from code verifier.
22
+
"""
23
+
@spec challenge(String.t()) :: String.t()
24
+
def challenge(verifier) when is_binary(verifier) do
25
+
result = :crypto.hash(:sha256, verifier)
26
+
27
+
Base.url_encode64(result, padding: false)
28
+
end
29
+
end
+69
test/annot_at/atproto/oauth/client_assertion_test.exs
+69
test/annot_at/atproto/oauth/client_assertion_test.exs
···
1
+
defmodule AnnotAt.Atproto.OAuth.ClientAssertionTest do
2
+
use ExUnit.Case, async: true
3
+
4
+
alias AnnotAt.Atproto.OAuth.ClientAssertion
5
+
6
+
@fixture Path.expand("../../../support/fixtures/atproto/es256_jwk.json", __DIR__)
7
+
8
+
# RFC 7638 SHA-256 thumbprint of the fixture key, computed externally
9
+
@fixture_thumbprint "i7xgge3EZQmET-d57_G53RfYSp6nmGAqIxVRdWARNDc"
10
+
11
+
setup do
12
+
jwk =
13
+
@fixture
14
+
|> File.read!()
15
+
|> Jason.decode!()
16
+
|> JOSE.JWK.from()
17
+
18
+
%{jwk: jwk}
19
+
end
20
+
21
+
test "sign/4 builds a JWT with RFC 7523 claims", %{jwk: jwk} do
22
+
token =
23
+
ClientAssertion.sign(
24
+
jwk,
25
+
"https://annot.at/oauth-client-metadata.json",
26
+
"https://bsky.social",
27
+
jti: "test-jti",
28
+
iat: 1_700_000_000
29
+
)
30
+
31
+
%JOSE.JWT{fields: claims} = JOSE.JWT.peek_payload(token)
32
+
33
+
assert claims["iss"] == "https://annot.at/oauth-client-metadata.json"
34
+
assert claims["sub"] == "https://annot.at/oauth-client-metadata.json"
35
+
assert claims["aud"] == "https://bsky.social"
36
+
assert claims["jti"] == "test-jti"
37
+
assert claims["iat"] == 1_700_000_000
38
+
assert claims["exp"] == 1_700_000_060
39
+
end
40
+
41
+
test "sign/4 sets alg, typ and kid in the protected header", %{jwk: jwk} do
42
+
token = ClientAssertion.sign(jwk, "client-id", "https://bsky.social", jti: "j", iat: 0)
43
+
44
+
[header_b64, _, _] = String.split(token, ".")
45
+
46
+
header =
47
+
header_b64
48
+
|> Base.url_decode64!(padding: false)
49
+
|> Jason.decode!()
50
+
51
+
assert header == %{"alg" => "ES256", "typ" => "JWT", "kid" => @fixture_thumbprint}
52
+
end
53
+
54
+
test "sign/4 signature verifies with the public key", %{jwk: jwk} do
55
+
token = ClientAssertion.sign(jwk, "client-id", "https://bsky.social", jti: "j", iat: 0)
56
+
57
+
public = JOSE.JWK.to_public(jwk)
58
+
assert {true, _, _} = JOSE.JWT.verify_strict(public, ["ES256"], token)
59
+
end
60
+
61
+
test "kid/1 prefers an explicit kid over the thumbprint", %{jwk: jwk} do
62
+
assert ClientAssertion.kid(jwk) == @fixture_thumbprint
63
+
64
+
{_, map} = JOSE.JWK.to_map(jwk)
65
+
with_kid = JOSE.JWK.from(Map.put(map, "kid", "my-key-1"))
66
+
67
+
assert ClientAssertion.kid(with_kid) == "my-key-1"
68
+
end
69
+
end
+70
test/annot_at/atproto/oauth/client_metadata_test.exs
+70
test/annot_at/atproto/oauth/client_metadata_test.exs
···
1
+
defmodule AnnotAt.Atproto.OAuth.ClientMetadataTest do
2
+
use ExUnit.Case, async: true
3
+
4
+
alias AnnotAt.Atproto.OAuth.ClientAssertion
5
+
alias AnnotAt.Atproto.OAuth.ClientMetadata
6
+
7
+
@fixture Path.expand("../../../support/fixtures/atproto/es256_jwk.json", __DIR__)
8
+
9
+
setup do
10
+
jwk =
11
+
@fixture
12
+
|> File.read!()
13
+
|> Jason.decode!()
14
+
|> JOSE.JWK.from()
15
+
16
+
%{jwk: jwk}
17
+
end
18
+
19
+
defp build(jwk, extra \\ []) do
20
+
ClientMetadata.build(
21
+
extra ++
22
+
[
23
+
client_id: "https://annot.at/oauth-client-metadata.json",
24
+
redirect_uris: ["https://annot.at/auth/callback"],
25
+
scope: "atproto",
26
+
jwk: jwk
27
+
]
28
+
)
29
+
end
30
+
31
+
test "build/1 sets the spec-required fields", %{jwk: jwk} do
32
+
metadata = build(jwk)
33
+
34
+
assert metadata["client_id"] == "https://annot.at/oauth-client-metadata.json"
35
+
assert metadata["application_type"] == "web"
36
+
assert metadata["grant_types"] == ["authorization_code", "refresh_token"]
37
+
assert metadata["response_types"] == ["code"]
38
+
assert metadata["redirect_uris"] == ["https://annot.at/auth/callback"]
39
+
assert metadata["scope"] == "atproto"
40
+
assert metadata["dpop_bound_access_tokens"] == true
41
+
assert metadata["token_endpoint_auth_method"] == "private_key_jwt"
42
+
assert metadata["token_endpoint_auth_signing_alg"] == "ES256"
43
+
end
44
+
45
+
test "build/1 publishes only the public key, with the assertion kid", %{jwk: jwk} do
46
+
%{"jwks" => %{"keys" => [key]}} = build(jwk)
47
+
48
+
refute Map.has_key?(key, "d")
49
+
assert key["kty"] == "EC"
50
+
assert key["crv"] == "P-256"
51
+
assert key["kid"] == ClientAssertion.kid(jwk)
52
+
end
53
+
54
+
test "build/1 includes optional fields only when given", %{jwk: jwk} do
55
+
metadata = build(jwk, client_name: "AnnotAt", client_uri: "https://annot.at")
56
+
57
+
assert metadata["client_name"] == "AnnotAt"
58
+
assert metadata["client_uri"] == "https://annot.at"
59
+
60
+
bare = build(jwk)
61
+
refute Map.has_key?(bare, "client_name")
62
+
refute Map.has_key?(bare, "client_uri")
63
+
end
64
+
65
+
test "build/1 raises when scope is missing atproto", %{jwk: jwk} do
66
+
assert_raise ArgumentError, fn ->
67
+
build(jwk, scope: "repo:site.standard.*")
68
+
end
69
+
end
70
+
end
+61
test/annot_at/atproto/oauth/dpop_test.exs
+61
test/annot_at/atproto/oauth/dpop_test.exs
···
1
+
defmodule AnnotAt.Atproto.OAuth.DPoPTest do
2
+
use ExUnit.Case, async: true
3
+
4
+
alias AnnotAt.Atproto.OAuth.DPoP
5
+
6
+
@fixture Path.expand("../../../support/fixtures/atproto/es256_jwk.json", __DIR__)
7
+
8
+
setup do
9
+
jwk =
10
+
@fixture
11
+
|> File.read!()
12
+
|> Jason.decode!()
13
+
|> JOSE.JWK.from()
14
+
15
+
%{jwk: jwk}
16
+
end
17
+
18
+
test "generate_key/0 returns an EC key" do
19
+
jwk = DPoP.generate_key()
20
+
{_, map} = JOSE.JWK.to_map(jwk)
21
+
assert map["kty"] == "EC"
22
+
assert map["crv"] == "P-256"
23
+
end
24
+
25
+
test "proof/4 builds a dpop+jwt with required claims", %{jwk: jwk} do
26
+
token =
27
+
DPoP.proof(jwk, "POST", "https://bsky.social/oauth/par?foo=bar",
28
+
jti: "test-jti",
29
+
iat: 1_700_000_000
30
+
)
31
+
32
+
%JOSE.JWT{fields: claims} = JOSE.JWT.peek_payload(token)
33
+
34
+
assert claims["jti"] == "test-jti"
35
+
assert claims["htm"] == "POST"
36
+
assert claims["htu"] == "https://bsky.social/oauth/par"
37
+
assert claims["iat"] == 1_700_000_000
38
+
refute Map.has_key?(claims, "nonce")
39
+
refute Map.has_key?(claims, "ath")
40
+
end
41
+
42
+
test "proof/4 includes nonce when provided", %{jwk: jwk} do
43
+
token =
44
+
DPoP.proof(jwk, "POST", "https://bsky.social/oauth/token", nonce: "n-1", jti: "j", iat: 0)
45
+
46
+
%JOSE.JWT{fields: claims} = JOSE.JWT.peek_payload(token)
47
+
assert claims["nonce"] == "n-1"
48
+
end
49
+
50
+
test "proof/4 includes ath when access_token provided", %{jwk: jwk} do
51
+
token =
52
+
DPoP.proof(jwk, "GET", "https://pds.example/xrpc/app.bsky.actor.getProfile",
53
+
access_token: "my-token",
54
+
jti: "j",
55
+
iat: 0
56
+
)
57
+
58
+
%JOSE.JWT{fields: claims} = JOSE.JWT.peek_payload(token)
59
+
assert claims["ath"] == DPoP.access_token_hash("my-token")
60
+
end
61
+
end
+25
test/annot_at/atproto/oauth/pkce_test.exs
+25
test/annot_at/atproto/oauth/pkce_test.exs
···
1
+
defmodule AnnotAt.Atproto.OAuth.PkceTest do
2
+
use ExUnit.Case, async: true
3
+
4
+
alias AnnotAt.Atproto.OAuth.PKCE
5
+
6
+
# RFC 7636 appendix B
7
+
@verifier "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
8
+
@challenge "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
9
+
10
+
test "challenge/1 S256 matches RFC 7636 appendix B" do
11
+
assert PKCE.challenge(@verifier) == @challenge
12
+
end
13
+
14
+
test "generate_verifier/0 returns base64url string of valid length" do
15
+
verifier = PKCE.generate_verifier()
16
+
17
+
assert String.match?(verifier, ~r/^[A-Za-z0-9_-]+$/)
18
+
assert byte_size(verifier) >= 43
19
+
assert byte_size(verifier) <= 128
20
+
end
21
+
22
+
test "challenge/1 is deterministic for the same verifier" do
23
+
assert PKCE.challenge(@verifier) == PKCE.challenge(@verifier)
24
+
end
25
+
end