From 37bab324d773d0f3b7864a30eb988fae054f5d57 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sun, 5 Jul 2026 21:42:38 -0500 Subject: [PATCH] fix OAuth refresh-token burn on client cancellation causing sign-outs When a Docker client canceled a slow /auth/token request mid-refresh, the token-refresh POST was aborted client-side but completed on the PDS, which rotated the refresh token. The rotated token was never received or persisted, so the next refresh replayed the consumed token, got invalid_grant, and the session (OAuth + UI) was deleted, signing the user out everywhere. - Detach refresh POSTs from the inbound request context via a per-session RoundTripper (WithoutCancel + 30s cap); once a refresh starts it completes - Persist session updates (rotated tokens, DPoP nonces) on a detached context - Gate session deletion on IsSessionInvalidError: cancellation, timeouts, and transport errors no longer delete sessions; genuine invalid_grant still does - Add phase timing to /auth/token and per-DID lock wait warnings to attribute the ~14s pre-refresh stalls that push requests past Docker's deadline Co-Authored-By: Claude Fable 5 --- pkg/appview/handlers/oauth_errors.go | 12 +- pkg/auth/oauth/client.go | 85 +++++++++- pkg/auth/oauth/client_test.go | 238 +++++++++++++++++++++++++++ pkg/auth/oauth/transport.go | 107 ++++++++++++ pkg/auth/oauth/transport_test.go | 129 +++++++++++++++ pkg/auth/servicetoken.go | 21 ++- pkg/auth/servicetoken_test.go | 192 +++++++++++++++++++++ pkg/auth/token/handler.go | 48 +++++- 8 files changed, 819 insertions(+), 13 deletions(-) create mode 100644 pkg/auth/oauth/transport.go create mode 100644 pkg/auth/oauth/transport_test.go diff --git a/pkg/appview/handlers/oauth_errors.go b/pkg/appview/handlers/oauth_errors.go index 3523965..9a9ec41 100644 --- a/pkg/appview/handlers/oauth_errors.go +++ b/pkg/appview/handlers/oauth_errors.go @@ -19,6 +19,12 @@ func isOAuthError(err error) bool { return false } + // A canceled or timed-out request says nothing about session validity; + // deleting the session on those signs the user out over a transient blip. + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + // Check structured error types first var xrpcErr *xrpc.Error if errors.As(err, &xrpcErr) && (xrpcErr.StatusCode == 401 || xrpcErr.StatusCode == 403) { @@ -56,8 +62,10 @@ func handleOAuthError(ctx context.Context, refresher *oauth.Refresher, did strin "did", did, "error", err) - // Invalidate all UI sessions for this DID - if delErr := refresher.DeleteSession(ctx, did); delErr != nil { + // Invalidate all UI sessions for this DID. Detached context: once we + // decide to delete, the cleanup must finish even if the inbound request + // is canceled mid-way. + if delErr := refresher.DeleteSession(context.WithoutCancel(ctx), did); delErr != nil { slog.Warn("Failed to delete OAuth session after error", "component", "handlers", "did", did, diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index 65c0738..439e136 100644 --- a/pkg/auth/oauth/client.go +++ b/pkg/auth/oauth/client.go @@ -277,18 +277,34 @@ func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(sessi mutex := mutexInterface.(*sync.Mutex) // Hold the lock for the ENTIRE operation (load + PDS request + nonce save) + lockStart := time.Now() mutex.Lock() defer mutex.Unlock() + lockWait := time.Since(lockStart) slog.Debug("Acquired session lock for DoWithSession", "component", "oauth/refresher", - "did", did) + "did", did, + "lockWait", lockWait.Round(time.Millisecond)) + if lockWait > 5*time.Second { + slog.Warn("Slow per-DID session lock acquisition", + "component", "oauth/refresher", + "did", did, + "lockWait", lockWait.Round(time.Millisecond)) + } // Load session while holding lock + resumeStart := time.Now() session, err := r.resumeSession(ctx, did) if err != nil { return err } + if resumeDur := time.Since(resumeStart); resumeDur > 5*time.Second { + slog.Warn("Slow OAuth session resume", + "component", "oauth/refresher", + "did", did, + "duration", resumeDur.Round(time.Millisecond)) + } // Execute the function (PDS request) while still holding lock // The session's PersistSessionCallback will save nonce updates to DB @@ -300,9 +316,11 @@ func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(sessi "component", "oauth/refresher", "did", did, "error", err) - // Don't hold the lock while deleting - release first + // Don't hold the lock while deleting - release first. Detached + // context: once we decide to delete, the cleanup must finish even if + // the inbound request is canceled mid-way. mutex.Unlock() - _ = r.DeleteSession(ctx, did) + _ = r.DeleteSession(context.WithoutCancel(ctx), did) mutex.Lock() // Re-acquire for the deferred unlock } @@ -314,6 +332,48 @@ func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(sessi return err } +// IsSessionInvalidError reports whether err indicates the OAuth session +// itself is invalid or revoked, i.e. deleting it (and signing the user out) +// is the right response. It is deliberately false for cancellation, deadline, +// and transport errors: deleting a session over a transient failure signs the +// user out everywhere for nothing. +func IsSessionInvalidError(err error) bool { + if err == nil { + return false + } + + // A canceled or timed-out request says nothing about session validity. + // Checked first so wrapped chains never fall through to string matching. + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + + var xrpcErr *xrpc.Error + if errors.As(err, &xrpcErr) && xrpcErr.StatusCode == 401 { + return true + } + var apiErr *atclient.APIError + if errors.As(err, &apiErr) { + if apiErr.StatusCode == 401 { + return true + } + switch apiErr.Name { + case "InvalidToken", "InvalidGrant", "InsufficientScope": + return true + } + } + + // The token-refresh failure from indigo arrives as a plain wrapped error + // ("auth server request failed (HTTP 400): invalid_grant"), not an + // APIError, so a string fallback is required. These substrings are + // auth-specific and won't appear in digests or URIs. + errStr := strings.ToLower(err.Error()) + return strings.Contains(errStr, "invalid_grant") || + strings.Contains(errStr, "invalid_token") || + strings.Contains(errStr, "insufficient_scope") || + strings.Contains(errStr, "token expired") +} + // isAuthError checks if an error looks like an OAuth/auth failure // Uses structured error types to avoid false positives from substring matching // (e.g., a digest hash containing "401" in a RecordNotFound error) @@ -322,6 +382,12 @@ func isAuthError(err error) bool { return false } + // Never treat cancellation/deadline as an auth failure, no matter what + // the wrapped chain's text looks like. + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + // Check structured error types first var xrpcErr *xrpc.Error if errors.As(err, &xrpcErr) && xrpcErr.StatusCode == 401 { @@ -386,11 +452,22 @@ func (r *Refresher) resumeSession(ctx context.Context, did string) (*oauth.Clien return nil, fmt.Errorf("failed to resume session: %w", err) } + // Token-refresh POSTs rotate the refresh token on the auth server; an + // inbound request cancellation must never abort one mid-flight, or the + // rotated token is stranded server-side and the session dies with + // invalid_grant "Refresh token replayed" on the next refresh. + session.Client = newRefreshDetachClient(session.Client, session.Data.AuthServerTokenEndpoint) + // Set up callback to persist token updates to SQLite // This ensures that when indigo automatically refreshes tokens or updates DPoP nonces, // the new state is saved to the database immediately session.PersistSessionCallback = func(callbackCtx context.Context, updatedData *oauth.ClientSessionData) { - if err := r.clientApp.Store.SaveSession(callbackCtx, *updatedData); err != nil { + // Indigo invokes this with the context of whatever request triggered + // the refresh — possibly already canceled. Once tokens have rotated, + // this save must complete or the session is bricked. + saveCtx, cancel := context.WithTimeout(context.WithoutCancel(callbackCtx), 10*time.Second) + defer cancel() + if err := r.clientApp.Store.SaveSession(saveCtx, *updatedData); err != nil { slog.Error("Failed to persist OAuth session update", "component", "oauth/refresher", "did", did, diff --git a/pkg/auth/oauth/client_test.go b/pkg/auth/oauth/client_test.go index 2587cf8..bcc7177 100644 --- a/pkg/auth/oauth/client_test.go +++ b/pkg/auth/oauth/client_test.go @@ -1,9 +1,19 @@ package oauth import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "sync" "testing" + "time" + "github.com/bluesky-social/indigo/atproto/atclient" + "github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/atproto/auth/oauth" + "github.com/bluesky-social/indigo/atproto/syntax" ) func TestNewClientApp(t *testing.T) { @@ -156,3 +166,231 @@ func TestRefresher_SetUISessionStore(t *testing.T) { t.Error("Expected UI session store to be nil after setting nil") } } + +// ---------------------------------------------------------------------------- +// Refresh-cancellation regression tests +// ---------------------------------------------------------------------------- + +// fakeAuthStore is a ClientAuthStore whose SaveSession honors context +// cancellation, so a test fails if session persistence runs on a canceled +// context. It also implements GetLatestSessionForDID (the sessionGetter +// extension the Refresher requires). +type fakeAuthStore struct { + mu sync.Mutex + sessions map[string]oauth.ClientSessionData // keyed by DID + deleted []string +} + +func newFakeAuthStore() *fakeAuthStore { + return &fakeAuthStore{sessions: make(map[string]oauth.ClientSessionData)} +} + +func (s *fakeAuthStore) GetSession(ctx context.Context, did syntax.DID, sessionID string) (*oauth.ClientSessionData, error) { + s.mu.Lock() + defer s.mu.Unlock() + sess, ok := s.sessions[did.String()] + if !ok { + return nil, fmt.Errorf("session not found") + } + return &sess, nil +} + +func (s *fakeAuthStore) SaveSession(ctx context.Context, sess oauth.ClientSessionData) error { + if err := ctx.Err(); err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + s.sessions[sess.AccountDID.String()] = sess + return nil +} + +func (s *fakeAuthStore) DeleteSession(ctx context.Context, did syntax.DID, sessionID string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.sessions, did.String()) + s.deleted = append(s.deleted, did.String()) + return nil +} + +func (s *fakeAuthStore) GetAuthRequestInfo(ctx context.Context, state string) (*oauth.AuthRequestData, error) { + return nil, fmt.Errorf("not implemented") +} +func (s *fakeAuthStore) SaveAuthRequestInfo(ctx context.Context, info oauth.AuthRequestData) error { + return nil +} +func (s *fakeAuthStore) DeleteAuthRequestInfo(ctx context.Context, state string) error { + return nil +} + +func (s *fakeAuthStore) GetLatestSessionForDID(ctx context.Context, did string) (*oauth.ClientSessionData, string, error) { + s.mu.Lock() + defer s.mu.Unlock() + sess, ok := s.sessions[did] + if !ok { + return nil, "", fmt.Errorf("no session for DID") + } + return &sess, sess.SessionID, nil +} + +func (s *fakeAuthStore) refreshToken(did string) string { + s.mu.Lock() + defer s.mu.Unlock() + return s.sessions[did].RefreshToken +} + +func (s *fakeAuthStore) deletedDIDs() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.deleted...) +} + +// spyUISessionStore records DeleteByDID calls. +type spyUISessionStore struct { + mu sync.Mutex + deleted []string +} + +func (s *spyUISessionStore) Create(did, handle, pdsEndpoint string, duration time.Duration) (string, error) { + return "", fmt.Errorf("not implemented") +} + +func (s *spyUISessionStore) DeleteByDID(did string) { + s.mu.Lock() + defer s.mu.Unlock() + s.deleted = append(s.deleted, did) +} + +// seedSession stores a session for did pointing at the given resource server +// and token endpoint, with a freshly generated DPoP key. +func seedSession(t *testing.T, store *fakeAuthStore, did, hostURL, tokenEndpoint string) { + t.Helper() + key, err := atcrypto.GeneratePrivateKeyP256() + if err != nil { + t.Fatal(err) + } + parsedDID, err := syntax.ParseDID(did) + if err != nil { + t.Fatal(err) + } + store.sessions[did] = oauth.ClientSessionData{ + AccountDID: parsedDID, + SessionID: "test-session", + HostURL: hostURL, + AuthServerURL: hostURL, + AuthServerTokenEndpoint: tokenEndpoint, + Scopes: []string{"atproto"}, + AccessToken: "old-access", + RefreshToken: "old-refresh", + DPoPPrivateKeyMultibase: key.Multibase(), + } +} + +func newTestRefresher(t *testing.T, store oauth.ClientAuthStore) *Refresher { + t.Helper() + clientApp, err := NewClientApp("http://localhost:5000", store, []string{"atproto"}, "", "test") + if err != nil { + t.Fatal(err) + } + return NewRefresher(clientApp) +} + +// TestDoWithSession_RefreshSurvivesRequestCancellation reproduces the +// production incident: the caller's context is canceled while the token +// refresh is in flight (after the auth server has already rotated the +// refresh token). The rotated token MUST be persisted and the session MUST +// NOT be deleted, or the next refresh fails with invalid_grant "Refresh +// token replayed" and the user is signed out. +func TestDoWithSession_RefreshSurvivesRequestCancellation(t *testing.T) { + const did = "did:web:refresh-cancel.example.com" + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + mux := http.NewServeMux() + srv := httptest.NewServer(mux) + defer srv.Close() + + // Resource endpoint: reject the stale access token so DoWithAuth triggers + // a refresh; accept the rotated one. + mux.HandleFunc("/xrpc/test.resource", func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") == "DPoP new-access" { + w.WriteHeader(http.StatusOK) + return + } + w.Header().Set("WWW-Authenticate", `Bearer error="invalid_token", error_description="expired"`) + w.WriteHeader(http.StatusUnauthorized) + }) + + // Token endpoint: simulate the Docker client hanging up mid-refresh by + // canceling the caller's context BEFORE responding, then return rotated + // tokens (the auth server has already committed the rotation by then). + mux.HandleFunc("/oauth/token", func(w http.ResponseWriter, r *http.Request) { + cancel() + fmt.Fprintf(w, `{"sub":%q,"access_token":"new-access","refresh_token":"new-refresh"}`, did) + }) + + store := newFakeAuthStore() + seedSession(t, store, did, srv.URL, srv.URL+"/oauth/token") + refresher := newTestRefresher(t, store) + uiStore := &spyUISessionStore{} + refresher.SetUISessionStore(uiStore) + + err := refresher.DoWithSession(ctx, did, func(session *oauth.ClientSession) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/xrpc/test.resource", nil) + if err != nil { + return err + } + resp, err := session.DoWithAuth(session.Client, req, syntax.NSID("com.atproto.server.getServiceAuth")) + if err != nil { + return err + } + defer resp.Body.Close() + return nil + }) + + // The overall operation may fail (the post-refresh retry of the resource + // request runs on the canceled inbound context) — that is fine, Docker + // retries. What must hold is that the rotated refresh token was saved and + // the session survived. + if got := store.refreshToken(did); got != "new-refresh" { + t.Errorf("rotated refresh token not persisted: got %q, want %q (op err: %v)", got, "new-refresh", err) + } + if deleted := store.deletedDIDs(); len(deleted) != 0 { + t.Errorf("session was deleted: %v", deleted) + } + if len(uiStore.deleted) != 0 { + t.Errorf("UI session was invalidated: %v", uiStore.deleted) + } +} + +func TestIsSessionInvalidError(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"plain canceled", context.Canceled, false}, + {"wrapped canceled", fmt.Errorf("token refresh failed: %w", context.Canceled), false}, + {"wrapped deadline", fmt.Errorf("fetch: %w", context.DeadlineExceeded), false}, + // Even if the message mentions an auth string, cancellation wins. + {"canceled with auth-ish text", fmt.Errorf("invalid_grant: %w", context.Canceled), false}, + {"api error 401", &atclient.APIError{StatusCode: 401}, true}, + {"api error InvalidGrant", &atclient.APIError{StatusCode: 400, Name: "InvalidGrant"}, true}, + {"api error InvalidToken", &atclient.APIError{StatusCode: 400, Name: "InvalidToken"}, true}, + {"api error 500", &atclient.APIError{StatusCode: 500, Name: "InternalServerError"}, false}, + // The refresh-replay failure arrives as a plain wrapped string from indigo. + {"plain invalid_grant string", errors.New("failed to refresh OAuth tokens: token refresh failed (HTTP 400): invalid_grant"), true}, + {"plain invalid_token string", errors.New("auth server request failed (HTTP 401): invalid_token"), true}, + {"connection refused", errors.New(`Post "https://pds.example.com/oauth/token": dial tcp: connection refused`), false}, + {"generic 500", errors.New("token refresh failed (HTTP 500): server exploded"), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsSessionInvalidError(tt.err); got != tt.want { + t.Errorf("IsSessionInvalidError(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} diff --git a/pkg/auth/oauth/transport.go b/pkg/auth/oauth/transport.go new file mode 100644 index 0000000..fa71c64 --- /dev/null +++ b/pkg/auth/oauth/transport.go @@ -0,0 +1,107 @@ +package oauth + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/url" + "sync" + "time" +) + +// refreshDetachTimeout bounds a detached token-refresh POST. Refreshes are +// normally sub-second; this only exists so a hung auth server can't pin the +// per-DID session lock forever. +const refreshDetachTimeout = 30 * time.Second + +var badTokenEndpointLogOnce sync.Once + +// refreshDetachTransport detaches the request context for OAuth token-refresh +// POSTs. A refresh is non-idempotent: the auth server rotates the refresh +// token as soon as it processes the request, whether or not we stick around +// for the response. If the inbound request context is canceled mid-refresh +// (e.g. Docker gives up on a slow /auth/token), aborting the POST strands the +// rotated token server-side; our stored refresh token is then already +// consumed, the next refresh fails with invalid_grant "Refresh token +// replayed", and the whole session gets deleted. Once a refresh starts it +// must run to completion, capped by its own timeout (same rationale as the +// upload finalization in pkg/hold/oci/xrpc.go). +// +// Only POSTs to the session's auth-server token endpoint are detached; every +// other request keeps normal cancellation semantics. +type refreshDetachTransport struct { + base http.RoundTripper + tokenEndpoint *url.URL + timeout time.Duration +} + +func (t *refreshDetachTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if req.Method != http.MethodPost || !sameEndpoint(req.URL, t.tokenEndpoint) { + return t.base.RoundTrip(req) + } + + // context.WithoutCancel keeps values (trace/log metadata) but drops + // cancellation and deadline. The cancel func must outlive RoundTrip: the + // body is read by the caller, so it is released on Body.Close() instead + // of a defer here. + ctx, cancel := context.WithTimeout(context.WithoutCancel(req.Context()), t.timeout) + resp, err := t.base.RoundTrip(req.Clone(ctx)) + if err != nil { + cancel() + return nil, err + } + resp.Body = &cancelOnClose{ReadCloser: resp.Body, cancel: cancel} + return resp, nil +} + +// cancelOnClose ties a context.CancelFunc to response-body Close so the +// detached timeout context stays alive for the full body read. +type cancelOnClose struct { + io.ReadCloser + cancel context.CancelFunc +} + +func (c *cancelOnClose) Close() error { + c.cancel() + return c.ReadCloser.Close() +} + +// newRefreshDetachClient wraps inner so that token-refresh POSTs to +// tokenEndpoint survive cancellation of the inbound request context. If the +// endpoint can't be parsed (corrupt session data), the client is returned +// unwrapped, preserving the old behavior. +func newRefreshDetachClient(inner *http.Client, tokenEndpoint string) *http.Client { + endpoint, err := url.Parse(tokenEndpoint) + if err != nil || endpoint.Host == "" { + badTokenEndpointLogOnce.Do(func() { + slog.Warn("Not detaching token-refresh context: unparseable auth server token endpoint", + "component", "oauth/refresher", + "tokenEndpoint", tokenEndpoint, + "error", err) + }) + return inner + } + + if inner == nil { + inner = http.DefaultClient + } + base := inner.Transport + if base == nil { + base = http.DefaultTransport + } + + wrapped := *inner // shallow copy: keep Jar, Timeout, redirect policy + wrapped.Transport = &refreshDetachTransport{base: base, tokenEndpoint: endpoint, timeout: refreshDetachTimeout} + return &wrapped +} + +// sameEndpoint reports whether two URLs address the same endpoint, comparing +// scheme, host, and path rather than raw strings so that canonicalization +// differences (default ports, escaping) don't cause a miss. +func sameEndpoint(a, b *url.URL) bool { + if a == nil || b == nil { + return false + } + return a.Scheme == b.Scheme && a.Host == b.Host && a.Path == b.Path +} diff --git a/pkg/auth/oauth/transport_test.go b/pkg/auth/oauth/transport_test.go new file mode 100644 index 0000000..893d9dd --- /dev/null +++ b/pkg/auth/oauth/transport_test.go @@ -0,0 +1,129 @@ +package oauth + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" +) + +// TestRefreshDetachTransport_DetachesTokenEndpointPOST verifies the core +// property: a token-refresh POST proceeds and its response body stays +// readable even when the inbound request context is already canceled. +func TestRefreshDetachTransport_DetachesTokenEndpointPOST(t *testing.T) { + var served bool + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + served = true + _, _ = w.Write([]byte(`{"access_token":"new"}`)) + })) + defer srv.Close() + + client := newRefreshDetachClient(&http.Client{}, srv.URL+"/oauth/token") + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // canceled before the request even starts + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, srv.URL+"/oauth/token", strings.NewReader("grant_type=refresh_token")) + if err != nil { + t.Fatal(err) + } + + resp, err := client.Do(req) + if err != nil { + t.Fatalf("expected detached POST to succeed despite canceled context, got: %v", err) + } + defer resp.Body.Close() + + if !served { + t.Fatal("token endpoint handler never ran") + } + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("body not readable after parent cancel: %v", err) + } + if string(body) != `{"access_token":"new"}` { + t.Fatalf("unexpected body: %s", body) + } +} + +// TestRefreshDetachTransport_PassThrough verifies that requests other than +// POSTs to the token endpoint keep normal cancellation semantics. +func TestRefreshDetachTransport_PassThrough(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + client := newRefreshDetachClient(&http.Client{}, srv.URL+"/oauth/token") + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + cases := []struct { + name string + method string + url string + }{ + {"GET to token endpoint", http.MethodGet, srv.URL + "/oauth/token"}, + {"POST to other path", http.MethodPost, srv.URL + "/xrpc/other"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + req, err := http.NewRequestWithContext(ctx, tc.method, tc.url, nil) + if err != nil { + t.Fatal(err) + } + _, err = client.Do(req) //nolint:bodyclose // request must fail + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled pass-through, got: %v", err) + } + }) + } +} + +// TestRefreshDetachTransport_Timeout verifies the detached context still has +// its own bound so a hung auth server can't pin the session lock forever. +func TestRefreshDetachTransport_Timeout(t *testing.T) { + blocked := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-blocked + })) + defer srv.Close() + defer close(blocked) + + endpoint, err := url.Parse(srv.URL + "/oauth/token") + if err != nil { + t.Fatal(err) + } + client := &http.Client{Transport: &refreshDetachTransport{ + base: http.DefaultTransport, + tokenEndpoint: endpoint, + timeout: 50 * time.Millisecond, + }} + + req, err := http.NewRequest(http.MethodPost, srv.URL+"/oauth/token", nil) + if err != nil { + t.Fatal(err) + } + _, err = client.Do(req) //nolint:bodyclose // request must fail + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected DeadlineExceeded from detached timeout, got: %v", err) + } +} + +// TestNewRefreshDetachClient_BadEndpoint verifies fail-open to the unwrapped +// client when the stored token endpoint is unusable. +func TestNewRefreshDetachClient_BadEndpoint(t *testing.T) { + inner := &http.Client{} + if got := newRefreshDetachClient(inner, "://not a url"); got != inner { + t.Error("expected unwrapped client for unparseable endpoint") + } + if got := newRefreshDetachClient(inner, ""); got != inner { + t.Error("expected unwrapped client for empty endpoint") + } +} diff --git a/pkg/auth/servicetoken.go b/pkg/auth/servicetoken.go index a327269..5faa039 100644 --- a/pkg/auth/servicetoken.go +++ b/pkg/auth/servicetoken.go @@ -216,13 +216,24 @@ func GetOrFetchServiceToken( "hint", "OAuth session not found in database or token refresh failed") } - // 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", + // Delete the stale OAuth session to force re-authentication (this also + // invalidates the UI session) — but only when the error says the + // session itself is invalid. Cancellation, timeouts, and network + // failures are transient; deleting on those signs the user out + // everywhere over a blip. The delete runs on a detached context so a + // canceled inbound request can't leave it half-done. + if oauth.IsSessionInvalidError(err) { + if delErr := refresher.DeleteSession(context.WithoutCancel(ctx), did); delErr != nil { + slog.Warn("Failed to delete stale OAuth session", + "component", "token/servicetoken", + "did", did, + "error", delErr) + } + } else { + slog.Debug("Keeping OAuth session despite service token failure (transient error)", "component", "token/servicetoken", "did", did, - "error", delErr) + "error", err) } if fetchErr != nil { diff --git a/pkg/auth/servicetoken_test.go b/pkg/auth/servicetoken_test.go index 126c49b..94bc8e9 100644 --- a/pkg/auth/servicetoken_test.go +++ b/pkg/auth/servicetoken_test.go @@ -2,7 +2,18 @@ package auth import ( "context" + "fmt" + "net/http" + "net/http/httptest" + "sync" "testing" + "time" + + "atcr.io/pkg/atproto" + "atcr.io/pkg/auth/oauth" + "github.com/bluesky-social/indigo/atproto/atcrypto" + indigo_oauth "github.com/bluesky-social/indigo/atproto/auth/oauth" + "github.com/bluesky-social/indigo/atproto/syntax" ) func TestGetOrFetchServiceToken_NilRefresher(t *testing.T) { @@ -25,3 +36,184 @@ func TestGetOrFetchServiceToken_NilRefresher(t *testing.T) { // Note: Full tests with mocked OAuth refresher and HTTP client will be added // in the comprehensive test implementation phase + +// ---------------------------------------------------------------------------- +// Session-deletion gating tests (refresh-cancellation incident regression) +// ---------------------------------------------------------------------------- + +// stubAuthStore is a minimal indigo ClientAuthStore + GetLatestSessionForDID +// implementation for driving a real Refresher against httptest servers. +type stubAuthStore struct { + mu sync.Mutex + sessions map[string]indigo_oauth.ClientSessionData + deleted []string +} + +func newStubAuthStore() *stubAuthStore { + return &stubAuthStore{sessions: make(map[string]indigo_oauth.ClientSessionData)} +} + +func (s *stubAuthStore) GetSession(ctx context.Context, did syntax.DID, sessionID string) (*indigo_oauth.ClientSessionData, error) { + s.mu.Lock() + defer s.mu.Unlock() + sess, ok := s.sessions[did.String()] + if !ok { + return nil, fmt.Errorf("session not found") + } + return &sess, nil +} + +func (s *stubAuthStore) SaveSession(ctx context.Context, sess indigo_oauth.ClientSessionData) error { + if err := ctx.Err(); err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + s.sessions[sess.AccountDID.String()] = sess + return nil +} + +func (s *stubAuthStore) DeleteSession(ctx context.Context, did syntax.DID, sessionID string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.sessions, did.String()) + s.deleted = append(s.deleted, did.String()) + return nil +} + +func (s *stubAuthStore) GetAuthRequestInfo(ctx context.Context, state string) (*indigo_oauth.AuthRequestData, error) { + return nil, fmt.Errorf("not implemented") +} +func (s *stubAuthStore) SaveAuthRequestInfo(ctx context.Context, info indigo_oauth.AuthRequestData) error { + return nil +} +func (s *stubAuthStore) DeleteAuthRequestInfo(ctx context.Context, state string) error { + return nil +} + +func (s *stubAuthStore) GetLatestSessionForDID(ctx context.Context, did string) (*indigo_oauth.ClientSessionData, string, error) { + s.mu.Lock() + defer s.mu.Unlock() + sess, ok := s.sessions[did] + if !ok { + return nil, "", fmt.Errorf("no session for DID") + } + return &sess, sess.SessionID, nil +} + +func (s *stubAuthStore) deletedDIDs() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.deleted...) +} + +type recordingUIStore struct { + mu sync.Mutex + deleted []string +} + +func (s *recordingUIStore) Create(did, handle, pdsEndpoint string, duration time.Duration) (string, error) { + return "", fmt.Errorf("not implemented") +} + +func (s *recordingUIStore) DeleteByDID(did string) { + s.mu.Lock() + defer s.mu.Unlock() + s.deleted = append(s.deleted, did) +} + +// setupServiceTokenScenario builds a fake PDS whose getServiceAuth endpoint +// always 401s with invalid_token (forcing a token refresh) and whose token +// endpoint behavior is supplied by the caller, plus a Refresher seeded with a +// session for did. +func setupServiceTokenScenario(t *testing.T, did string, tokenHandler http.HandlerFunc) (*oauth.Refresher, *stubAuthStore, *recordingUIStore, string) { + t.Helper() + + mux := http.NewServeMux() + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + + mux.HandleFunc(atproto.ServerGetServiceAuth, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("WWW-Authenticate", `Bearer error="invalid_token", error_description="expired"`) + w.WriteHeader(http.StatusUnauthorized) + }) + mux.HandleFunc("/oauth/token", tokenHandler) + + key, err := atcrypto.GeneratePrivateKeyP256() + if err != nil { + t.Fatal(err) + } + parsedDID, err := syntax.ParseDID(did) + if err != nil { + t.Fatal(err) + } + + store := newStubAuthStore() + store.sessions[did] = indigo_oauth.ClientSessionData{ + AccountDID: parsedDID, + SessionID: "test-session", + HostURL: srv.URL, + AuthServerURL: srv.URL, + AuthServerTokenEndpoint: srv.URL + "/oauth/token", + Scopes: []string{"atproto"}, + AccessToken: "old-access", + RefreshToken: "old-refresh", + DPoPPrivateKeyMultibase: key.Multibase(), + } + + clientApp, err := oauth.NewClientApp("http://localhost:5000", store, []string{"atproto"}, "", "test") + if err != nil { + t.Fatal(err) + } + refresher := oauth.NewRefresher(clientApp) + uiStore := &recordingUIStore{} + refresher.SetUISessionStore(uiStore) + + return refresher, store, uiStore, srv.URL +} + +// TestGetOrFetchServiceToken_TransientErrorKeepsSession asserts that a +// transient refresh failure (5xx from the auth server) does NOT delete the +// OAuth session or invalidate UI sessions. +func TestGetOrFetchServiceToken_TransientErrorKeepsSession(t *testing.T) { + const did = "did:web:transient.example.com" + refresher, store, uiStore, pdsURL := setupServiceTokenScenario(t, did, + func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + }) + + _, err := GetOrFetchServiceToken(context.Background(), refresher, did, "did:web:hold1.example.com", pdsURL) + if err == nil { + t.Fatal("expected error from failed service token fetch") + } + if deleted := store.deletedDIDs(); len(deleted) != 0 { + t.Errorf("session deleted on transient error: %v", deleted) + } + if len(uiStore.deleted) != 0 { + t.Errorf("UI sessions invalidated on transient error: %v", uiStore.deleted) + } +} + +// TestGetOrFetchServiceToken_InvalidGrantDeletesSession asserts that a +// genuine invalid_grant (refresh token replayed/revoked) still deletes the +// session and invalidates UI sessions. +func TestGetOrFetchServiceToken_InvalidGrantDeletesSession(t *testing.T) { + const did = "did:web:invalidgrant.example.com" + refresher, store, uiStore, pdsURL := setupServiceTokenScenario(t, did, + func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"invalid_grant","error_description":"Refresh token replayed"}`)) + }) + + _, err := GetOrFetchServiceToken(context.Background(), refresher, did, "did:web:hold2.example.com", pdsURL) + if err == nil { + t.Fatal("expected error from invalid_grant") + } + if deleted := store.deletedDIDs(); len(deleted) != 1 || deleted[0] != did { + t.Errorf("expected session deletion for %s, got: %v", did, deleted) + } + if len(uiStore.deleted) != 1 || uiStore.deleted[0] != did { + t.Errorf("expected UI session invalidation for %s, got: %v", did, uiStore.deleted) + } +} diff --git a/pkg/auth/token/handler.go b/pkg/auth/token/handler.go index 339fad6..5ea3439 100644 --- a/pkg/auth/token/handler.go +++ b/pkg/auth/token/handler.go @@ -170,8 +170,14 @@ func sendOAuthSessionExpiredError(w http.ResponseWriter, r *http.Request) { render.JSON(w, r, resp) } +// slowPhaseThreshold flags /auth/token phases that take long enough to risk +// the Docker client's ~15s deadline; production incidents showed ~14s stalls +// before the first PDS call, and these phase timings exist to attribute them. +const slowPhaseThreshold = 5 * time.Second + // ServeHTTP handles the token request func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + phaseStart := time.Now() slog.Debug("Received token request", "method", r.Method, "path", r.URL.Path) // Only accept GET requests (per Docker spec) @@ -290,6 +296,17 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } + // Credential phase covers device-secret/app-password validation including + // any OAuth session validation (and its lazy token refresh). + credDur := time.Since(phaseStart) + if credDur > slowPhaseThreshold { + slog.Warn("slow /auth/token phase", + "phase", "credentials", + "did", did, + "authMethod", authMethod, + "duration", credDur.Round(time.Millisecond)) + } + // Validate that the user has permission for the requested access // Use the actual handle from the validated credentials, not the Basic Auth username if err := auth.ValidateAccess(did, handle, access); err != nil { @@ -333,8 +350,18 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } // Drain the gate first so a denial wins over a transient fetch error. + drainStart := time.Now() + var gateDur time.Duration if runGate { - if res := <-gateCh; res.err != nil { + res := <-gateCh + gateDur = time.Since(drainStart) + if gateDur > slowPhaseThreshold { + slog.Warn("slow /auth/token phase", + "phase", "gate", + "did", did, + "duration", gateDur.Round(time.Millisecond)) + } + if res.err != nil { slog.Info("Authorization denied", "did", did, "error", res.err) _ = errcode.ServeJSON(w, errcode.ErrorCodeDenied.WithMessage(res.err.Error())) return @@ -345,8 +372,18 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // stamp the JWT's exp from the cached expiry. They expire concurrently; // when Docker hits 401, the next /auth/token call mints both fresh. issueExp := h.issuer.expiration + var fetchDur time.Duration if runFetch { res := <-fetchCh + // Both goroutines started together, so measure from the drain start, + // not after the gate drain, to reflect the fetch's real wall time. + fetchDur = time.Since(drainStart) + if fetchDur > slowPhaseThreshold { + slog.Warn("slow /auth/token phase", + "phase", "service-auth-fetch", + "did", did, + "duration", fetchDur.Round(time.Millisecond)) + } if res.err != nil { slog.Warn("service-auth pre-mint failed", "did", did, "error", res.err) _ = errcode.ServeJSON(w, errcode.ErrorCodeUnavailable.WithMessage(fmt.Sprintf("service-auth fetch failed: %v", res.err))) @@ -373,7 +410,14 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - slog.Debug("Issued JWT token", "tokenLength", len(tokenString), "did", did, "authMethod", authMethod) + slog.Debug("Issued JWT token", + "tokenLength", len(tokenString), + "did", did, + "authMethod", authMethod, + "credentialsDur", credDur.Round(time.Millisecond), + "gateDur", gateDur.Round(time.Millisecond), + "fetchDur", fetchDur.Round(time.Millisecond), + "totalDur", time.Since(phaseStart).Round(time.Millisecond)) // Return token response now := time.Now()