Something went wrong. Try again.
Go boilerplate library for building atproto apps
atproto go
Something went wrong. Try again.
atp public.go
12 kB · 423 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424package atp
import ( "context" "encoding/json" "errors" "fmt" "net" "net/http" "net/url" "slices" "strings" "sync" "time")
// PublicAPIBase is the Bluesky public API endpoint used for profile and handle lookups.// This is a variable so tests can override it.var PublicAPIBase = "https://public.api.bsky.app"
// PLCDirectory is used to resolve did:plc identifiers to DID documents.// This is a variable so tests can override it.var PLCDirectory = "https://plc.directory"
// ErrSSRFBlocked is returned when a request is blocked due to a private/internal destination.var ErrSSRFBlocked = errors.New("request blocked: potential SSRF detected")
// ResolveHandle is a convenience that resolves an AT Protocol handle to a DID// without needing to construct a PublicClient.func ResolveHandle(ctx context.Context, handle string) (string, error) { return NewPublicClient().ResolveHandle(ctx, handle)}
const resolverCacheTTL = time.Hour
type cachedValue struct { value string expiry time.Time}
// PublicClient provides unauthenticated read access to public AT Protocol APIs.// Use this to resolve handles, look up profiles, and read public records without// requiring an OAuth session.//// Handle→DID and DID→PDS lookups are cached with a 1-hour TTL. Call// InvalidateHandle or InvalidateDID when identity changes are detected.type PublicClient struct { httpClient *http.Client
pdsMu sync.RWMutex pdsCache map[string]cachedValue // DID → PDS URL
handleMu sync.RWMutex handleCache map[string]cachedValue // handle → DID}
// NewPublicClient creates a PublicClient with a 30-second timeout.// To add OTel instrumentation, use NewPublicClientWithHTTP and pass an// otelhttp-wrapped transport.func NewPublicClient() *PublicClient { return NewPublicClientWithHTTP(&http.Client{ Timeout: 30 * time.Second, })}
// NewPublicClientWithHTTP creates a PublicClient using the provided http.Client.// This lets callers inject custom transports (e.g. with OTel or rate limiting).func NewPublicClientWithHTTP(hc *http.Client) *PublicClient { return &PublicClient{ httpClient: hc, pdsCache: make(map[string]cachedValue), handleCache: make(map[string]cachedValue), }}
// ResolveHandle resolves an AT Protocol handle to a DID string.// Accepts both Unicode and ASCII (punycode) IDN forms; input is normalized// to ASCII before lookup so cache hits are independent of the form supplied.func (c *PublicClient) ResolveHandle(ctx context.Context, handle string) (string, error) { handle = NormalizeHandle(handle)
c.handleMu.RLock() if v, ok := c.handleCache[handle]; ok && time.Now().Before(v.expiry) { c.handleMu.RUnlock() return v.value, nil } c.handleMu.RUnlock()
reqURL := fmt.Sprintf("%s/xrpc/com.atproto.identity.resolveHandle?handle=%s", PublicAPIBase, url.QueryEscape(handle))
req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) if err != nil { return "", fmt.Errorf("build request: %w", err) }
resp, err := c.httpClient.Do(req) if err != nil { return "", fmt.Errorf("resolve handle: %w", err) } defer resp.Body.Close()
if resp.StatusCode == http.StatusNotFound { return "", fmt.Errorf("handle not found: %s", handle) } if resp.StatusCode != http.StatusOK { return "", fmt.Errorf("resolve handle: HTTP %d", resp.StatusCode) }
var result struct { DID string `json:"did"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return "", fmt.Errorf("decode response: %w", err) }
c.handleMu.Lock() c.handleCache[handle] = cachedValue{value: result.DID, expiry: time.Now().Add(resolverCacheTTL)} c.handleMu.Unlock()
return result.DID, nil}
// GetPDSEndpoint resolves a DID to the user's PDS base URL.// Results are cached with a 1-hour TTL.func (c *PublicClient) GetPDSEndpoint(ctx context.Context, did string) (string, error) { c.pdsMu.RLock() if v, ok := c.pdsCache[did]; ok && time.Now().Before(v.expiry) { c.pdsMu.RUnlock() return v.value, nil } c.pdsMu.RUnlock()
var pdsEndpoint string
switch { case strings.HasPrefix(did, "did:plc:"): reqURL := fmt.Sprintf("%s/%s", PLCDirectory, did) req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) if err != nil { return "", fmt.Errorf("build request: %w", err) } resp, err := c.httpClient.Do(req) if err != nil { return "", fmt.Errorf("fetch DID document: %w", err) } defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { return "", fmt.Errorf("DID resolution: HTTP %d", resp.StatusCode) }
var didDoc struct { Service []struct { ID string `json:"id"` Type string `json:"type"` ServiceEndpoint string `json:"serviceEndpoint"` } `json:"service"` } if err := json.NewDecoder(resp.Body).Decode(&didDoc); err != nil { return "", fmt.Errorf("decode DID document: %w", err) } for _, svc := range didDoc.Service { if svc.ID == "#atproto_pds" || svc.Type == "AtprotoPersonalDataServer" { pdsEndpoint = svc.ServiceEndpoint break } }
case strings.HasPrefix(did, "did:web:"): domain := strings.TrimPrefix(did, "did:web:") domain = strings.ReplaceAll(domain, "%3A", ":") domain = strings.ReplaceAll(domain, "%2F", "/") if idx := strings.Index(domain, "/"); idx != -1 { domain = domain[:idx] } host := domain if h, _, err := net.SplitHostPort(domain); err == nil { host = h } if err := validateDomain(host); err != nil { return "", err } pdsEndpoint = "https://" + domain }
if pdsEndpoint == "" { return "", fmt.Errorf("could not resolve PDS endpoint for %s", did) }
c.pdsMu.Lock() c.pdsCache[did] = cachedValue{value: pdsEndpoint, expiry: time.Now().Add(resolverCacheTTL)} c.pdsMu.Unlock()
return pdsEndpoint, nil}
// InvalidateHandle removes a handle from the resolver cache so the next// ResolveHandle call refetches from the directory. Call when a firehose// identity event signals that a handle's DID mapping has changed.func (c *PublicClient) InvalidateHandle(handle string) { handle = NormalizeHandle(handle) if handle == "" { return } c.handleMu.Lock() delete(c.handleCache, handle) c.handleMu.Unlock()}
// InvalidateDID drops any cached entries pointing at this DID — both the// PDS endpoint cache and any handle→DID mappings whose resolved DID is the// given one. Used when a DID's repo is gone (account deleted/takendown) or// when a handle has been reassigned away from this DID.func (c *PublicClient) InvalidateDID(did string) { if did == "" { return } c.pdsMu.Lock() delete(c.pdsCache, did) c.pdsMu.Unlock()
c.handleMu.Lock() for h, v := range c.handleCache { if v.value == did { delete(c.handleCache, h) } } c.handleMu.Unlock()}
// PublicProfile is a user's public profile as returned by the Bluesky public API.type PublicProfile struct { DID string `json:"did"` Handle string `json:"handle"` DisplayName *string `json:"displayName,omitempty"` Avatar *string `json:"avatar,omitempty"`}
// GetProfile fetches a user's public profile by DID or handle.func (c *PublicClient) GetProfile(ctx context.Context, actor string) (*PublicProfile, error) { reqURL := fmt.Sprintf("%s/xrpc/app.bsky.actor.getProfile?actor=%s", PublicAPIBase, url.QueryEscape(actor))
req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) if err != nil { return nil, fmt.Errorf("build request: %w", err) }
resp, err := c.httpClient.Do(req) if err != nil { return nil, fmt.Errorf("fetch profile: %w", err) } defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("get profile: HTTP %d", resp.StatusCode) }
var profile PublicProfile if err := json.NewDecoder(resp.Body).Decode(&profile); err != nil { return nil, fmt.Errorf("decode profile: %w", err) } return &profile, nil}
// ListPublicRecordsOpts configures a ListPublicRecords call.// All fields are optional; the zero value yields the lexicon defaults.type ListPublicRecordsOpts struct { Limit int // 0 means server default Cursor string // empty for first page Reverse bool // true returns newest-first}
// ListPublicRecords fetches records from a public collection.// Queries the user's PDS directly, so it works with any collection NSID.func (c *PublicClient) ListPublicRecords(ctx context.Context, did, collection string, opts ListPublicRecordsOpts) ([]Record, string, error) { pdsEndpoint, err := c.GetPDSEndpoint(ctx, did) if err != nil { return nil, "", fmt.Errorf("resolve PDS: %w", err) }
q := url.Values{} q.Set("repo", did) q.Set("collection", collection) if opts.Limit > 0 { q.Set("limit", fmt.Sprintf("%d", opts.Limit)) } if opts.Cursor != "" { q.Set("cursor", opts.Cursor) } if opts.Reverse { q.Set("reverse", "true") } reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?%s", pdsEndpoint, q.Encode())
req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) if err != nil { return nil, "", fmt.Errorf("build request: %w", err) }
resp, err := c.httpClient.Do(req) if err != nil { return nil, "", fmt.Errorf("list records: %w", err) } defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { return nil, "", fmt.Errorf("list records: HTTP %d", resp.StatusCode) }
var result struct { Records []struct { URI string `json:"uri"` CID string `json:"cid"` Value map[string]any `json:"value"` } `json:"records"` Cursor string `json:"cursor"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return nil, "", fmt.Errorf("decode records: %w", err) }
records := make([]Record, len(result.Records)) for i, r := range result.Records { records[i] = Record{URI: r.URI, CID: r.CID, Value: r.Value} } return records, result.Cursor, nil}
// GetPublicRecord fetches a single public record from a user's PDS.func (c *PublicClient) GetPublicRecord(ctx context.Context, did, collection, rkey string) (*Record, error) { pdsEndpoint, err := c.GetPDSEndpoint(ctx, did) if err != nil { return nil, fmt.Errorf("resolve PDS: %w", err) }
reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s", pdsEndpoint, url.QueryEscape(did), url.QueryEscape(collection), url.QueryEscape(rkey))
req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) if err != nil { return nil, fmt.Errorf("build request: %w", err) }
resp, err := c.httpClient.Do(req) if err != nil { return nil, fmt.Errorf("get record: %w", err) } defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("get record: HTTP %d", resp.StatusCode) }
var r struct { URI string `json:"uri"` CID string `json:"cid"` Value map[string]any `json:"value"` } if err := json.NewDecoder(resp.Body).Decode(&r); err != nil { return nil, fmt.Errorf("decode record: %w", err) } return &Record{URI: r.URI, CID: r.CID, Value: r.Value}, nil}
// ListAllRecords paginates through every record in a collection on the user's// PDS and returns them all, newest-first. Cap of 10k records (100 pages × 100 records).func (c *PublicClient) ListAllRecords(ctx context.Context, did, collection string) ([]Record, error) { const pageSize = 100 const maxPages = 100
var all []Record cursor := "" for page := 0; page < maxPages; page++ { records, next, err := c.ListPublicRecords(ctx, did, collection, ListPublicRecordsOpts{ Limit: pageSize, Cursor: cursor, Reverse: true, }) if err != nil { return nil, err } all = append(all, records...) if next == "" || len(records) == 0 { return all, nil } cursor = next } return all, nil}
// isPrivateIP reports whether ip is in a private/reserved range.func isPrivateIP(ip net.IP) bool { return ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsPrivate() || ip.IsUnspecified() || ip.Equal(net.ParseIP("[IP_ADDRESS]")) // cloud metadata}
// validateDomain blocks requests to private/internal hosts.func validateDomain(domain string) error { if domain == "localhost" || strings.HasSuffix(domain, ".local") { return ErrSSRFBlocked } if ip := net.ParseIP(domain); ip != nil { if isPrivateIP(ip) { return ErrSSRFBlocked } return nil } ips, err := net.LookupIP(domain) if err != nil { return nil // let the HTTP request fail naturally } if slices.ContainsFunc(ips, isPrivateIP) { return ErrSSRFBlocked } return nil}