diff --git a/cmd/appview/serve.go b/cmd/appview/serve.go index 47d255d..297eb9a 100644 --- a/cmd/appview/serve.go +++ b/cmd/appview/serve.go @@ -14,7 +14,6 @@ import ( "syscall" "time" - "github.com/bluesky-social/indigo/atproto/syntax" "github.com/distribution/distribution/v3/registry" "github.com/distribution/distribution/v3/registry/handlers" "github.com/spf13/cobra" @@ -186,17 +185,17 @@ func serveRegistry(cmd *cobra.Command, args []string) error { } else { // Register UI routes with dependencies routes.RegisterUIRoutes(mainRouter, routes.UIDependencies{ - Database: uiDatabase, - ReadOnlyDB: uiReadOnlyDB, - SessionStore: uiSessionStore, + Database: uiDatabase, + ReadOnlyDB: uiReadOnlyDB, + SessionStore: uiSessionStore, OAuthClientApp: oauthClientApp, - OAuthStore: oauthStore, - Refresher: refresher, - BaseURL: baseURL, - DeviceStore: deviceStore, - HealthChecker: healthChecker, - ReadmeCache: readmeCache, - Templates: uiTemplates, + OAuthStore: oauthStore, + Refresher: refresher, + BaseURL: baseURL, + DeviceStore: deviceStore, + HealthChecker: healthChecker, + ReadmeCache: readmeCache, + Templates: uiTemplates, }) } } @@ -215,30 +214,8 @@ func serveRegistry(cmd *cobra.Command, args []string) error { oauthServer.SetPostAuthCallback(func(ctx context.Context, did, handle, pdsEndpoint, sessionID string) error { slog.Debug("OAuth post-auth callback", "component", "appview/callback", "did", did) - // Parse DID for session resume - didParsed, err := syntax.ParseDID(did) - if err != nil { - slog.Warn("Failed to parse DID", "component", "appview/callback", "did", did, "error", err) - return nil // Non-fatal - } - - // Resume OAuth session to get authenticated client - session, err := oauthClientApp.ResumeSession(ctx, didParsed, sessionID) - if err != nil { - slog.Warn("Failed to resume session", "component", "appview/callback", "did", did, "error", err) - // Fallback: update user without avatar - _ = db.UpsertUser(uiDatabase, &db.User{ - DID: did, - Handle: handle, - PDSEndpoint: pdsEndpoint, - Avatar: "", - LastSeen: time.Now(), - }) - return nil // Non-fatal - } - - // Create authenticated atproto client using the indigo session's API client - client := atproto.NewClientWithIndigoClient(pdsEndpoint, did, session.APIClient()) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + client := atproto.NewClientWithSessionProvider(pdsEndpoint, did, refresher) // Ensure sailor profile exists (creates with default hold if configured) slog.Debug("Ensuring profile exists", "component", "appview/callback", "did", did, "default_hold_did", defaultHoldDID) diff --git a/pkg/appview/handlers/api.go b/pkg/appview/handlers/api.go index 8d2cfaf..a08564e 100644 --- a/pkg/appview/handlers/api.go +++ b/pkg/appview/handlers/api.go @@ -43,18 +43,9 @@ func (h *StarRepositoryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request return } - // Get OAuth session for the authenticated user - slog.Debug("Getting OAuth session for star", "user_did", user.DID) - session, err := h.Refresher.GetSession(r.Context(), user.DID) - if err != nil { - slog.Warn("Failed to get OAuth session for star", "user_did", user.DID, "error", err) - http.Error(w, fmt.Sprintf("Failed to get OAuth session: %v", err), http.StatusUnauthorized) - return - } - - // Get user's PDS client (use indigo's API client which handles DPoP automatically) - apiClient := session.APIClient() - pdsClient := atproto.NewClientWithIndigoClient(user.PDSEndpoint, user.DID, apiClient) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + slog.Debug("Creating PDS client for star", "user_did", user.DID) + pdsClient := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) // Create star record starRecord := atproto.NewStarRecord(ownerDID, repository) @@ -106,18 +97,9 @@ func (h *UnstarRepositoryHandler) ServeHTTP(w http.ResponseWriter, r *http.Reque return } - // Get OAuth session for the authenticated user - slog.Debug("Getting OAuth session for unstar", "user_did", user.DID) - session, err := h.Refresher.GetSession(r.Context(), user.DID) - if err != nil { - slog.Warn("Failed to get OAuth session for unstar", "user_did", user.DID, "error", err) - http.Error(w, fmt.Sprintf("Failed to get OAuth session: %v", err), http.StatusUnauthorized) - return - } - - // Get user's PDS client (use indigo's API client which handles DPoP automatically) - apiClient := session.APIClient() - pdsClient := atproto.NewClientWithIndigoClient(user.PDSEndpoint, user.DID, apiClient) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + slog.Debug("Creating PDS client for unstar", "user_did", user.DID) + pdsClient := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) // Delete star record from user's PDS rkey := atproto.StarRecordKey(ownerDID, repository) @@ -172,19 +154,9 @@ func (h *CheckStarHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - // Get OAuth session for the authenticated user - session, err := h.Refresher.GetSession(r.Context(), user.DID) - if err != nil { - slog.Debug("Failed to get OAuth session for check star", "user_did", user.DID, "error", err) - // No OAuth session - return not starred - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]bool{"starred": false}) - return - } - - // Get user's PDS client (use indigo's API client which handles DPoP automatically) - apiClient := session.APIClient() - pdsClient := atproto.NewClientWithIndigoClient(user.PDSEndpoint, user.DID, apiClient) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + // Note: Error handling moves to the PDS call - if session doesn't exist, GetRecord will fail + pdsClient := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) // Check if star record exists rkey := atproto.StarRecordKey(ownerDID, repository) diff --git a/pkg/appview/handlers/images.go b/pkg/appview/handlers/images.go index a561a6c..7e4949c 100644 --- a/pkg/appview/handlers/images.go +++ b/pkg/appview/handlers/images.go @@ -30,16 +30,8 @@ func (h *DeleteTagHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { repo := chi.URLParam(r, "repository") tag := chi.URLParam(r, "tag") - // Get OAuth session for the authenticated user - session, err := h.Refresher.GetSession(r.Context(), user.DID) - if err != nil { - http.Error(w, fmt.Sprintf("Failed to get OAuth session: %v", err), http.StatusUnauthorized) - return - } - - // Create ATProto client with OAuth credentials - apiClient := session.APIClient() - pdsClient := atproto.NewClientWithIndigoClient(user.PDSEndpoint, user.DID, apiClient) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + pdsClient := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) // Compute rkey for tag record (repository_tag with slashes replaced) rkey := fmt.Sprintf("%s_%s", repo, tag) @@ -108,16 +100,8 @@ func (h *DeleteManifestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request return } - // Get OAuth session for the authenticated user - session, err := h.Refresher.GetSession(r.Context(), user.DID) - if err != nil { - http.Error(w, fmt.Sprintf("Failed to get OAuth session: %v", err), http.StatusUnauthorized) - return - } - - // Create ATProto client with OAuth credentials - apiClient := session.APIClient() - pdsClient := atproto.NewClientWithIndigoClient(user.PDSEndpoint, user.DID, apiClient) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + pdsClient := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) // If tagged and confirmed, delete all tags first if tagged && confirmed { diff --git a/pkg/appview/handlers/repository.go b/pkg/appview/handlers/repository.go index 4383660..ccdc630 100644 --- a/pkg/appview/handlers/repository.go +++ b/pkg/appview/handlers/repository.go @@ -163,18 +163,13 @@ func (h *RepositoryPageHandler) ServeHTTP(w http.ResponseWriter, r *http.Request isStarred := false user := middleware.GetUser(r) if user != nil && h.Refresher != nil && h.Directory != nil { - // Get OAuth session for the authenticated user - session, err := h.Refresher.GetSession(r.Context(), user.DID) - if err == nil { - // Get user's PDS client - apiClient := session.APIClient() - pdsClient := atproto.NewClientWithIndigoClient(user.PDSEndpoint, user.DID, apiClient) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + pdsClient := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) - // Check if star record exists - rkey := atproto.StarRecordKey(owner.DID, repository) - _, err = pdsClient.GetRecord(r.Context(), atproto.StarCollection, rkey) - isStarred = (err == nil) - } + // Check if star record exists + rkey := atproto.StarRecordKey(owner.DID, repository) + _, err := pdsClient.GetRecord(r.Context(), atproto.StarCollection, rkey) + isStarred = (err == nil) } // Check if current user is the repository owner diff --git a/pkg/appview/handlers/settings.go b/pkg/appview/handlers/settings.go index 5430281..99f426e 100644 --- a/pkg/appview/handlers/settings.go +++ b/pkg/appview/handlers/settings.go @@ -26,20 +26,8 @@ func (h *SettingsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - // Get OAuth session for the user - session, err := h.Refresher.GetSession(r.Context(), user.DID) - if err != nil { - // OAuth session not found or expired - redirect to re-authenticate - slog.Warn("OAuth session not found, redirecting to login", "component", "settings", "did", user.DID, "error", err) - http.Redirect(w, r, "/auth/oauth/login?return_to=/settings", http.StatusFound) - return - } - - // Use indigo's API client directly - it handles all auth automatically - apiClient := session.APIClient() - - // Create ATProto client with indigo's XRPC client - client := atproto.NewClientWithIndigoClient(user.PDSEndpoint, user.DID, apiClient) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + client := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) // Fetch sailor profile profile, err := storage.GetProfile(r.Context(), client) @@ -96,20 +84,8 @@ func (h *UpdateDefaultHoldHandler) ServeHTTP(w http.ResponseWriter, r *http.Requ holdEndpoint := r.FormValue("hold_endpoint") - // Get OAuth session for the user - session, err := h.Refresher.GetSession(r.Context(), user.DID) - if err != nil { - // OAuth session not found or expired - redirect to re-authenticate - slog.Warn("OAuth session not found, redirecting to login", "component", "settings", "did", user.DID, "error", err) - http.Redirect(w, r, "/auth/oauth/login?return_to=/settings", http.StatusFound) - return - } - - // Use indigo's API client directly - it handles all auth automatically - apiClient := session.APIClient() - - // Create ATProto client with indigo's XRPC client - client := atproto.NewClientWithIndigoClient(user.PDSEndpoint, user.DID, apiClient) + // Create ATProto client with session provider (uses DoWithSession for DPoP nonce safety) + client := atproto.NewClientWithSessionProvider(user.PDSEndpoint, user.DID, h.Refresher) // Fetch existing profile or create new one profile, err := storage.GetProfile(r.Context(), client) diff --git a/pkg/appview/middleware/registry.go b/pkg/appview/middleware/registry.go index 30b6389..6dd572f 100644 --- a/pkg/appview/middleware/registry.go +++ b/pkg/appview/middleware/registry.go @@ -409,15 +409,9 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name var atprotoClient *atproto.Client if nr.refresher != nil { - // Try OAuth flow first - session, err := nr.refresher.GetSession(ctx, did) - if err == nil { - // OAuth session available - use indigo's API client (handles DPoP automatically) - apiClient := session.APIClient() - atprotoClient = atproto.NewClientWithIndigoClient(pdsEndpoint, did, apiClient) - } else { - slog.Debug("OAuth refresh failed, falling back to Basic Auth", "component", "registry/middleware", "did", did, "error", err) - } + // Use session provider for locked OAuth sessions + // This prevents DPoP nonce race conditions during concurrent layer uploads + atprotoClient = atproto.NewClientWithSessionProvider(pdsEndpoint, did, nr.refresher) } // Fall back to Basic Auth token cache if OAuth not available diff --git a/pkg/atproto/client.go b/pkg/atproto/client.go index a64302a..ab80f98 100644 --- a/pkg/atproto/client.go +++ b/pkg/atproto/client.go @@ -12,6 +12,7 @@ import ( "strings" "github.com/bluesky-social/indigo/atproto/atclient" + indigo_oauth "github.com/bluesky-social/indigo/atproto/auth/oauth" ) // Sentinel errors @@ -19,14 +20,22 @@ var ( ErrRecordNotFound = errors.New("record not found") ) +// SessionProvider provides locked OAuth sessions for PDS operations. +// This interface allows the ATProto client to use DoWithSession() for each PDS call, +// preventing DPoP nonce race conditions during concurrent operations. +type SessionProvider interface { + // DoWithSession executes fn with a locked OAuth session. + // The lock is held for the entire duration, serializing DPoP nonce updates. + DoWithSession(ctx context.Context, did string, fn func(session *indigo_oauth.ClientSession) error) error +} + // Client wraps ATProto operations for the registry type Client struct { pdsEndpoint string did string accessToken string // For Basic Auth only httpClient *http.Client - useIndigoClient bool // true if using indigo's OAuth client (handles auth automatically) - indigoClient *atclient.APIClient // indigo's API client for OAuth requests + sessionProvider SessionProvider // For locked OAuth sessions (prevents DPoP nonce races) } // NewClient creates a new ATProto client for Basic Auth tokens (app passwords) @@ -39,15 +48,20 @@ func NewClient(pdsEndpoint, did, accessToken string) *Client { } } -// NewClientWithIndigoClient creates an ATProto client using indigo's API client -// This uses indigo's native XRPC methods with automatic DPoP handling -func NewClientWithIndigoClient(pdsEndpoint, did string, indigoClient *atclient.APIClient) *Client { +// NewClientWithSessionProvider creates an ATProto client that uses locked OAuth sessions. +// This is the preferred constructor for concurrent operations (e.g., Docker layer uploads) +// as it prevents DPoP nonce race conditions by serializing PDS calls per-DID. +// +// Each PDS call acquires a per-DID lock, ensuring that: +// - Only one goroutine at a time can negotiate DPoP nonces with the PDS +// - The session's nonce is saved to DB before other goroutines load it +// - Concurrent manifest operations don't cause nonce thrashing +func NewClientWithSessionProvider(pdsEndpoint, did string, sessionProvider SessionProvider) *Client { return &Client{ pdsEndpoint: pdsEndpoint, did: did, - useIndigoClient: true, - indigoClient: indigoClient, - httpClient: indigoClient.Client, // Keep for any fallback cases + sessionProvider: sessionProvider, + httpClient: &http.Client{}, } } @@ -67,10 +81,13 @@ func (c *Client) PutRecord(ctx context.Context, collection, rkey string, record "record": record, } - // Use indigo API client (OAuth with DPoP) - if c.useIndigoClient && c.indigoClient != nil { + // Use session provider (locked OAuth with DPoP) - prevents nonce races + if c.sessionProvider != nil { var result Record - err := c.indigoClient.Post(ctx, "com.atproto.repo.putRecord", payload, &result) + err := c.sessionProvider.DoWithSession(ctx, c.did, func(session *indigo_oauth.ClientSession) error { + apiClient := session.APIClient() + return apiClient.Post(ctx, "com.atproto.repo.putRecord", payload, &result) + }) if err != nil { return nil, fmt.Errorf("putRecord failed: %w", err) } @@ -113,16 +130,19 @@ func (c *Client) PutRecord(ctx context.Context, collection, rkey string, record // GetRecord retrieves a record from the ATProto repository func (c *Client) GetRecord(ctx context.Context, collection, rkey string) (*Record, error) { - // Use indigo API client (OAuth with DPoP) - if c.useIndigoClient && c.indigoClient != nil { - params := map[string]any{ - "repo": c.did, - "collection": collection, - "rkey": rkey, - } + params := map[string]any{ + "repo": c.did, + "collection": collection, + "rkey": rkey, + } + // Use session provider (locked OAuth with DPoP) - prevents nonce races + if c.sessionProvider != nil { var result Record - err := c.indigoClient.Get(ctx, "com.atproto.repo.getRecord", params, &result) + err := c.sessionProvider.DoWithSession(ctx, c.did, func(session *indigo_oauth.ClientSession) error { + apiClient := session.APIClient() + return apiClient.Get(ctx, "com.atproto.repo.getRecord", params, &result) + }) if err != nil { // Check for RecordNotFound error from indigo's APIError type var apiErr *atclient.APIError @@ -187,10 +207,13 @@ func (c *Client) DeleteRecord(ctx context.Context, collection, rkey string) erro "rkey": rkey, } - // Use indigo API client (OAuth with DPoP) - if c.useIndigoClient && c.indigoClient != nil { - var result map[string]any // deleteRecord returns empty object on success - err := c.indigoClient.Post(ctx, "com.atproto.repo.deleteRecord", payload, &result) + // Use session provider (locked OAuth with DPoP) - prevents nonce races + if c.sessionProvider != nil { + err := c.sessionProvider.DoWithSession(ctx, c.did, func(session *indigo_oauth.ClientSession) error { + apiClient := session.APIClient() + var result map[string]any // deleteRecord returns empty object on success + return apiClient.Post(ctx, "com.atproto.repo.deleteRecord", payload, &result) + }) if err != nil { return fmt.Errorf("deleteRecord failed: %w", err) } @@ -279,20 +302,23 @@ type Link struct { // UploadBlob uploads binary data to the PDS and returns a blob reference func (c *Client) UploadBlob(ctx context.Context, data []byte, mimeType string) (*ATProtoBlobRef, error) { - // Use indigo API client (OAuth with DPoP) - if c.useIndigoClient && c.indigoClient != nil { + // Use session provider (locked OAuth with DPoP) - prevents nonce races + if c.sessionProvider != nil { var result struct { Blob ATProtoBlobRef `json:"blob"` } - err := c.indigoClient.LexDo(ctx, - "POST", - mimeType, - "com.atproto.repo.uploadBlob", - nil, - data, - &result, - ) + err := c.sessionProvider.DoWithSession(ctx, c.did, func(session *indigo_oauth.ClientSession) error { + apiClient := session.APIClient() + return apiClient.LexDo(ctx, + "POST", + mimeType, + "com.atproto.repo.uploadBlob", + nil, + data, + &result, + ) + }) if err != nil { return nil, fmt.Errorf("uploadBlob failed: %w", err) } @@ -510,21 +536,7 @@ type ProfileRecord struct { // GetActorProfile fetches an actor's profile from their PDS // The actor parameter can be a DID or handle func (c *Client) GetActorProfile(ctx context.Context, actor string) (*ActorProfile, error) { - // Use indigo API client (OAuth with DPoP) - if c.useIndigoClient && c.indigoClient != nil { - params := map[string]any{ - "actor": actor, - } - - var profile ActorProfile - err := c.indigoClient.Get(ctx, "app.bsky.actor.getProfile", params, &profile) - if err != nil { - return nil, fmt.Errorf("getProfile failed: %w", err) - } - return &profile, nil - } - - // Basic Auth (app passwords) + // Basic Auth (app passwords) or unauthenticated url := fmt.Sprintf("%s/xrpc/app.bsky.actor.getProfile?actor=%s", c.pdsEndpoint, actor) req, err := http.NewRequestWithContext(ctx, "GET", url, nil) @@ -563,19 +575,21 @@ func (c *Client) GetActorProfile(ctx context.Context, actor string) (*ActorProfi // GetProfileRecord fetches the app.bsky.actor.profile record from PDS // This returns the raw profile record with blob references (not CDN URLs) func (c *Client) GetProfileRecord(ctx context.Context, did string) (*ProfileRecord, error) { - // Use indigo API client (OAuth with DPoP) - if c.useIndigoClient && c.indigoClient != nil { - params := map[string]any{ - "repo": did, - "collection": "app.bsky.actor.profile", - "rkey": "self", - } + params := map[string]any{ + "repo": did, + "collection": "app.bsky.actor.profile", + "rkey": "self", + } + // Use session provider (locked OAuth with DPoP) - prevents nonce races + if c.sessionProvider != nil { var result struct { Value ProfileRecord `json:"value"` } - - err := c.indigoClient.Get(ctx, "com.atproto.repo.getRecord", params, &result) + err := c.sessionProvider.DoWithSession(ctx, c.did, func(session *indigo_oauth.ClientSession) error { + apiClient := session.APIClient() + return apiClient.Get(ctx, "com.atproto.repo.getRecord", params, &result) + }) if err != nil { return nil, fmt.Errorf("getRecord failed: %w", err) } diff --git a/pkg/atproto/client_test.go b/pkg/atproto/client_test.go index bb98541..9fce76a 100644 --- a/pkg/atproto/client_test.go +++ b/pkg/atproto/client_test.go @@ -23,8 +23,8 @@ func TestNewClient(t *testing.T) { if client.accessToken != "token123" { t.Errorf("accessToken = %v, want token123", client.accessToken) } - if client.useIndigoClient { - t.Error("useIndigoClient should be false for Basic Auth client") + if client.sessionProvider != nil { + t.Error("sessionProvider should be nil for Basic Auth client") } } @@ -1003,21 +1003,6 @@ func TestClientPDSEndpoint(t *testing.T) { } } -// TestNewClientWithIndigoClient tests client initialization with Indigo client -func TestNewClientWithIndigoClient(t *testing.T) { - // Note: We can't easily create a real indigo client in tests without complex setup - // We pass nil for the indigo client, which is acceptable for testing the constructor - // The actual client.go code will handle nil indigo client by checking before use - - // Skip this test for now as it requires a real indigo client - // The function is tested indirectly through integration tests - t.Skip("Skipping TestNewClientWithIndigoClient - requires real indigo client setup") - - // When properly set up with a real indigo client, the test would look like: - // client := NewClientWithIndigoClient("https://pds.example.com", "did:plc:test123", indigoClient) - // if !client.useIndigoClient { t.Error("useIndigoClient should be true") } -} - // TestListRecordsError tests error handling in ListRecords func TestListRecordsError(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index 13c6eff..6dcf8a5 100644 --- a/pkg/auth/oauth/client.go +++ b/pkg/auth/oauth/client.go @@ -169,46 +169,68 @@ func (r *Refresher) SetUISessionStore(store UISessionStore) { r.uiSessionStore = store } -// GetSession gets a fresh OAuth session for a DID -// Loads session from database on every request (database is source of truth) -// Uses per-DID locking to prevent concurrent requests from racing on DPoP nonce updates +// DoWithSession executes a function with a locked OAuth session. +// The lock is held for the entire duration of the function, preventing DPoP nonce races. +// +// This is the preferred way to make PDS requests that require OAuth/DPoP authentication. +// The lock is held through the entire PDS interaction, ensuring that: +// 1. Only one goroutine at a time can negotiate DPoP nonces with the PDS for a given DID +// 2. The session's PersistSessionCallback saves the updated nonce before other goroutines load +// 3. Concurrent layer uploads don't race on stale nonces // // Why locking is critical: // During docker push, multiple layers upload concurrently. Each layer creates a new // ClientSession by loading from database. Without locking, this race condition occurs: -// 1. Layer A loads session with stale DPoP nonce from DB -// 2. Layer B loads session with same stale nonce (A hasn't updated DB yet) -// 3. Layer A makes request → 401 "use_dpop_nonce" → gets fresh nonce → saves to DB -// 4. Layer B makes request → 401 "use_dpop_nonce" (using stale nonce from step 2) -// 5. DPoP nonce thrashing continues, eventually causing 500 errors +// 1. Layer A loads session with stale DPoP nonce from DB +// 2. Layer B loads session with same stale nonce (A hasn't updated DB yet) +// 3. Layer A makes request → 401 "use_dpop_nonce" → gets fresh nonce → saves to DB +// 4. Layer B makes request → 401 "use_dpop_nonce" (using stale nonce from step 2) +// 5. DPoP nonce thrashing continues, eventually causing 500 errors // // With per-DID locking: -// 1. Layer A acquires lock, loads session, handles nonce negotiation, saves, releases lock -// 2. Layer B acquires lock AFTER A releases, loads fresh nonce from DB, succeeds -func (r *Refresher) GetSession(ctx context.Context, did string) (*oauth.ClientSession, error) { - // Get or create a mutex for this DID to prevent concurrent session loads - // This prevents DPoP nonce race conditions when multiple layers upload simultaneously +// 1. Layer A acquires lock, loads session, handles nonce negotiation, saves, releases lock +// 2. Layer B acquires lock AFTER A releases, loads fresh nonce from DB, succeeds +// +// Example usage: +// +// var result MyResult +// err := refresher.DoWithSession(ctx, did, func(session *oauth.ClientSession) error { +// resp, err := session.DoWithAuth(session.Client, req, "com.atproto.server.getServiceAuth") +// if err != nil { +// return err +// } +// // Parse response into result... +// return nil +// }) +func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(session *oauth.ClientSession) error) error { + // Get or create a mutex for this DID mutexInterface, _ := r.didLocks.LoadOrStore(did, &sync.Mutex{}) mutex := mutexInterface.(*sync.Mutex) - // Serialize session loading per DID + // Hold the lock for the ENTIRE operation (load + PDS request + nonce save) mutex.Lock() defer mutex.Unlock() - slog.Debug("Acquired session lock for DID", + slog.Debug("Acquired session lock for DoWithSession", "component", "oauth/refresher", "did", did) + // Load session while holding lock session, err := r.resumeSession(ctx, did) if err != nil { - return nil, err + return err } - slog.Debug("Released session lock for DID", - "component", "oauth/refresher", - "did", did) + // Execute the function (PDS request) while still holding lock + // The session's PersistSessionCallback will save nonce updates to DB + err = fn(session) - return session, nil + slog.Debug("Released session lock for DoWithSession", + "component", "oauth/refresher", + "did", did, + "success", err == nil) + + return err } // resumeSession loads a session from storage diff --git a/pkg/auth/token/servicetoken.go b/pkg/auth/token/servicetoken.go index 2635405..378fbb3 100644 --- a/pkg/auth/token/servicetoken.go +++ b/pkg/auth/token/servicetoken.go @@ -15,6 +15,7 @@ import ( "atcr.io/pkg/auth" "atcr.io/pkg/auth/oauth" "github.com/bluesky-social/indigo/atproto/atclient" + indigo_oauth "github.com/bluesky-social/indigo/atproto/auth/oauth" ) // getErrorHint provides context-specific troubleshooting hints based on API error type @@ -47,6 +48,9 @@ func getErrorHint(apiErr *atclient.APIError) string { // GetOrFetchServiceToken gets a service token for hold authentication. // Checks cache first, then fetches from PDS with OAuth/DPoP if needed. // This is the canonical implementation used by both middleware and crew registration. +// +// IMPORTANT: Uses DoWithSession() to hold a per-DID lock through the entire PDS interaction. +// This prevents DPoP nonce race conditions when multiple Docker layers upload concurrently. func GetOrFetchServiceToken( ctx context.Context, refresher *oauth.Refresher, @@ -74,9 +78,118 @@ func GetOrFetchServiceToken( slog.Debug("Service token expiring soon, proactively renewing", "did", did) } - session, err := refresher.GetSession(ctx, did) + // Use DoWithSession to hold the lock through the entire PDS interaction. + // This prevents DPoP nonce races when multiple goroutines try to fetch service tokens. + var serviceToken string + var fetchErr error + + err := refresher.DoWithSession(ctx, did, func(session *indigo_oauth.ClientSession) error { + // Double-check cache after acquiring lock - another goroutine may have + // populated it while we were waiting (classic double-checked locking pattern) + cachedToken, expiresAt := GetServiceToken(did, holdDID) + if cachedToken != "" && time.Until(expiresAt) > 10*time.Second { + slog.Debug("Service token cache hit after lock acquisition", + "did", did, + "expiresIn", time.Until(expiresAt).Round(time.Second)) + serviceToken = cachedToken + return nil + } + + // Cache still empty/expired - proceed with PDS call + // Request 5-minute expiry (PDS may grant less) + // exp must be absolute Unix timestamp, not relative duration + // Note: OAuth scope includes #atcr_hold fragment, but service auth aud must be bare DID + expiryTime := time.Now().Unix() + 300 // 5 minutes from now + serviceAuthURL := fmt.Sprintf("%s%s?aud=%s&lxm=%s&exp=%d", + pdsEndpoint, + atproto.ServerGetServiceAuth, + url.QueryEscape(holdDID), + url.QueryEscape("com.atproto.repo.getRecord"), + expiryTime, + ) + + req, err := http.NewRequestWithContext(ctx, "GET", serviceAuthURL, nil) + if err != nil { + fetchErr = fmt.Errorf("failed to create service auth request: %w", err) + return fetchErr + } + + // Use OAuth session to authenticate to PDS (with DPoP) + // The lock is held, so DPoP nonce negotiation is serialized per-DID + resp, err := session.DoWithAuth(session.Client, req, "com.atproto.server.getServiceAuth") + if err != nil { + // Auth error - may indicate expired tokens or corrupted session + InvalidateServiceToken(did, holdDID) + + // Inspect the error to extract detailed information from indigo's APIError + var apiErr *atclient.APIError + if errors.As(err, &apiErr) { + // Log detailed API error information + slog.Error("OAuth authentication failed during service token request", + "component", "token/servicetoken", + "did", did, + "holdDID", holdDID, + "pdsEndpoint", pdsEndpoint, + "url", serviceAuthURL, + "error", err, + "httpStatus", apiErr.StatusCode, + "errorName", apiErr.Name, + "errorMessage", apiErr.Message, + "hint", getErrorHint(apiErr)) + } else { + // Fallback for non-API errors (network errors, etc.) + slog.Error("OAuth authentication failed during service token request", + "component", "token/servicetoken", + "did", did, + "holdDID", holdDID, + "pdsEndpoint", pdsEndpoint, + "url", serviceAuthURL, + "error", err, + "errorType", fmt.Sprintf("%T", err), + "hint", "Network error or unexpected failure during OAuth request") + } + + fetchErr = fmt.Errorf("OAuth validation failed: %w", err) + return fetchErr + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + // Service auth failed + bodyBytes, _ := io.ReadAll(resp.Body) + InvalidateServiceToken(did, holdDID) + slog.Error("Service token request returned non-200 status", + "component", "token/servicetoken", + "did", did, + "holdDID", holdDID, + "pdsEndpoint", pdsEndpoint, + "statusCode", resp.StatusCode, + "responseBody", string(bodyBytes), + "hint", "PDS rejected the service token request - check PDS logs for details") + fetchErr = fmt.Errorf("service auth failed with status %d: %s", resp.StatusCode, string(bodyBytes)) + return fetchErr + } + + // Parse response to get service token + var result struct { + Token string `json:"token"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + fetchErr = fmt.Errorf("failed to decode service auth response: %w", err) + return fetchErr + } + + if result.Token == "" { + fetchErr = fmt.Errorf("empty token in service auth response") + return fetchErr + } + + serviceToken = result.Token + return nil + }) + if err != nil { - // OAuth session unavailable - fail + // DoWithSession failed (session load or callback error) InvalidateServiceToken(did, holdDID) // Try to extract detailed error information @@ -92,7 +205,8 @@ func GetOrFetchServiceToken( "errorName", apiErr.Name, "errorMessage", apiErr.Message, "hint", getErrorHint(apiErr)) - } else { + } else if fetchErr == nil { + // Session load failed (not a fetch error) slog.Error("Failed to get OAuth session for service token", "component", "token/servicetoken", "did", did, @@ -112,103 +226,12 @@ func GetOrFetchServiceToken( "error", delErr) } + if fetchErr != nil { + return "", fetchErr + } return "", fmt.Errorf("failed to get OAuth session: %w", err) } - // Call com.atproto.server.getServiceAuth on the user's PDS - // Request 5-minute expiry (PDS may grant less) - // exp must be absolute Unix timestamp, not relative duration - // Note: OAuth scope includes #atcr_hold fragment, but service auth aud must be bare DID - expiryTime := time.Now().Unix() + 300 // 5 minutes from now - serviceAuthURL := fmt.Sprintf("%s%s?aud=%s&lxm=%s&exp=%d", - pdsEndpoint, - atproto.ServerGetServiceAuth, - url.QueryEscape(holdDID), - url.QueryEscape("com.atproto.repo.getRecord"), - expiryTime, - ) - - req, err := http.NewRequestWithContext(ctx, "GET", serviceAuthURL, nil) - if err != nil { - return "", fmt.Errorf("failed to create service auth request: %w", err) - } - - // Use OAuth session to authenticate to PDS (with DPoP) - resp, err := session.DoWithAuth(session.Client, req, "com.atproto.server.getServiceAuth") - if err != nil { - // Auth error - may indicate expired tokens or corrupted session - InvalidateServiceToken(did, holdDID) - - // Inspect the error to extract detailed information from indigo's APIError - var apiErr *atclient.APIError - if errors.As(err, &apiErr) { - // Log detailed API error information - slog.Error("OAuth authentication failed during service token request", - "component", "token/servicetoken", - "did", did, - "holdDID", holdDID, - "pdsEndpoint", pdsEndpoint, - "url", serviceAuthURL, - "error", err, - "httpStatus", apiErr.StatusCode, - "errorName", apiErr.Name, - "errorMessage", apiErr.Message, - "hint", getErrorHint(apiErr)) - } else { - // Fallback for non-API errors (network errors, etc.) - slog.Error("OAuth authentication failed during service token request", - "component", "token/servicetoken", - "did", did, - "holdDID", holdDID, - "pdsEndpoint", pdsEndpoint, - "url", serviceAuthURL, - "error", err, - "errorType", fmt.Sprintf("%T", err), - "hint", "Network error or unexpected failure during OAuth request") - } - - // Delete the stale OAuth session to force re-authentication - // This also invalidates the UI session automatically - if delErr := refresher.DeleteSession(ctx, did); delErr != nil { - slog.Warn("Failed to delete stale OAuth session", - "component", "token/servicetoken", - "did", did, - "error", delErr) - } - - return "", fmt.Errorf("OAuth validation failed: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - // Service auth failed - bodyBytes, _ := io.ReadAll(resp.Body) - InvalidateServiceToken(did, holdDID) - slog.Error("Service token request returned non-200 status", - "component", "token/servicetoken", - "did", did, - "holdDID", holdDID, - "pdsEndpoint", pdsEndpoint, - "statusCode", resp.StatusCode, - "responseBody", string(bodyBytes), - "hint", "PDS rejected the service token request - check PDS logs for details") - return "", fmt.Errorf("service auth failed with status %d: %s", resp.StatusCode, string(bodyBytes)) - } - - // Parse response to get service token - var result struct { - Token string `json:"token"` - } - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return "", fmt.Errorf("failed to decode service auth response: %w", err) - } - - if result.Token == "" { - return "", fmt.Errorf("empty token in service auth response") - } - - serviceToken := result.Token - // Cache the token (parses JWT to extract actual expiry) if err := SetServiceToken(did, holdDID, serviceToken); err != nil { slog.Warn("Failed to cache service token", "error", err, "did", did, "holdDID", holdDID)