Something went wrong. Try again.
Cooperative email for PDS operators
Something went wrong. Try again.
15 kB · 479 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480// SPDX-License-Identifier: AGPL-3.0-or-later
package main
import ( "context" "encoding/json" "net/http" "net/http/httptest" "testing"
"comail/internal/jetstream")
func TestResolvePDS(t *testing.T) { plc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/did:plc:test2345test2345test2345" { http.NotFound(w, r) return } json.NewEncoder(w).Encode(map[string]any{ "id": "did:plc:test2345test2345test2345", "service": []map[string]string{ { "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": "https://pds.example.com", }, }, }) })) defer plc.Close()
ctx := context.Background() client := plc.Client()
pds, err := resolvePDS(ctx, client, plc.URL, "did:plc:test2345test2345test2345") if err != nil { t.Fatalf("resolvePDS: %v", err) } if pds != "https://pds.example.com" { t.Errorf("pds = %q, want https://pds.example.com", pds) }}
func TestResolvePDSNoPDSService(t *testing.T) { plc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]any{ "id": "did:plc:test2345test2345test2345", "service": []map[string]string{}, }) })) defer plc.Close()
ctx := context.Background() _, err := resolvePDS(ctx, plc.Client(), plc.URL, "did:plc:test2345test2345test2345") if err == nil { t.Fatal("expected error for DID with no PDS service") }}
func TestResolvePDS404(t *testing.T) { plc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) })) defer plc.Close()
ctx := context.Background() _, err := resolvePDS(ctx, plc.Client(), plc.URL, "did:plc:doesnotexist2345does2345") if err == nil { t.Fatal("expected error for unknown DID") }}
func TestFetchAttestations(t *testing.T) { pds := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/xrpc/com.atproto.repo.listRecords" { http.NotFound(w, r) return } repo := r.URL.Query().Get("repo") collection := r.URL.Query().Get("collection") if repo != "did:plc:test2345test2345test2345" || collection != "email.atmos.attestation" { http.Error(w, "bad params", http.StatusBadRequest) return }
json.NewEncoder(w).Encode(map[string]any{ "records": []map[string]any{ { "uri": "at://did:plc:test2345test2345test2345/email.atmos.attestation/3minjnprupb2q", "cid": "bafyreiabc123", "value": map[string]any{ "$type": "email.atmos.attestation", "domain": "example.com", "dkimSelectors": []string{"protonmail", "protonmail2"}, "relayMember": false, "createdAt": "2026-04-04T00:00:00Z", }, }, { "uri": "at://did:plc:test2345test2345test2345/email.atmos.attestation/3minjnprupb2r", "cid": "bafyreiabc456", "value": map[string]any{ "$type": "email.atmos.attestation", "domain": "other.org", "dkimSelectors": []string{"default"}, "relayMember": true, "createdAt": "2026-04-05T00:00:00Z", }, }, }, }) })) defer pds.Close()
ctx := context.Background() atts, err := fetchAttestations(ctx, pds.Client(), pds.URL, "did:plc:test2345test2345test2345") if err != nil { t.Fatalf("fetchAttestations: %v", err) } if len(atts) != 2 { t.Fatalf("got %d attestations, want 2", len(atts)) }
// First record if atts[0].DID != "did:plc:test2345test2345test2345" { t.Errorf("att[0].DID = %q", atts[0].DID) } if atts[0].Domain != "example.com" { t.Errorf("att[0].Domain = %q, want example.com", atts[0].Domain) } if atts[0].RKey != "3minjnprupb2q" { t.Errorf("att[0].RKey = %q, want 3minjnprupb2q", atts[0].RKey) } if len(atts[0].DKIMSelectors) != 2 { t.Errorf("att[0].DKIMSelectors = %v, want 2 selectors", atts[0].DKIMSelectors) } if atts[0].Operation != "create" { t.Errorf("att[0].Operation = %q, want create", atts[0].Operation) } if atts[0].RelayMember { t.Error("att[0].RelayMember should be false") }
// Second record if atts[1].Domain != "other.org" { t.Errorf("att[1].Domain = %q, want other.org", atts[1].Domain) } if !atts[1].RelayMember { t.Error("att[1].RelayMember should be true") }}
func TestFetchAttestationsEmpty(t *testing.T) { pds := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]any{ "records": []any{}, }) })) defer pds.Close()
ctx := context.Background() atts, err := fetchAttestations(ctx, pds.Client(), pds.URL, "did:plc:test2345test2345test2345") if err != nil { t.Fatalf("fetchAttestations: %v", err) } if len(atts) != 0 { t.Errorf("got %d attestations, want 0", len(atts)) }}
func TestFetchAttestationsSkipsMalformed(t *testing.T) { pds := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]any{ "records": []map[string]any{ { "uri": "at://did:plc:test2345test2345test2345/email.atmos.attestation/good", "cid": "bafyreiabc123", "value": map[string]any{"domain": "good.com", "dkimSelectors": []string{"s1"}, "createdAt": "2026-04-04T00:00:00Z"}, }, { "uri": "at://did:plc:test2345test2345test2345/email.atmos.attestation/bad", "cid": "bafyreiabc456", "value": "not-a-json-object", // malformed }, }, }) })) defer pds.Close()
ctx := context.Background() atts, err := fetchAttestations(ctx, pds.Client(), pds.URL, "did:plc:test2345test2345test2345") if err != nil { t.Fatalf("fetchAttestations: %v", err) } if len(atts) != 1 { t.Fatalf("got %d attestations, want 1 (malformed skipped)", len(atts)) } if atts[0].Domain != "good.com" { t.Errorf("att[0].Domain = %q, want good.com", atts[0].Domain) }}
func TestExtractRKey(t *testing.T) { tests := []struct { uri string want string }{ {"at://did:plc:abc/email.atmos.attestation/3minjnprupb2q", "3minjnprupb2q"}, {"at://did:plc:abc/email.atmos.attestation/rkey123", "rkey123"}, {"", ""}, } for _, tt := range tests { got := extractRKey(tt.uri) if got != tt.want { t.Errorf("extractRKey(%q) = %q, want %q", tt.uri, got, tt.want) } }}
func TestValidatePDSURL(t *testing.T) { tests := []struct { url string wantErr bool }{ {"https://pds.example.com", false}, {"https://bsky.social", false}, {"http://pds.example.com", true}, // not HTTPS {"http://localhost:8081", true}, // not HTTPS {"http://192.0.2.1:4646", true}, // not HTTPS (RFC 5737 TEST-NET) {"ftp://pds.example.com", true}, // wrong scheme {"", true}, // empty {"://no-scheme", true}, // malformed } for _, tt := range tests { err := validatePDSURL(tt.url) if (err != nil) != tt.wantErr { t.Errorf("validatePDSURL(%q) err=%v, wantErr=%v", tt.url, err, tt.wantErr) } }}
// newTestBootstrapper creates a bootstrapper pointing at mock PLC+PDS servers.// Returns the bootstrapper and cleanup function.func newTestBootstrapper(t *testing.T, adminToken string, handler func(jetstream.ReceivedAttestation) error) (*bootstrapper, *httptest.Server) { t.Helper()
// Mock PDS pds := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]any{ "records": []map[string]any{ { "uri": "at://did:plc:test2345test2345test2345/email.atmos.attestation/rk1", "cid": "bafyreiabc123", "value": map[string]any{ "domain": "test.com", "dkimSelectors": []string{"s1"}, "createdAt": "2026-04-04T00:00:00Z", }, }, }, }) })) t.Cleanup(pds.Close)
// Mock PLC directory that points to our TLS PDS plc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]any{ "service": []map[string]string{ {"id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": pds.URL}, }, }) })) t.Cleanup(plc.Close)
b := &bootstrapper{ client: pds.Client(), // TLS client that trusts the test CA pdsClient: pds.Client(), // tests dial localhost — share the trust; // the SSRF guard is exercised in ssrf_test.go (loopback IPs would // be rejected by a production newPDSClient, but here we WANT to // reach the httptest server, so we bypass it. This is a test // affordance; production constructs via newPDSClient). plcURL: plc.URL, adminToken: adminToken, onAtt: handler, } return b, pds}
func TestBootstrapHandlerIntegration(t *testing.T) { var received []jetstream.ReceivedAttestation b, _ := newTestBootstrapper(t, "", func(att jetstream.ReceivedAttestation) error { received = append(received, att) return nil })
req := httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) w := httptest.NewRecorder() b.ServeHTTP(w, req)
if w.Code != http.StatusOK { t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String()) }
if len(received) != 1 { t.Fatalf("handler received %d attestations, want 1", len(received)) } if received[0].Domain != "test.com" { t.Errorf("received domain = %q, want test.com", received[0].Domain) }
var resp bootstrapResponse if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { t.Fatalf("decode response: %v", err) } if resp.Processed != 1 { t.Errorf("response.processed = %d, want 1", resp.Processed) }}
func TestBootstrapHandlerMissingDID(t *testing.T) { b := &bootstrapper{client: http.DefaultClient, pdsClient: http.DefaultClient, plcURL: "http://localhost"}
req := httptest.NewRequest("POST", "/admin/bootstrap", nil) w := httptest.NewRecorder() b.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest { t.Errorf("status = %d, want 400", w.Code) }}
func TestBootstrapHandlerWrongMethod(t *testing.T) { b := &bootstrapper{client: http.DefaultClient, pdsClient: http.DefaultClient, plcURL: "http://localhost"}
req := httptest.NewRequest("GET", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) w := httptest.NewRecorder() b.ServeHTTP(w, req)
if w.Code != http.StatusMethodNotAllowed { t.Errorf("status = %d, want 405", w.Code) }}
func TestBootstrapHandlerAuthRequired(t *testing.T) { b, _ := newTestBootstrapper(t, "secret-token-123", func(att jetstream.ReceivedAttestation) error { return nil })
// No token → 401 req := httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) w := httptest.NewRecorder() b.ServeHTTP(w, req) if w.Code != http.StatusUnauthorized { t.Errorf("no token: status = %d, want 401", w.Code) }
// Wrong token → 401 req = httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) req.Header.Set("Authorization", "Bearer wrong-token") w = httptest.NewRecorder() b.ServeHTTP(w, req) if w.Code != http.StatusUnauthorized { t.Errorf("wrong token: status = %d, want 401", w.Code) }
// Correct token → 200 req = httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) req.Header.Set("Authorization", "Bearer secret-token-123") w = httptest.NewRecorder() b.ServeHTTP(w, req) if w.Code != http.StatusOK { t.Errorf("correct token: status = %d, want 200; body: %s", w.Code, w.Body.String()) }}
func TestBootstrapHandlerRateLimit(t *testing.T) { b, _ := newTestBootstrapper(t, "", func(att jetstream.ReceivedAttestation) error { return nil })
// First call succeeds req := httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) w := httptest.NewRecorder() b.ServeHTTP(w, req) if w.Code != http.StatusOK { t.Fatalf("first call: status = %d, want 200", w.Code) }
// Second call to same DID within cooldown → 429 req = httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) w = httptest.NewRecorder() b.ServeHTTP(w, req) if w.Code != http.StatusTooManyRequests { t.Errorf("repeat call: status = %d, want 429", w.Code) }
// Different DID should still work req = httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:aaaa2345bbbb2345cccc2345", nil) w = httptest.NewRecorder() b.ServeHTTP(w, req) // May fail on PLC resolve since the mock only has one DID, but should NOT be 429 if w.Code == http.StatusTooManyRequests { t.Error("different DID should not be rate limited") }}
func TestBootstrapHandlerSSRFBlocksHTTP(t *testing.T) { // PLC that returns a plain HTTP (not HTTPS) PDS endpoint plc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]any{ "service": []map[string]string{ {"id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": "http://192.0.2.1:4646"}, }, }) })) defer plc.Close()
b := &bootstrapper{ client: http.DefaultClient, pdsClient: http.DefaultClient, plcURL: plc.URL, onAtt: func(att jetstream.ReceivedAttestation) error { return nil }, }
req := httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) w := httptest.NewRecorder() b.ServeHTTP(w, req)
if w.Code != http.StatusBadGateway { t.Errorf("SSRF HTTP: status = %d, want 502", w.Code) }}
func TestBootstrapErrorsDoNotLeakInternals(t *testing.T) { plc := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]any{ "service": []map[string]string{ {"id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": "https://unreachable.invalid"}, }, }) })) defer plc.Close()
b := &bootstrapper{ client: http.DefaultClient, pdsClient: http.DefaultClient, plcURL: plc.URL, onAtt: func(att jetstream.ReceivedAttestation) error { return nil }, }
req := httptest.NewRequest("POST", "/admin/bootstrap?did=did:plc:test2345test2345test2345", nil) w := httptest.NewRecorder() b.ServeHTTP(w, req)
body := w.Body.String() // Error response should NOT contain the internal PDS URL if contains(body, "unreachable.invalid") { t.Errorf("error response leaks PDS URL: %s", body) } // Should give a generic error if !contains(body, "failed to fetch attestations from PDS") { t.Errorf("expected generic error message, got: %s", body) }}
func contains(s, substr string) bool { return len(s) >= len(substr) && searchString(s, substr)}
func searchString(s, substr string) bool { for i := 0; i <= len(s)-len(substr); i++ { if s[i:i+len(substr)] == substr { return true } } return false}