Something went wrong. Try again.
Cooperative email for PDS operators
Something went wrong. Try again.
18 kB · 645 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646// SPDX-License-Identifier: AGPL-3.0-or-later
package server
import ( "bytes" "context" "encoding/json" "net/http" "net/http/httptest" "strings" "testing" "time"
"github.com/fxamacker/cbor/v2" "github.com/gorilla/websocket"
"comail/internal/store")
func testServer(t *testing.T) (*Server, *store.Store) { t.Helper() // Use temp file instead of :memory: because database/sql connection // pooling gives each connection its own :memory: DB, causing // "no such table" errors under concurrent access / -race. dbPath := t.TempDir() + "/test.sqlite" s, err := store.New(dbPath) if err != nil { t.Fatal(err) } t.Cleanup(func() { s.Close() }) srv := New(s, "did:key:zTestLabeler") return srv, s}
func insertTestLabel(t *testing.T, s *store.Store, uri, val string) int64 { t.Helper() seq, err := s.InsertLabel(context.Background(), &store.Label{ Src: "did:key:zTestLabeler", URI: uri, Val: val, Cts: time.Now().UTC().Format(time.RFC3339), Neg: false, Sig: []byte("testsig"), RawCBOR: []byte("testraw"), }) if err != nil { t.Fatal(err) } return seq}
func TestQueryLabels(t *testing.T) { srv, s := testServer(t) insertTestLabel(t, s, "did:plc:member1", "verified-mail-operator") insertTestLabel(t, s, "did:plc:member2", "relay-member")
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
resp, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:member1") if err != nil { t.Fatal(err) } defer resp.Body.Close()
if resp.StatusCode != 200 { t.Fatalf("status = %d, want 200", resp.StatusCode) }
var result queryLabelsResponse if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { t.Fatal(err) }
if len(result.Labels) != 1 { t.Fatalf("got %d labels, want 1", len(result.Labels)) } if result.Labels[0].Val != "verified-mail-operator" { t.Errorf("val = %q", result.Labels[0].Val) } if result.Labels[0].Sig == nil { t.Error("sig should be present") }}
func TestQueryLabelsNoPattern(t *testing.T) { srv, _ := testServer(t) ts := httptest.NewServer(srv.Handler()) defer ts.Close()
resp, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels") if err != nil { t.Fatal(err) } defer resp.Body.Close()
if resp.StatusCode != http.StatusBadRequest { t.Errorf("status = %d, want 400", resp.StatusCode) }}
func TestQueryLabelsPagination(t *testing.T) { srv, s := testServer(t) for i := 0; i < 5; i++ { insertTestLabel(t, s, "did:plc:bulk", "verified-mail-operator") }
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
resp, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:bulk&limit=2") if err != nil { t.Fatal(err) } defer resp.Body.Close()
var result queryLabelsResponse if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { t.Fatal(err) }
if len(result.Labels) != 2 { t.Fatalf("got %d labels, want 2", len(result.Labels)) } if result.Cursor == "" { t.Error("expected cursor for pagination") }
// Fetch next page resp2, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:bulk&limit=2&cursor=" + result.Cursor) if err != nil { t.Fatal(err) } defer resp2.Body.Close()
var result2 queryLabelsResponse if err := json.NewDecoder(resp2.Body).Decode(&result2); err != nil { t.Fatal(err) }
if len(result2.Labels) != 2 { t.Fatalf("page 2: got %d labels, want 2", len(result2.Labels)) }}
func TestQueryLabelsLimitCappedAt250(t *testing.T) { srv, _ := testServer(t) ts := httptest.NewServer(srv.Handler()) defer ts.Close()
// Request limit=999, should be capped to 250 (no error) resp, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:test&limit=999") if err != nil { t.Fatal(err) } defer resp.Body.Close()
if resp.StatusCode != 200 { t.Errorf("limit=999 status = %d, want 200 (capped, not rejected)", resp.StatusCode) }}
func TestQueryLabelsInvalidLimit(t *testing.T) { srv, _ := testServer(t) ts := httptest.NewServer(srv.Handler()) defer ts.Close()
// limit=0 should be rejected resp, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:test&limit=0") if err != nil { t.Fatal(err) } defer resp.Body.Close() if resp.StatusCode != http.StatusBadRequest { t.Errorf("limit=0 status = %d, want 400", resp.StatusCode) }
// limit=-1 should be rejected resp2, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:test&limit=-1") if err != nil { t.Fatal(err) } defer resp2.Body.Close() if resp2.StatusCode != http.StatusBadRequest { t.Errorf("limit=-1 status = %d, want 400", resp2.StatusCode) }
// limit=abc should be rejected resp3, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:test&limit=abc") if err != nil { t.Fatal(err) } defer resp3.Body.Close() if resp3.StatusCode != http.StatusBadRequest { t.Errorf("limit=abc status = %d, want 400", resp3.StatusCode) }}
func TestQueryLabelsWildcard(t *testing.T) { srv, s := testServer(t) insertTestLabel(t, s, "did:plc:member1", "verified-mail-operator") insertTestLabel(t, s, "did:plc:member2", "relay-member")
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
resp, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:*") if err != nil { t.Fatal(err) } defer resp.Body.Close()
var result queryLabelsResponse if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { t.Fatal(err) }
if len(result.Labels) != 2 { t.Fatalf("wildcard: got %d labels, want 2", len(result.Labels)) }}
// decodeCBORFrame decodes a binary XRPC subscription frame into header + body.func decodeCBORFrame(t *testing.T, data []byte) (frameHeader, labelsBody) { t.Helper() dec := cbor.NewDecoder(bytes.NewReader(data))
var header frameHeader if err := dec.Decode(&header); err != nil { t.Fatalf("decode frame header: %v", err) }
var body labelsBody if err := dec.Decode(&body); err != nil { t.Fatalf("decode frame body: %v", err) } return header, body}
func TestSubscribeLabelsBackfill(t *testing.T) { srv, s := testServer(t) insertTestLabel(t, s, "did:plc:member1", "verified-mail-operator") insertTestLabel(t, s, "did:plc:member2", "relay-member")
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/xrpc/com.atproto.label.subscribeLabels?cursor=0" conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) if err != nil { t.Fatalf("dial: %v", err) } defer conn.Close()
// Should receive 2 binary CBOR-framed messages (backfill) for i := 0; i < 2; i++ { conn.SetReadDeadline(time.Now().Add(2 * time.Second)) msgType, msg, err := conn.ReadMessage() if err != nil { t.Fatalf("read message %d: %v", i, err) } if msgType != websocket.BinaryMessage { t.Errorf("message %d: type = %d, want BinaryMessage", i, msgType) }
header, body := decodeCBORFrame(t, msg) if header.Op != 1 { t.Errorf("message %d: op = %d, want 1", i, header.Op) } if header.T != "#labels" { t.Errorf("message %d: t = %q, want #labels", i, header.T) } if body.Seq < 1 { t.Errorf("message %d: seq = %d, want >= 1", i, body.Seq) } if len(body.Labels) != 1 { t.Errorf("message %d: got %d labels, want 1", i, len(body.Labels)) } }}
func TestGetVerificationStatus(t *testing.T) { srv, s := testServer(t) ctx := context.Background()
const testDID = "did:plc:mmmm2222nnnn3333oooo4444" const unknownDID = "did:plc:zzzz2222yyyy3333xxxx4444"
// Insert an attestation if err := s.UpsertAttestation(ctx, &store.Attestation{ DID: testDID, RKey: "example.com", Domain: "example.com", DKIMSelectors: []string{"default", "mail"}, RelayMember: true, Verified: true, CreatedAt: time.Now().UTC(), }); err != nil { t.Fatal(err) } insertTestLabel(t, s, testDID, "verified-mail-operator")
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
// Query with DID resp, err := http.Get(ts.URL + "/xrpc/email.atmos.getVerificationStatus?did=" + testDID) if err != nil { t.Fatal(err) } defer resp.Body.Close()
if resp.StatusCode != 200 { t.Fatalf("status = %d, want 200", resp.StatusCode) }
var result verificationStatusResponse if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { t.Fatal(err) }
if result.DID != testDID { t.Errorf("DID = %q", result.DID) } if len(result.Attestations) != 1 { t.Fatalf("got %d attestations, want 1", len(result.Attestations)) } if !result.Attestations[0].Verified { t.Error("attestation should be verified") } if result.Attestations[0].Domain != "example.com" { t.Errorf("domain = %q", result.Attestations[0].Domain) } if len(result.ActiveLabels) != 1 { t.Fatalf("got %d active labels, want 1", len(result.ActiveLabels)) } if result.ActiveLabels[0] != "verified-mail-operator" { t.Errorf("active label = %q", result.ActiveLabels[0]) }
// Query without DID should fail resp2, err := http.Get(ts.URL + "/xrpc/email.atmos.getVerificationStatus") if err != nil { t.Fatal(err) } defer resp2.Body.Close() if resp2.StatusCode != http.StatusBadRequest { t.Errorf("no-did status = %d, want 400", resp2.StatusCode) }
// Query unknown DID should return empty arrays resp3, err := http.Get(ts.URL + "/xrpc/email.atmos.getVerificationStatus?did=" + unknownDID) if err != nil { t.Fatal(err) } defer resp3.Body.Close()
var result3 verificationStatusResponse if err := json.NewDecoder(resp3.Body).Decode(&result3); err != nil { t.Fatal(err) } if len(result3.Attestations) != 0 { t.Errorf("unknown DID: got %d attestations, want 0", len(result3.Attestations)) } if len(result3.ActiveLabels) != 0 { t.Errorf("unknown DID: got %d labels, want 0", len(result3.ActiveLabels)) }}
func TestQueryLabelsSourcesFilter(t *testing.T) { srv, s := testServer(t) ctx := context.Background()
// Insert labels from two different sources s.InsertLabel(ctx, &store.Label{ Src: "did:key:zLabelerA", URI: "did:plc:member1", Val: "verified-mail-operator", Cts: "2026-01-01T00:00:00Z", Sig: []byte("sig1"), RawCBOR: []byte("raw1"), }) s.InsertLabel(ctx, &store.Label{ Src: "did:key:zLabelerB", URI: "did:plc:member1", Val: "relay-member", Cts: "2026-01-01T00:00:00Z", Sig: []byte("sig2"), RawCBOR: []byte("raw2"), })
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
// Filter by source A — should only get 1 label resp, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:member1&sources=did:key:zLabelerA") if err != nil { t.Fatal(err) } defer resp.Body.Close()
var result queryLabelsResponse if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { t.Fatal(err) } if len(result.Labels) != 1 { t.Fatalf("sources filter: got %d labels, want 1", len(result.Labels)) } if result.Labels[0].Src != "did:key:zLabelerA" { t.Errorf("src = %q, want did:key:zLabelerA", result.Labels[0].Src) }
// No sources filter — should get both resp2, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:member1") if err != nil { t.Fatal(err) } defer resp2.Body.Close()
var result2 queryLabelsResponse if err := json.NewDecoder(resp2.Body).Decode(&result2); err != nil { t.Fatal(err) } if len(result2.Labels) != 2 { t.Fatalf("no sources filter: got %d labels, want 2", len(result2.Labels)) }}
func TestQueryLabelsMultiPattern(t *testing.T) { srv, s := testServer(t) // Insert labels for two different DIDs insertTestLabel(t, s, "did:plc:aaa", "verified-mail-operator") insertTestLabel(t, s, "did:plc:bbb", "relay-member") insertTestLabel(t, s, "did:plc:aaa", "relay-member")
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
// Query with two exact patterns — should return all 3 labels in seq order resp, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:aaa&uriPatterns=did:plc:bbb") if err != nil { t.Fatal(err) } defer resp.Body.Close()
var result queryLabelsResponse if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { t.Fatal(err) }
if len(result.Labels) != 3 { t.Fatalf("multi-pattern: got %d labels, want 3", len(result.Labels)) }
// Paginate with limit=2 — cursor should work across patterns resp2, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:aaa&uriPatterns=did:plc:bbb&limit=2") if err != nil { t.Fatal(err) } defer resp2.Body.Close()
var page1 queryLabelsResponse if err := json.NewDecoder(resp2.Body).Decode(&page1); err != nil { t.Fatal(err) } if len(page1.Labels) != 2 { t.Fatalf("page1: got %d labels, want 2", len(page1.Labels)) } if page1.Cursor == "" { t.Fatal("page1: expected cursor") }
resp3, err := http.Get(ts.URL + "/xrpc/com.atproto.label.queryLabels?uriPatterns=did:plc:aaa&uriPatterns=did:plc:bbb&limit=2&cursor=" + page1.Cursor) if err != nil { t.Fatal(err) } defer resp3.Body.Close()
var page2 queryLabelsResponse if err := json.NewDecoder(resp3.Body).Decode(&page2); err != nil { t.Fatal(err) } if len(page2.Labels) != 1 { t.Fatalf("page2: got %d labels, want 1", len(page2.Labels)) }}
// internal ref: getVerificationStatus surfaces the reconciliation// lifecycle (status / failure_kind / reason / attempt_count / next_check_at)// so the relay account page can explain why a domain isn't verified yet.func TestGetVerificationStatusSurfacesLifecycle(t *testing.T) { srv, s := testServer(t) ctx := context.Background()
const did = "did:plc:lifelifelifelifelifelife" next := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC) if err := s.UpsertAttestation(ctx, &store.Attestation{ DID: did, RKey: "stuck.example", Domain: "stuck.example", RelayMember: true, Verified: false, Status: "failed", FailureKind: "transient", LastError: "DKIM: no record for selector atmos20260527r", AttemptCount: 3, NextCheckAt: next, CreatedAt: time.Now().UTC(), }); err != nil { t.Fatal(err) }
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
resp, err := http.Get(ts.URL + "/xrpc/email.atmos.getVerificationStatus?did=" + did) if err != nil { t.Fatal(err) } defer resp.Body.Close()
var result verificationStatusResponse if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { t.Fatal(err) } if len(result.Attestations) != 1 { t.Fatalf("got %d attestations, want 1", len(result.Attestations)) } a := result.Attestations[0] if a.Status != "failed" { t.Errorf("status = %q, want failed", a.Status) } if a.FailureKind != "transient" { t.Errorf("failureKind = %q, want transient", a.FailureKind) } if a.Reason != "DKIM: no record for selector atmos20260527r" { t.Errorf("reason = %q", a.Reason) } if a.AttemptCount != 3 { t.Errorf("attemptCount = %d, want 3", a.AttemptCount) } if a.NextCheckAt != next.Format(time.RFC3339) { t.Errorf("nextCheckAt = %q, want %q", a.NextCheckAt, next.Format(time.RFC3339)) }}
func TestGetVerificationStatusRejectsInvalidDID(t *testing.T) { srv, _ := testServer(t) ts := httptest.NewServer(srv.Handler()) defer ts.Close()
resp, err := http.Get(ts.URL + "/xrpc/email.atmos.getVerificationStatus?did=not-a-did") if err != nil { t.Fatal(err) } defer resp.Body.Close() if resp.StatusCode != http.StatusBadRequest { t.Errorf("invalid DID status = %d, want 400", resp.StatusCode) }}
func TestHealthz(t *testing.T) { srv, _ := testServer(t) ts := httptest.NewServer(srv.Handler()) defer ts.Close()
resp, err := http.Get(ts.URL + "/healthz") if err != nil { t.Fatal(err) } defer resp.Body.Close() if resp.StatusCode != 200 { t.Errorf("healthz status = %d, want 200", resp.StatusCode) }}
func TestSubscribeLabelsConnectionLimit(t *testing.T) { srv, _ := testServer(t) ts := httptest.NewServer(srv.Handler()) defer ts.Close()
wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/xrpc/com.atproto.label.subscribeLabels"
// Open maxWebSocketConns connections var conns []*websocket.Conn for i := 0; i < maxWebSocketConns; i++ { conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) if err != nil { t.Fatalf("dial %d: %v", i, err) } conns = append(conns, conn) } defer func() { for _, c := range conns { c.Close() } }()
// Next connection should be rejected _, resp, err := websocket.DefaultDialer.Dial(wsURL, nil) if err == nil { t.Error("expected connection to be rejected at limit") } if resp != nil && resp.StatusCode != http.StatusServiceUnavailable { t.Errorf("status = %d, want 503", resp.StatusCode) }}
func TestShutdownWebSocketsSendsClose(t *testing.T) { srv, s := testServer(t) ctx := context.Background()
insertTestLabel(t, s, "did:plc:member1", "verified-mail-operator")
ts := httptest.NewServer(srv.Handler()) defer ts.Close()
wsURL := "ws" + strings.TrimPrefix(ts.URL, "http") + "/xrpc/com.atproto.label.subscribeLabels?cursor=0"
// Read the backfilled label first so connection is established conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) if err != nil { t.Fatal(err) } defer conn.Close()
conn.SetReadDeadline(time.Now().Add(2 * time.Second)) _, _, err = conn.ReadMessage() if err != nil { t.Fatalf("read backfill: %v", err) }
_ = ctx // keep consistent with other tests
// Shut down WebSockets — should send close frame srv.ShutdownWebSockets()
// Next read should get a close message or error conn.SetReadDeadline(time.Now().Add(2 * time.Second)) _, _, err = conn.ReadMessage() if err == nil { t.Error("expected error after shutdown, got nil") } if !websocket.IsCloseError(err, websocket.CloseGoingAway) { // Connection may also just be closed — both are acceptable if !strings.Contains(err.Error(), "use of closed") && !strings.Contains(err.Error(), "connection reset") && !websocket.IsUnexpectedCloseError(err) { t.Logf("got error type: %v (acceptable)", err) } }}