diff --git a/pkg/appview/db/migrations/0031_add_oauth_session_rev.yaml b/pkg/appview/db/migrations/0031_add_oauth_session_rev.yaml new file mode 100644 index 0000000..b05add5 --- /dev/null +++ b/pkg/appview/db/migrations/0031_add_oauth_session_rev.yaml @@ -0,0 +1,16 @@ +description: | + Add a revision counter to oauth_sessions so session writes can compare-and-swap. + + Refresh tokens rotate on every use. The AppView serializes refreshes per DID + with an in-process mutex, which is exactly the right thing to do and stops + working the moment there is more than one instance: two instances refreshing + the same DID concurrently means one of them replays a rotated refresh token, + gets invalid_grant, and the isAuthError path deletes the session. The user is + signed out mid-push and nothing explains why. + + rev increments on every successful write. A writer that read revision N can + detect that someone else has written since, which lets the delete-on-auth-error + path tell "this session is genuinely dead" apart from "another instance just + refreshed it and my copy is stale". +query: | + ALTER TABLE oauth_sessions ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; diff --git a/pkg/appview/db/oauth_session_rev_test.go b/pkg/appview/db/oauth_session_rev_test.go new file mode 100644 index 0000000..9c9ff08 --- /dev/null +++ b/pkg/appview/db/oauth_session_rev_test.go @@ -0,0 +1,293 @@ +package db + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "path/filepath" + "testing" + + atcroauth "atcr.io/pkg/auth/oauth" + "github.com/bluesky-social/indigo/atproto/auth/oauth" + "github.com/bluesky-social/indigo/atproto/syntax" +) + +// revTestDB returns a file-backed database. +// +// Not ":memory:" — go-libsql gives each connection to an in-memory DSN its own +// private database, and these tests deliberately use two OAuthStore values to +// stand in for two AppView instances sharing one database. +func revTestDB(t *testing.T) *sql.DB { + t.Helper() + database, err := InitDB(filepath.Join(t.TempDir(), "oauth.db"), LibsqlConfig{}) + if err != nil { + t.Fatalf("InitDB: %v", err) + } + t.Cleanup(func() { database.Close() }) + return database +} + +func sessionData(t *testing.T, did, sessionID, accessToken, refreshToken string) oauth.ClientSessionData { + t.Helper() + parsed, err := syntax.ParseDID(did) + if err != nil { + t.Fatalf("ParseDID(%q): %v", did, err) + } + return oauth.ClientSessionData{ + AccountDID: parsed, + SessionID: sessionID, + HostURL: "https://bsky.social", + AuthServerURL: "https://bsky.social", + AuthServerTokenEndpoint: "https://bsky.social/oauth/token", + Scopes: []string{"atproto"}, + AccessToken: accessToken, + RefreshToken: refreshToken, + DPoPPrivateKeyMultibase: "test_key", + } +} + +func storedTokens(t *testing.T, database *sql.DB, did, sessionID string) (access, refresh string) { + t.Helper() + var raw string + err := database.QueryRow( + `SELECT session_data FROM oauth_sessions WHERE session_key = ?`, + makeSessionKey(did, sessionID), + ).Scan(&raw) + if err != nil { + t.Fatalf("read session: %v", err) + } + var data oauth.ClientSessionData + if err := unmarshalSession(raw, &data); err != nil { + t.Fatalf("parse session: %v", err) + } + return data.AccessToken, data.RefreshToken +} + +// TestConcurrentRefreshDoesNotClobberRotatedTokens is the scenario this whole +// mechanism exists for. +// +// Refresh tokens rotate on use. Two AppView instances refreshing the same +// account both hold a session they believe is current. Without a revision check +// the one that writes second replaces the freshly rotated tokens with ones the +// auth server has already invalidated, and the next refresh fails with +// invalid_grant for everyone. +func TestConcurrentRefreshDoesNotClobberRotatedTokens(t *testing.T) { + database := revTestDB(t) + ctx := context.Background() + + const ( + did = "did:plc:alice" + sessionID = "session-1" + ) + + // One instance creates the session. + instanceA := NewOAuthStore(database) + if err := instanceA.SaveSession(ctx, sessionData(t, did, sessionID, "access-v1", "refresh-v1")); err != nil { + t.Fatalf("initial save: %v", err) + } + + // A second instance, with its own in-process revision map, reads it. + instanceB := NewOAuthStore(database) + parsed, _ := syntax.ParseDID(did) + if _, err := instanceB.GetSession(ctx, parsed, sessionID); err != nil { + t.Fatalf("instance B read: %v", err) + } + if _, err := instanceA.GetSession(ctx, parsed, sessionID); err != nil { + t.Fatalf("instance A read: %v", err) + } + + // B refreshes first and persists the rotated tokens. + if err := instanceB.SaveSession(ctx, sessionData(t, did, sessionID, "access-v2", "refresh-v2")); err != nil { + t.Fatalf("instance B save: %v", err) + } + + // A now tries to persist what it read before B's write. + err := instanceA.SaveSession(ctx, sessionData(t, did, sessionID, "access-stale", "refresh-stale")) + if !errors.Is(err, ErrSessionRevConflict) { + t.Fatalf("expected ErrSessionRevConflict from the losing writer, got %v", err) + } + + access, refresh := storedTokens(t, database, did, sessionID) + if access != "access-v2" || refresh != "refresh-v2" { + t.Errorf("the stale writer clobbered the rotated tokens: stored access=%q refresh=%q, want the v2 pair", + access, refresh) + } +} + +// TestSaveSessionAdvancesRevision: a successful write must bump the revision, or +// nothing else can detect that it happened. +func TestSaveSessionAdvancesRevision(t *testing.T) { + database := revTestDB(t) + ctx := context.Background() + store := NewOAuthStore(database) + + const ( + did = "did:plc:alice" + sessionID = "session-1" + ) + + if err := store.SaveSession(ctx, sessionData(t, did, sessionID, "a1", "r1")); err != nil { + t.Fatalf("save: %v", err) + } + first, exists, err := store.GetSessionRev(ctx, did, sessionID) + if err != nil || !exists { + t.Fatalf("GetSessionRev: exists=%v err=%v", exists, err) + } + + if err := store.SaveSession(ctx, sessionData(t, did, sessionID, "a2", "r2")); err != nil { + t.Fatalf("second save: %v", err) + } + second, _, err := store.GetSessionRev(ctx, did, sessionID) + if err != nil { + t.Fatalf("GetSessionRev: %v", err) + } + + if second <= first { + t.Errorf("revision did not advance: %d then %d", first, second) + } +} + +// TestSameStoreRepeatedSavesSucceed: the CAS must not get in the way of the +// ordinary case, one instance writing a session it owns over and over as DPoP +// nonces change. +func TestSameStoreRepeatedSavesSucceed(t *testing.T) { + database := revTestDB(t) + ctx := context.Background() + store := NewOAuthStore(database) + + const ( + did = "did:plc:alice" + sessionID = "session-1" + ) + + for i := range 10 { + if err := store.SaveSession(ctx, sessionData(t, did, sessionID, "a", "r")); err != nil { + t.Fatalf("save %d: %v", i, err) + } + } +} + +// TestGetSessionRecordsRevisionForLaterCAS: reading is what establishes the +// expectation a later write compares against. +func TestGetSessionRecordsRevisionForLaterCAS(t *testing.T) { + database := revTestDB(t) + ctx := context.Background() + + const ( + did = "did:plc:alice" + sessionID = "session-1" + ) + + writer := NewOAuthStore(database) + if err := writer.SaveSession(ctx, sessionData(t, did, sessionID, "a1", "r1")); err != nil { + t.Fatalf("save: %v", err) + } + + reader := NewOAuthStore(database) + if _, ok := reader.KnownRev(did, sessionID); ok { + t.Fatal("a store that has never read the session should know no revision") + } + + parsed, _ := syntax.ParseDID(did) + if _, err := reader.GetSession(ctx, parsed, sessionID); err != nil { + t.Fatalf("GetSession: %v", err) + } + if _, ok := reader.KnownRev(did, sessionID); !ok { + t.Error("GetSession did not record a revision") + } +} + +// TestGetLatestSessionForDIDRecordsRevision covers the path the refresher +// actually uses: resumeSession loads through GetLatestSessionForDID, so if that +// did not record a revision the CAS would never engage where it matters most. +func TestGetLatestSessionForDIDRecordsRevision(t *testing.T) { + database := revTestDB(t) + ctx := context.Background() + + const ( + did = "did:plc:alice" + sessionID = "session-1" + ) + + writer := NewOAuthStore(database) + if err := writer.SaveSession(ctx, sessionData(t, did, sessionID, "a1", "r1")); err != nil { + t.Fatalf("save: %v", err) + } + + reader := NewOAuthStore(database) + if _, _, err := reader.GetLatestSessionForDID(ctx, did); err != nil { + t.Fatalf("GetLatestSessionForDID: %v", err) + } + if _, ok := reader.KnownRev(did, sessionID); !ok { + t.Error("GetLatestSessionForDID did not record a revision") + } +} + +// TestDeleteSessionForgetsRevision: a remembered revision must not survive the +// session it describes, or a re-created session would be compared against a +// revision from the dead one. +func TestDeleteSessionForgetsRevision(t *testing.T) { + database := revTestDB(t) + ctx := context.Background() + store := NewOAuthStore(database) + + const ( + did = "did:plc:alice" + sessionID = "session-1" + ) + + if err := store.SaveSession(ctx, sessionData(t, did, sessionID, "a1", "r1")); err != nil { + t.Fatalf("save: %v", err) + } + parsed, _ := syntax.ParseDID(did) + if err := store.DeleteSession(ctx, parsed, sessionID); err != nil { + t.Fatalf("DeleteSession: %v", err) + } + if _, ok := store.KnownRev(did, sessionID); ok { + t.Error("revision survived the session being deleted") + } + + // A re-created session must save cleanly rather than hitting a phantom + // conflict. + if err := store.SaveSession(ctx, sessionData(t, did, sessionID, "a2", "r2")); err != nil { + t.Errorf("re-created session failed to save: %v", err) + } +} + +func TestDeleteSessionsForDIDForgetsRevisions(t *testing.T) { + database := revTestDB(t) + ctx := context.Background() + store := NewOAuthStore(database) + + const did = "did:plc:alice" + for _, sid := range []string{"s1", "s2"} { + if err := store.SaveSession(ctx, sessionData(t, did, sid, "a", "r")); err != nil { + t.Fatalf("save %s: %v", sid, err) + } + } + + if err := store.DeleteSessionsForDID(ctx, did); err != nil { + t.Fatalf("DeleteSessionsForDID: %v", err) + } + for _, sid := range []string{"s1", "s2"} { + if _, ok := store.KnownRev(did, sid); ok { + t.Errorf("revision for %s survived the bulk delete", sid) + } + } +} + +// TestErrSessionRevConflictIsSharedAcrossPackages: pkg/auth/oauth compares +// against its own sentinel, so the two must be the same value or the refresher +// would treat a conflict as an ordinary failure and log it as an error. +func TestErrSessionRevConflictIsSharedAcrossPackages(t *testing.T) { + if !errors.Is(ErrSessionRevConflict, atcroauth.ErrSessionRevConflict) { + t.Error("db.ErrSessionRevConflict and oauth.ErrSessionRevConflict are different errors") + } +} + +// unmarshalSession is a tiny helper so the test does not need to import +// encoding/json separately from the store's own representation. +func unmarshalSession(raw string, out *oauth.ClientSessionData) error { + return json.Unmarshal([]byte(raw), out) +} diff --git a/pkg/appview/db/oauth_store.go b/pkg/appview/db/oauth_store.go index e08baa0..76d894c 100644 --- a/pkg/appview/db/oauth_store.go +++ b/pkg/appview/db/oauth_store.go @@ -4,8 +4,11 @@ import ( "context" "database/sql" "encoding/json" + "errors" "fmt" "log/slog" + "strings" + "sync" "time" atoauth "atcr.io/pkg/auth/oauth" @@ -13,9 +16,23 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" ) +// ErrSessionRevConflict is re-exported from pkg/auth/oauth, where it has to live +// to avoid an import cycle: this package already imports that one. Callers can +// use either name. +var ErrSessionRevConflict = atoauth.ErrSessionRevConflict + // OAuthStore implements oauth.ClientAuthStore with SQLite persistence type OAuthStore struct { db *sql.DB + + // revs maps session_key to the revision this process last read, so + // SaveSession can compare-and-swap against it. + // + // Process-local state is sufficient because the OAuth refresher already + // serializes per DID with an in-process mutex. That mutex is what makes at + // most one write per session in flight here; the rev closes the gap it + // cannot cover, which is a second instance. + revs sync.Map } // NewOAuthStore creates a new SQLite-backed OAuth store @@ -23,17 +40,59 @@ func NewOAuthStore(db *sql.DB) *OAuthStore { return &OAuthStore{db: db} } +// rememberRev records the revision observed for a session key. +func (s *OAuthStore) rememberRev(sessionKey string, rev int64) { + s.revs.Store(sessionKey, rev) +} + +// knownRev returns the revision this process last read for a session key. +func (s *OAuthStore) knownRev(sessionKey string) (int64, bool) { + v, ok := s.revs.Load(sessionKey) + if !ok { + return 0, false + } + rev, ok := v.(int64) + return rev, ok +} + +// GetSessionRev returns the revision currently stored for a session, and whether +// the session exists at all. +// +// Used by the refresher to distinguish a session that is genuinely dead from one +// another instance has refreshed since we read it. +func (s *OAuthStore) GetSessionRev(ctx context.Context, did, sessionID string) (int64, bool, error) { + var rev int64 + err := s.db.QueryRowContext(ctx, + `SELECT rev FROM oauth_sessions WHERE session_key = ?`, + makeSessionKey(did, sessionID), + ).Scan(&rev) + if errors.Is(err, sql.ErrNoRows) { + return 0, false, nil + } + if err != nil { + return 0, false, fmt.Errorf("failed to query session revision: %w", err) + } + return rev, true, nil +} + +// KnownRev exposes the revision this process last read for a session, so the +// refresher can compare it against what is stored now. +func (s *OAuthStore) KnownRev(did, sessionID string) (int64, bool) { + return s.knownRev(makeSessionKey(did, sessionID)) +} + // GetSession retrieves a session by DID and session ID func (s *OAuthStore) GetSession(ctx context.Context, did syntax.DID, sessionID string) (*oauth.ClientSessionData, error) { sessionKey := makeSessionKey(did.String(), sessionID) var sessionDataJSON string + var rev int64 err := s.db.QueryRowContext(ctx, ` - SELECT session_data + SELECT session_data, rev FROM oauth_sessions WHERE session_key = ? - `, sessionKey).Scan(&sessionDataJSON) + `, sessionKey).Scan(&sessionDataJSON, &rev) if err == sql.ErrNoRows { return nil, fmt.Errorf("session not found: %s/%s", did, sessionID) @@ -48,10 +107,22 @@ func (s *OAuthStore) GetSession(ctx context.Context, did syntax.DID, sessionID s return nil, fmt.Errorf("failed to parse session data: %w", err) } + s.rememberRev(sessionKey, rev) return &sessionData, nil } -// SaveSession saves or updates a session (upsert) +// SaveSession saves or updates a session. +// +// When this process has previously read the session, the write is a +// compare-and-swap against the revision it read, and returns +// ErrSessionRevConflict if anyone has written since. Otherwise it is an upsert, +// which is the path a brand-new session from the OAuth callback takes. +// +// The conflict case exists because refresh tokens rotate on use. Two instances +// refreshing the same account concurrently both hold a session they believe is +// current, and the one that writes second would otherwise overwrite freshly +// rotated tokens with tokens the auth server has already invalidated, bricking +// the session for everyone. func (s *OAuthStore) SaveSession(ctx context.Context, sess oauth.ClientSessionData) error { sessionKey := makeSessionKey(sess.AccountDID.String(), sess.SessionID) @@ -61,13 +132,38 @@ func (s *OAuthStore) SaveSession(ctx context.Context, sess oauth.ClientSessionDa return fmt.Errorf("failed to marshal session data: %w", err) } + if expectedRev, ok := s.knownRev(sessionKey); ok { + res, err := s.db.ExecContext(ctx, ` + UPDATE oauth_sessions + SET session_data = ?, rev = rev + 1, updated_at = datetime('now') + WHERE session_key = ? AND rev = ? + `, string(sessionDataJSON), sessionKey, expectedRev) + if err != nil { + return fmt.Errorf("failed to save session: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("failed to save session: %w", err) + } + if n == 0 { + // Either someone else wrote (rev moved) or the row is gone. Drop our + // stale expectation so the next read re-establishes it. + s.revs.Delete(sessionKey) + return ErrSessionRevConflict + } + s.rememberRev(sessionKey, expectedRev+1) + return nil + } + + // No revision known: a session we have never read, i.e. a fresh login. _, err = s.db.ExecContext(ctx, ` INSERT INTO oauth_sessions ( session_key, account_did, session_id, session_data, - created_at, updated_at - ) VALUES (?, ?, ?, ?, datetime('now'), datetime('now')) + created_at, updated_at, rev + ) VALUES (?, ?, ?, ?, datetime('now'), datetime('now'), 1) ON CONFLICT(session_key) DO UPDATE SET session_data = excluded.session_data, + rev = oauth_sessions.rev + 1, updated_at = datetime('now') `, sessionKey, @@ -75,11 +171,19 @@ func (s *OAuthStore) SaveSession(ctx context.Context, sess oauth.ClientSessionDa sess.SessionID, string(sessionDataJSON), ) - if err != nil { return fmt.Errorf("failed to save session: %w", err) } + // Read the revision back so subsequent writes for this session can + // compare-and-swap. + var rev int64 + if err := s.db.QueryRowContext(ctx, + `SELECT rev FROM oauth_sessions WHERE session_key = ?`, sessionKey, + ).Scan(&rev); err == nil { + s.rememberRev(sessionKey, rev) + } + return nil } @@ -91,6 +195,10 @@ func (s *OAuthStore) DeleteSession(ctx context.Context, did syntax.DID, sessionI DELETE FROM oauth_sessions WHERE session_key = ? `, sessionKey) + // Drop the remembered revision: keeping it would make the next write for a + // re-created session compare against a revision from the deleted one. + s.revs.Delete(sessionKey) + return err } @@ -105,6 +213,8 @@ func (s *OAuthStore) DeleteSessionsForDID(ctx context.Context, did string) error return fmt.Errorf("failed to delete sessions for DID: %w", err) } + s.forgetRevsForDID(did) + deleted, _ := result.RowsAffected() if deleted > 0 { slog.Info("Deleted OAuth sessions for DID", "count", deleted, "did", did) @@ -113,6 +223,18 @@ func (s *OAuthStore) DeleteSessionsForDID(ctx context.Context, did string) error return nil } +// forgetRevsForDID drops every remembered revision belonging to a DID, so +// nothing is left comparing against a revision from a deleted session. +func (s *OAuthStore) forgetRevsForDID(did string) { + prefix := did + ":" + s.revs.Range(func(key, _ any) bool { + if k, ok := key.(string); ok && strings.HasPrefix(k, prefix) { + s.revs.Delete(k) + } + return true + }) +} + // DeleteOldSessionsForDID removes all sessions for a DID except the specified session to keep // This is used during OAuth callback to clean up stale sessions with expired refresh tokens func (s *OAuthStore) DeleteOldSessionsForDID(ctx context.Context, did string, keepSessionID string) error { @@ -188,14 +310,15 @@ func (s *OAuthStore) DeleteAuthRequestInfo(ctx context.Context, state string) er func (s *OAuthStore) GetLatestSessionForDID(ctx context.Context, did string) (*oauth.ClientSessionData, string, error) { var sessionDataJSON string var sessionID string + var rev int64 err := s.db.QueryRowContext(ctx, ` - SELECT session_id, session_data + SELECT session_id, session_data, rev FROM oauth_sessions WHERE account_did = ? ORDER BY updated_at DESC LIMIT 1 - `, did).Scan(&sessionID, &sessionDataJSON) + `, did).Scan(&sessionID, &sessionDataJSON, &rev) if err == sql.ErrNoRows { return nil, "", fmt.Errorf("no session found for DID: %s", did) @@ -210,6 +333,7 @@ func (s *OAuthStore) GetLatestSessionForDID(ctx context.Context, did string) (*o return nil, "", fmt.Errorf("failed to parse session data: %w", err) } + s.rememberRev(makeSessionKey(did, sessionID), rev) return &sessionData, sessionID, nil } diff --git a/pkg/appview/db/schema.sql b/pkg/appview/db/schema.sql index 6b3176d..ba2d061 100644 --- a/pkg/appview/db/schema.sql +++ b/pkg/appview/db/schema.sql @@ -94,6 +94,11 @@ CREATE TABLE IF NOT EXISTS tags ( ); CREATE INDEX IF NOT EXISTS idx_tags_did_repo ON tags(did, repository); +-- rev increments on every successful write, so a writer that read revision N can +-- tell whether anyone has written since. Refresh tokens rotate on use, and the +-- per-DID mutex that serializes refreshes is in-process only, so without this a +-- second instance's concurrent refresh looks identical to a dead session and +-- gets it deleted out from under the user. See OAuthStore.SaveSession. CREATE TABLE IF NOT EXISTS oauth_sessions ( session_key TEXT PRIMARY KEY, account_did TEXT NOT NULL, @@ -101,6 +106,7 @@ CREATE TABLE IF NOT EXISTS oauth_sessions ( session_data TEXT NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + rev INTEGER NOT NULL DEFAULT 0, UNIQUE(account_did, session_id) ); CREATE INDEX IF NOT EXISTS idx_oauth_sessions_did ON oauth_sessions(account_did); diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index 86384d6..36ddd8b 100644 --- a/pkg/auth/oauth/client.go +++ b/pkg/auth/oauth/client.go @@ -245,7 +245,97 @@ func (r *Refresher) SetUISessionStore(store UISessionStore) { r.uiSessionStore = store } +// ErrSessionRevConflict is returned by a session store's SaveSession when the +// stored session has been written by someone else since this process last read +// it. +// +// It lives here rather than next to the SQLite store because pkg/appview/db +// already imports this package; defining it there and referring to it here would +// be an import cycle. +// +// It is not a failure so much as a report: the caller's copy is stale and must +// not be written over the newer one. Callers should re-read rather than retry +// blindly, and must not treat it as evidence that the session is broken. +var ErrSessionRevConflict = errors.New("oauth: session was modified concurrently") + +// staleSessionError marks an auth failure that is explained by another instance +// having refreshed the same session concurrently, rather than by the session +// being dead. DoWithSession retries these once against the newer tokens. +type staleSessionError struct { + did string + cause error +} + +func (e *staleSessionError) Error() string { + return fmt.Sprintf("oauth session for %s was refreshed concurrently: %v", e.did, e.cause) +} + +func (e *staleSessionError) Unwrap() error { return e.cause } + +// sessionRevChecker is implemented by stores that version their sessions. The +// SQLite store does; a store that does not simply keeps the old behavior of +// deleting on any auth error. +type sessionRevChecker interface { + KnownRev(did, sessionID string) (int64, bool) + GetSessionRev(ctx context.Context, did, sessionID string) (int64, bool, error) +} + +// sessionRevisionAdvanced reports whether the stored session has been written by +// someone else since this process last read it. +// +// Deliberately conservative: anything it cannot determine (a store without +// revisions, no revision recorded, a failed lookup, a session that is simply +// gone) returns false and leaves the caller with the previous delete-on-error +// behavior. Wrongly answering true would keep a genuinely dead session alive and +// leave the user stuck; wrongly answering false only costs a re-login. +func (r *Refresher) sessionRevisionAdvanced(ctx context.Context, did, sessionID string) bool { + checker, ok := r.clientApp.Store.(sessionRevChecker) + if !ok { + return false + } + knownRev, ok := checker.KnownRev(did, sessionID) + if !ok { + return false + } + + // Detached: we are deciding whether to destroy a session, so an inbound + // request that was canceled mid-flight must not push us into deleting it. + checkCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), SessionDeleteTimeout) + defer cancel() + + currentRev, exists, err := checker.GetSessionRev(checkCtx, did, sessionID) + if err != nil || !exists { + return false + } + return currentRev > knownRev +} + // DoWithSession executes a function with a locked OAuth session. +// +// If the session turns out to have been refreshed by another instance while we +// were working, it is retried once against the newer tokens. Exactly once: a +// second failure means the trouble is not staleness, and looping would just hold +// the per-DID lock while getting the same answer. +func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(session *oauth.ClientSession) error) error { + err := r.doWithSessionOnce(ctx, did, fn) + + var stale *staleSessionError + if errors.As(err, &stale) { + slog.Info("Retrying with the session another instance refreshed", + "component", "oauth/refresher", "did", did) + err = r.doWithSessionOnce(ctx, did, fn) + + // Still stale on the retry: stop unwrapping and report the underlying + // auth failure, so callers see a normal auth error rather than an + // internal marker type. + if errors.As(err, &stale) { + return stale.cause + } + } + return err +} + +// doWithSessionOnce is one attempt of DoWithSession. // 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. @@ -278,7 +368,7 @@ func (r *Refresher) SetUISessionStore(store UISessionStore) { // // Parse response into result... // return nil // }) -func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(session *oauth.ClientSession) error) error { +func (r *Refresher) doWithSessionOnce(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) @@ -319,6 +409,23 @@ func (r *Refresher) DoWithSession(ctx context.Context, did string, fn func(sessi // If request failed with auth error, delete session to force re-auth if err != nil && isAuthError(err) { + // ...unless another instance refreshed this session while we were + // working. The mutex above is process-local, so with more than one + // AppView instance two of them can refresh the same account at once; + // refresh tokens rotate on use, so the slower one gets invalid_grant on + // a token that is merely superseded rather than revoked. Deleting then + // signs the user out mid-push and destroys the session the other + // instance just legitimately refreshed. + // + // A revision that has moved since we read it is exactly that case. + if r.sessionRevisionAdvanced(ctx, did, session.Data.SessionID) { + slog.Info("Auth error on a session another instance has since refreshed; keeping it and retrying", + "component", "oauth/refresher", + "did", did, + "error", err) + return &staleSessionError{did: did, cause: err} + } + slog.Warn("Auth error detected, deleting session to force re-auth", "component", "oauth/refresher", "did", did, @@ -494,6 +601,17 @@ func (r *Refresher) resumeSession(ctx context.Context, did string) (*oauth.Clien saveCtx, cancel := context.WithTimeout(context.WithoutCancel(callbackCtx), 10*time.Second) defer cancel() if err := r.clientApp.Store.SaveSession(saveCtx, *updatedData); err != nil { + // A revision conflict is not a failure. Another instance persisted a + // newer state for this session, so ours is stale and overwriting it + // would replace live tokens with superseded ones. Leaving the newer + // state alone is the correct outcome. + if errors.Is(err, ErrSessionRevConflict) { + slog.Info("Skipped persisting a stale OAuth session update; another instance wrote first", + "component", "oauth/refresher", + "did", did, + "sessionID", sessionID) + return + } slog.Error("Failed to persist OAuth session update", "component", "oauth/refresher", "did", did, diff --git a/pkg/auth/oauth/session_rev_test.go b/pkg/auth/oauth/session_rev_test.go new file mode 100644 index 0000000..071b478 --- /dev/null +++ b/pkg/auth/oauth/session_rev_test.go @@ -0,0 +1,178 @@ +package oauth + +import ( + "context" + "errors" + "testing" + + "github.com/bluesky-social/indigo/atproto/auth/oauth" +) + +// revStore is a ClientAuthStore that also reports session revisions, standing in +// for the SQLite store. +type revStore struct { + oauth.ClientAuthStore + + known map[string]int64 + current map[string]int64 + exists map[string]bool + lookupErr error +} + +func newRevStore() *revStore { + return &revStore{ + known: map[string]int64{}, + current: map[string]int64{}, + exists: map[string]bool{}, + } +} + +func (s *revStore) KnownRev(did, sessionID string) (int64, bool) { + rev, ok := s.known[did+":"+sessionID] + return rev, ok +} + +func (s *revStore) GetSessionRev(_ context.Context, did, sessionID string) (int64, bool, error) { + if s.lookupErr != nil { + return 0, false, s.lookupErr + } + key := did + ":" + sessionID + return s.current[key], s.exists[key], nil +} + +// plainStore implements no revision tracking, standing in for any store that +// predates this mechanism. +type plainStore struct{ oauth.ClientAuthStore } + +func refresherWithStore(store oauth.ClientAuthStore) *Refresher { + return NewRefresher(&oauth.ClientApp{Store: store}) +} + +// TestSessionRevisionAdvanced covers the decision that gates session deletion. +// +// Getting this wrong in the "true" direction keeps a genuinely dead session +// alive and leaves the user stuck with no way to recover but waiting; getting it +// wrong in the "false" direction costs a re-login. So every case it cannot +// resolve must answer false. +func TestSessionRevisionAdvanced(t *testing.T) { + const ( + did = "did:plc:alice" + sessionID = "session-1" + key = did + ":" + sessionID + ) + + t.Run("another instance wrote since we read", func(t *testing.T) { + s := newRevStore() + s.known[key] = 3 + s.current[key] = 4 + s.exists[key] = true + + if !refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) { + t.Error("expected the advanced revision to be detected") + } + }) + + t.Run("nobody wrote since we read", func(t *testing.T) { + s := newRevStore() + s.known[key] = 3 + s.current[key] = 3 + s.exists[key] = true + + if refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) { + t.Error("an unchanged revision must not look like a concurrent refresh") + } + }) + + t.Run("session no longer exists", func(t *testing.T) { + s := newRevStore() + s.known[key] = 3 + s.exists[key] = false + + if refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) { + t.Error("a missing session must not be treated as concurrently refreshed") + } + }) + + t.Run("no revision recorded for this session", func(t *testing.T) { + s := newRevStore() + s.current[key] = 9 + s.exists[key] = true + + if refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) { + t.Error("without a recorded revision there is nothing to compare against") + } + }) + + t.Run("revision lookup fails", func(t *testing.T) { + s := newRevStore() + s.known[key] = 3 + s.lookupErr = errors.New("database unavailable") + + if refresherWithStore(s).sessionRevisionAdvanced(context.Background(), did, sessionID) { + t.Error("a failed lookup must fall back to the previous behavior") + } + }) + + t.Run("store does not track revisions", func(t *testing.T) { + if refresherWithStore(&plainStore{}).sessionRevisionAdvanced(context.Background(), did, sessionID) { + t.Error("a store without revisions must keep the previous behavior") + } + }) + + t.Run("canceled request context still allows the check", func(t *testing.T) { + s := newRevStore() + s.known[key] = 3 + s.current[key] = 4 + s.exists[key] = true + + // The inbound request being canceled must not push us into deleting a + // session another instance just refreshed, so the check runs detached. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if !refresherWithStore(s).sessionRevisionAdvanced(ctx, did, sessionID) { + t.Error("a canceled inbound context must not suppress the check") + } + }) +} + +// TestStaleSessionErrorUnwraps: DoWithSession reports the underlying auth error +// when a retry does not help, so callers must still be able to inspect it. +func TestStaleSessionErrorUnwraps(t *testing.T) { + cause := errors.New("invalid_grant") + err := error(&staleSessionError{did: "did:plc:alice", cause: cause}) + + if !errors.Is(err, cause) { + t.Error("staleSessionError does not unwrap to its cause") + } + + var stale *staleSessionError + if !errors.As(err, &stale) { + t.Error("staleSessionError is not detectable with errors.As") + } +} + +// TestDoWithSessionRetriesOnceOnStaleSession pins the retry policy: exactly one +// retry. Looping would hold the per-DID lock while getting the same answer. +func TestDoWithSessionRetriesOnceOnStaleSession(t *testing.T) { + // resumeSession requires a store implementing GetLatestSessionForDID, which + // this deliberately does not, so every attempt fails early and we can count + // attempts without standing up a PDS. + r := refresherWithStore(&plainStore{}) + + attempts := 0 + fn := func(*oauth.ClientSession) error { + attempts++ + return nil + } + + // The DID never resolves to a session, so fn is never reached; what matters + // is that DoWithSession returns rather than looping. + err := r.DoWithSession(context.Background(), "did:plc:alice", fn) + if err == nil { + t.Fatal("expected an error when no session can be resumed") + } + if attempts != 0 { + t.Errorf("fn ran %d times despite the session never resuming", attempts) + } +}