diff --git a/pkg/connector/auth_recovery.go b/pkg/connector/auth_recovery.go index dc08999..f08204d 100644 --- a/pkg/connector/auth_recovery.go +++ b/pkg/connector/auth_recovery.go @@ -2,7 +2,10 @@ package connector import ( "context" + "errors" "fmt" + "net" + "net/http" "github.com/beeper/line/pkg/line" ) @@ -65,6 +68,27 @@ func callLineWithRecovery[T any](ctx context.Context, client *line.Client, deps return client, res, err } +func isTransientLineError(err error) bool { + // An unanswered PIN prompt times out like a network request, but retrying + // would only send the user another prompt. + if errors.Is(err, errLinePINNotVerified) { + return false + } + var netErr net.Error + if errors.As(err, &netErr) { + return true + } + statusCode := line.HTTPStatusCode(err) + return statusCode >= http.StatusInternalServerError || statusCode == http.StatusTooManyRequests +} + +func isRecoveryRejected(err error) bool { + return !isTransientLineError(err) && + !errors.Is(err, context.Canceled) && + !errors.Is(err, errLineSessionInvalidated) && + !errors.Is(err, errLineClientSuperseded) +} + func (lc *LineClient) isTokenError(err error) bool { if line.IsNoUsableE2EEGroupKey(err) || line.IsNoUsableE2EEPublicKey(err) { return false @@ -153,6 +177,15 @@ func (lc *LineClient) recoverClientAfterAuthError(ctx context.Context, failedCli return lc.newClient(), nil } +func (lc *LineClient) recoverSessionAfterAuthError(ctx context.Context, failedClient *line.Client, err error) (*line.Client, error) { + recoveryToken := lc.getAccessToken() + recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, err) + if errRecover != nil && ctx.Err() == nil && isRecoveryRejected(errRecover) { + lc.markRecoveryRejected(ctx, recoveryToken, errRecover) + } + return recoveredClient, errRecover +} + func (lc *LineClient) callLine(ctx context.Context, call func(*line.Client) error) (*line.Client, error) { return lc.callLineUsing(ctx, nil, call) } diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 4369df3..4635b48 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -24,6 +24,7 @@ var ( errLineSessionInvalidated = errors.New("LINE session invalidated by another client") errLineClientSuperseded = errors.New("LINE client was superseded") errLineQRLoginRequired = errors.New("LINE requires a new QR scan. Reconnect in Beeper to continue") + errLinePINNotVerified = errors.New("PIN verification failed") ) func init() { @@ -81,6 +82,9 @@ type LineClient struct { tokenMu sync.RWMutex recoverMu sync.Mutex recoverTime time.Time + recoverErr error + recoverRetryAt time.Time + recoverBackoff reconnectBackoff recoveryStopped bool missingE2EEKeyMu sync.Mutex missingE2EEKeyMarked bool @@ -356,10 +360,24 @@ func (lc *LineClient) markLoggedOutByOtherClient(ctx context.Context, err error) lc.markLoggedOutByOtherClientLocked(ctx, err) } -// markLoggedOutByOtherClientLocked applies the persistent forced-logout -// transition. The caller must hold recoverMu so token rotation and invalidation -// cannot interleave. func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err error) { + lc.invalidateSessionLocked(ctx, err, false) +} + +// markRecoveryRejected invalidates the session unless another recovery already +// replaced the token the rejected recovery started from. +func (lc *LineClient) markRecoveryRejected(ctx context.Context, recoveryToken string, err error) { + lc.recoverMu.Lock() + defer lc.recoverMu.Unlock() + if lc.isSessionInvalidated() || lc.getAccessToken() != recoveryToken { + return + } + lc.invalidateSessionLocked(ctx, err, true) +} + +// invalidateSessionLocked applies the persistent forced-logout transition. The +// caller must hold recoverMu so token rotation and invalidation cannot interleave. +func (lc *LineClient) invalidateSessionLocked(ctx context.Context, err error, expired bool) { if lc.UserLogin == nil { lc.invalidateAccessToken() lc.obsTokenCache.Invalidate() @@ -380,6 +398,9 @@ func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err lc.obsTokenCache.Invalidate() stateError := status.BridgeStateErrorCode("line-logged-out") stateMessage := "LINE logged this Chrome Extension session out because another LINE client connected. Click Reconnect in Beeper to reconnect LINE." + if expired { + stateError, stateMessage = "line-token-expired", "" + } invalidSenderKey := line.IsInvalidSenderKey(err) needsFullE2EELogin := invalidSenderKey lc.missingE2EEKeyMu.Lock() @@ -397,7 +418,7 @@ func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err stateError = "line-e2ee-key-missing" stateMessage = lineMissingE2EEKeyMessage } - lc.saveSessionInvalidated(ctx) + lc.saveSessionInvalidated(ctx, expired) if sendState && lc.UserLogin.Bridge != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("LINE session invalidated; marking login bad credentials") } @@ -407,6 +428,7 @@ func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err Error: stateError, Message: stateMessage, UserAction: status.UserActionRelogin, + Info: map[string]any{"go_error": err.Error()}, }) } lc.cancelActiveRun() @@ -459,7 +481,7 @@ func (lc *LineClient) sendSSEDisconnectedIfCurrent(ctx context.Context, err erro return true } -func (lc *LineClient) saveSessionInvalidated(ctx context.Context) { +func (lc *LineClient) saveSessionInvalidated(ctx context.Context, expired bool) { if lc.UserLogin == nil || lc.UserLogin.UserLogin == nil { return } @@ -469,6 +491,7 @@ func (lc *LineClient) saveSessionInvalidated(ctx context.Context) { } meta.AccessToken = "" meta.SessionInvalidated = true + meta.SessionExpired = expired if err := lc.UserLogin.Save(ctx); err != nil && lc.UserLogin.Bridge != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to save LINE session invalidation") } @@ -518,7 +541,6 @@ func (lc *LineClient) applyRefreshedLoginE2EEKeys(meta *UserLoginMetadata, res * } // recoverToken attempts to restore a valid session by refreshing, then re-logging in. -// Returns nil on success. On failure the caller should send StateBadCredentials. func (lc *LineClient) recoverToken(ctx context.Context) error { return lc.recoverTokenWith(ctx, lc.refreshAndSave, lc.tryLogin) } @@ -537,7 +559,7 @@ func (lc *LineClient) recoverTokenWith( return nil } else if ctx.Err() != nil { return ctx.Err() - } else if lc.isLoggedOut(err) || errors.Is(err, errLineSessionInvalidated) || errors.Is(err, errLineClientSuperseded) || errors.Is(err, context.Canceled) { + } else if lc.isLoggedOut(err) || isTransientLineError(err) || errors.Is(err, errLineSessionInvalidated) || errors.Is(err, errLineClientSuperseded) || errors.Is(err, context.Canceled) { return err } lc.UserLogin.Bridge.Log.Info().Msg("Refresh failed, attempting re-login with stored credentials...") @@ -566,8 +588,15 @@ func (lc *LineClient) runTokenRecovery(ctx context.Context, recover func(context if !lc.recoverTime.IsZero() && time.Since(lc.recoverTime) < recentTokenRecoveryWindow { return nil } + if time.Now().Before(lc.recoverRetryAt) { + return fmt.Errorf("waiting to retry LINE token recovery: %w", lc.recoverErr) + } if err := recover(ctx); err != nil { + if ctx.Err() == nil { + lc.recoverErr = err + lc.recoverRetryAt = time.Now().Add(lc.recoverBackoff.next()) + } return err } if ctx.Err() != nil { @@ -579,6 +608,9 @@ func (lc *LineClient) runTokenRecovery(ctx context.Context, recover func(context if lc.isSessionInvalidated() { return errLineSessionInvalidated } + lc.recoverBackoff.reset() + lc.recoverErr = nil + lc.recoverRetryAt = time.Time{} lc.recoverTime = time.Now() return nil } @@ -588,14 +620,11 @@ func (lc *LineClient) Connect(ctx context.Context) { if !started { return } - defer lc.wg.Done() + handedOff := false workersStarted := false defer func() { - if !workersStarted { - if lc.E2EE != nil { - lc.E2EE.Close() - } - run.cancel() + if !handedOff { + lc.finishStartup(run, workersStarted) } }() @@ -629,7 +658,10 @@ func (lc *LineClient) Connect(ctx context.Context) { } if !lc.hasAccessToken() { if lc.isSessionInvalidated() { - lc.markLoggedOutByOtherClient(ctx, errLineSessionInvalidated) + meta, _ := lc.UserLogin.Metadata.(*UserLoginMetadata) + lc.recoverMu.Lock() + lc.invalidateSessionLocked(ctx, errLineSessionInvalidated, meta != nil && meta.SessionExpired) + lc.recoverMu.Unlock() return } if err := lc.runTokenRecovery(ctx, lc.tryLogin); err != nil { @@ -651,31 +683,76 @@ func (lc *LineClient) Connect(ctx context.Context) { } // Verify the token is still valid before proceeding - if err := lc.ensureValidToken(ctx); err != nil { - if ctx.Err() != nil { + if tokenErr := lc.ensureValidToken(ctx); tokenErr != nil { + if ctx.Err() == nil && lc.isLoggedOut(tokenErr) { + lc.markLoggedOutByOtherClient(ctx, tokenErr) + } + if ctx.Err() != nil || lc.isSessionInvalidated() { return } - if lc.isLoggedOut(err) { - lc.markLoggedOutByOtherClient(ctx, err) + if lc.shouldWaitForToken(tokenErr) { + if !lc.sendSSEDisconnectedIfCurrent(ctx, tokenErr) { + return + } + lc.UserLogin.Bridge.Log.Warn().Err(tokenErr).Msg("Failed to verify LINE token, retrying before starting the session") + // Startup work below runs once per connection, so it waits for a working + // token instead of running against a failing one. Connect itself returns so + // other logins aren't held up. + handedOff = true + go func() { + lc.finishStartup(run, lc.waitForValidToken(ctx) && lc.startSession(ctx)) + }() return } - lc.UserLogin.BridgeState.Send(status.BridgeState{ - StateEvent: status.StateBadCredentials, - Error: "line-token-expired", - Info: map[string]any{"go_error": err.Error()}, - UserAction: status.UserActionRelogin, - }) - return + lc.UserLogin.Bridge.Log.Warn().Err(tokenErr).Msg("Failed to verify LINE token, starting anyway") + } + workersStarted = lc.startSession(ctx) +} + +func (lc *LineClient) shouldWaitForToken(err error) bool { + return lc.isTokenError(err) || isTransientLineError(err) +} + +func (lc *LineClient) finishStartup(run *lineClientRun, workersStarted bool) { + if !workersStarted { + if lc.E2EE != nil { + lc.E2EE.Close() + } + run.cancel() } + lc.wg.Done() +} + +func (lc *LineClient) waitForValidToken(ctx context.Context) bool { + var backoff reconnectBackoff + for waitForSSEReconnect(ctx, backoff.next()) { + err := lc.ensureValidToken(ctx) + switch { + case err == nil: + return true + case ctx.Err() != nil: + return false + case lc.isLoggedOut(err): + lc.markLoggedOutByOtherClient(ctx, err) + return false + case lc.isSessionInvalidated(): + return false + case !lc.shouldWaitForToken(err): + lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to verify LINE token, starting anyway") + return true + } + lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Still failing to verify LINE token, retrying") + } + return false +} + +func (lc *LineClient) startSession(ctx context.Context) bool { // LINE Chrome disables mobile notifications while its session is active by // default. Explicitly clear that setting once per bridge connection so the // user's phone continues receiving notifications. lc.configurePhoneNotifications(ctx) - if ctx.Err() != nil { - return - } - if !lc.sendConnectedStateIfCurrent(ctx) { - return + if ctx.Err() != nil || !lc.sendConnectedStateIfCurrent(ctx) { + return false } // Initialize E2EE manager and load keys @@ -732,14 +809,28 @@ func (lc *LineClient) Connect(ctx context.Context) { // sender's existing membership look like a fresh join. lc.syncChatsNow(ctx) if ctx.Err() != nil { - return + return false } lc.wg.Add(3) go lc.syncDMChats(ctx) go lc.prefetchMessages(ctx) go lc.pollLoop(ctx) - workersStarted = true + return true +} + +func verifyReloginProfile(ctx context.Context, accessToken string) (*line.Profile, error) { + for attempt := 1; ; attempt++ { + profile, err := newLineAPIClient(accessToken).GetProfileContext(ctx) + if err == nil || attempt >= 3 || !isTransientLineError(err) { + return profile, err + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(time.Duration(attempt) * 2 * time.Second): + } + } } func (lc *LineClient) tryLogin(ctx context.Context) error { @@ -801,7 +892,7 @@ func (lc *LineClient) tryLogin(ctx context.Context) error { waitClient.LoginAttempt = res.Attempt waitRes, err := waitClient.WaitForLogin(res.Verifier, res.NoE2EE) if err != nil { - return fmt.Errorf("PIN verification failed: %w", err) + return fmt.Errorf("%w: %w", errLinePINNotVerified, err) } if waitRes.AuthToken == "" { return fmt.Errorf("PIN verification completed but no auth token received") @@ -826,7 +917,7 @@ func (lc *LineClient) tryLogin(ctx context.Context) error { return errLineClientSuperseded } if res.Attempt != nil { - profile, err := newLineAPIClient(accessToken).GetProfileContext(ctx) + profile, err := verifyReloginProfile(ctx, accessToken) if err != nil { return fmt.Errorf("verify re-login account: %w", err) } @@ -973,20 +1064,16 @@ func (lc *LineClient) refreshLoginE2EEKeys(res *line.LoginResult, meta *UserLogi } func (lc *LineClient) ensureValidToken(ctx context.Context) error { - _, _, err := callLineResult(lc, ctx, func(client *line.Client) (*line.Profile, error) { - return getProfileWithToken(ctx, client.AccessToken) + _, _, err := callLineWithRecovery(ctx, nil, lineCallDeps[*line.Profile]{ + newClient: func() *line.Client { return lc.newClient() }, + recover: lc.recoverSessionAfterAuthError, + isAuthError: lc.isTokenError, + beforeCall: func() error { return lc.checkLineCall(ctx) }, + call: func(client *line.Client) (*line.Profile, error) { + return getProfileWithToken(ctx, client.AccessToken) + }, }) - if err == nil { - return nil - } - if ctx.Err() != nil { - return ctx.Err() - } - if line.IsAuthError(err) { - return err - } - lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("GetProfile failed with non-auth error, continuing anyway") - return nil + return err } func (lc *LineClient) Disconnect() { diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index ccf3e06..037f908 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -130,6 +130,7 @@ type UserLoginMetadata struct { AccessToken string `json:"access_token"` RefreshToken string `json:"refresh_token,omitempty"` SessionInvalidated bool `json:"session_invalidated,omitempty"` + SessionExpired bool `json:"session_expired,omitempty"` Email string `json:"email,omitempty"` Password string `json:"password,omitempty"` Certificate string `json:"certificate,omitempty"` diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index c74ad6c..cf5e981 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "math/rand/v2" "slices" "sort" "strconv" @@ -21,7 +22,6 @@ import ( "maunium.net/go/mautrix/bridgev2/database" "maunium.net/go/mautrix/bridgev2/networkid" "maunium.net/go/mautrix/bridgev2/simplevent" - "maunium.net/go/mautrix/bridgev2/status" "maunium.net/go/mautrix/event" "github.com/beeper/line/pkg/line" @@ -38,6 +38,7 @@ const ( groupPortalCreateWait = 30 * time.Second beeperExcludeFromTimelineKey = "com.beeper.exclude_from_timeline" defaultReceiveAuthProbeInterval = 150 * time.Second + maxReconnectDelay = 5 * time.Minute ) var ( @@ -1590,11 +1591,25 @@ func isReceiveAuthProbeDue(receiveCtx context.Context, nextProbeAt time.Time) bo return errors.Is(context.Cause(receiveCtx), errReceiveAuthProbeDue) || !receiveAuthProbeNow().Before(nextProbeAt) } -func waitForSSEReconnect(ctx context.Context) bool { - if sseReconnectDelay <= 0 { +type reconnectBackoff struct { + delay time.Duration +} + +func (b *reconnectBackoff) next() time.Duration { + b.delay = min(max(b.delay*2, sseReconnectDelay), maxReconnectDelay) + half := b.delay / 2 + return half + rand.N(half+1) +} + +func (b *reconnectBackoff) reset() { + b.delay = 0 +} + +func waitForSSEReconnect(ctx context.Context, delay time.Duration) bool { + if delay <= 0 { return ctx.Err() == nil } - timer := time.NewTimer(sseReconnectDelay) + timer := time.NewTimer(delay) defer timer.Stop() select { case <-ctx.Done(): @@ -1639,7 +1654,9 @@ func (lc *LineClient) pollLoop(ctx context.Context) { } sseDisconnected := false + var backoff reconnectBackoff handler := func(eventType, data string) { + backoff.reset() if sseDisconnected && lc.sendSSEConnectedIfCurrent(ctx) { sseDisconnected = false } @@ -1728,21 +1745,14 @@ func (lc *LineClient) pollLoop(ctx context.Context) { if probeAndReschedule() { return } - if !waitForSSEReconnect(receiveCtx) { - continue - } - continue + } else if line.IsAuthError(err) && lc.handleReceiveAuthError(ctx, client, err) { + return } - - if line.IsAuthError(err) { - if lc.handleReceiveAuthError(ctx, client, err) { - return - } - } else if !sseDisconnected && lc.sendSSEDisconnectedIfCurrent(ctx, err) { + if !sseDisconnected && lc.sendSSEDisconnectedIfCurrent(ctx, err) { sseDisconnected = true } } - if !waitForSSEReconnect(receiveCtx) { + if !waitForSSEReconnect(receiveCtx, backoff.next()) { continue } } @@ -1775,7 +1785,7 @@ func (lc *LineClient) handleReceiveAuthProbe(ctx context.Context) bool { return lc.handleReceiveAuthError(ctx, probeClient, probeErr) } - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, probeClient, probeErr) + recoveredClient, errRecover := lc.recoverSessionAfterAuthError(ctx, probeClient, probeErr) if errRecover != nil { if errors.Is(errRecover, errLineSessionInvalidated) || errors.Is(errRecover, errLineClientSuperseded) || ctx.Err() != nil { return true @@ -1803,7 +1813,7 @@ func (lc *LineClient) handleReceiveAuthProbe(ctx context.Context) bool { // the detailed forced-logout envelope before deciding whether recovery is safe. func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient *line.Client, err error) bool { if lc.isLoggedOut(err) { - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, err) + recoveredClient, errRecover := lc.recoverSessionAfterAuthError(ctx, failedClient, err) if errRecover != nil { return true } @@ -1820,7 +1830,7 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient * return true } if lc.isLoggedOut(profileErr) { - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, profileClient, profileErr) + recoveredClient, errRecover := lc.recoverSessionAfterAuthError(ctx, profileClient, profileErr) if errRecover != nil { return true } @@ -1830,7 +1840,7 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient * return false } - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, err) + recoveredClient, errRecover := lc.recoverSessionAfterAuthError(ctx, failedClient, err) if recoveredClient != nil { return false } @@ -1838,24 +1848,16 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient * return true } - if ctx.Err() != nil { + if ctx.Err() != nil || lc.isSessionInvalidated() { return true } if errors.Is(errRecover, errLineSessionInvalidated) || errors.Is(errRecover, errLineClientSuperseded) { return true } if lc.UserLogin != nil && lc.UserLogin.Bridge != nil { - lc.UserLogin.Bridge.Log.Error().Err(errRecover).Msg("Failed to recover session, stopping poll loop") - } - if lc.UserLogin != nil && lc.UserLogin.BridgeState != nil { - lc.UserLogin.BridgeState.Send(status.BridgeState{ - StateEvent: status.StateBadCredentials, - Error: "line-logged-out", - Message: "LINE session was invalidated (logged out by another client). Please re-authenticate the bridge.", - UserAction: status.UserActionRelogin, - }) + lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover session, retrying") } - return true + return false } func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) {