diff --git a/pkg/connector/auth_recovery.go b/pkg/connector/auth_recovery.go index 031c572..dc08999 100644 --- a/pkg/connector/auth_recovery.go +++ b/pkg/connector/auth_recovery.go @@ -113,7 +113,7 @@ func (lc *LineClient) recoverClientAfterAuthError(ctx context.Context, failedCli Msg("Retrying LINE request after response from stale access token") } lc.recoverMu.Unlock() - return newLineAPIClient(currentToken), nil + return lc.newClientWithToken(currentToken), nil } if line.IsLoggedOut(err) { diff --git a/pkg/connector/client.go b/pkg/connector/client.go index a881521..97113f3 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -68,6 +68,8 @@ type LineClient struct { deletedChatsMu sync.Mutex deletedChats map[string]string + obsTokenCache line.OBSTokenCache + tokenMu sync.RWMutex recoverMu sync.Mutex recoverTime time.Time @@ -222,7 +224,13 @@ func (lc *LineClient) claimForcedLogoutState() bool { } func (lc *LineClient) newClient() *line.Client { - return newLineAPIClient(lc.getAccessToken()) + return lc.newClientWithToken(lc.getAccessToken()) +} + +func (lc *LineClient) newClientWithToken(token string) *line.Client { + client := newLineAPIClient(token) + client.OBSTokenCache = &lc.obsTokenCache + return client } func (lc *LineClient) avatarFromPicturePath(picturePath string) *bridgev2.Avatar { @@ -314,7 +322,7 @@ func (lc *LineClient) refreshAndSave(ctx context.Context) error { // Rotating the main access token invalidates any OBS token derived from it, // so drop the cached one — the next OBS call will mint a fresh one. - line.InvalidateOBSTokenCache() + lc.obsTokenCache.Invalidate() meta := lc.UserLogin.Metadata.(*UserLoginMetadata) meta.AccessToken = accessToken @@ -346,7 +354,7 @@ func (lc *LineClient) markLoggedOutByOtherClient(ctx context.Context, err error) func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err error) { if lc.UserLogin == nil { lc.invalidateAccessToken() - line.InvalidateOBSTokenCache() + lc.obsTokenCache.Invalidate() lc.cancelActiveRun() return } @@ -361,7 +369,7 @@ func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err } sendState := lc.claimForcedLogoutState() lc.invalidateAccessToken() - line.InvalidateOBSTokenCache() + 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." invalidSenderKey := line.IsInvalidSenderKey(err) @@ -854,7 +862,7 @@ func (lc *LineClient) tryLogin(ctx context.Context) error { } lc.setTokens(accessToken, refreshToken) lc.Mid = staged.Mid - line.InvalidateOBSTokenCache() + lc.obsTokenCache.Invalidate() if mgr != nil { lc.missingE2EEKeyMarked = false } diff --git a/pkg/line/client.go b/pkg/line/client.go index bd27fc0..888faff 100644 --- a/pkg/line/client.go +++ b/pkg/line/client.go @@ -42,10 +42,11 @@ var ( ) type Client struct { - LoginAttempt *LoginAttempt - HTTPClient *http.Client - OBSClient *http.Client - AccessToken string + LoginAttempt *LoginAttempt + HTTPClient *http.Client + OBSClient *http.Client + AccessToken string + OBSTokenCache *OBSTokenCache channelTokenMu sync.Mutex channelTokenCache map[string]cachedChannelAccessToken @@ -84,9 +85,10 @@ type cachedChannelAccessToken struct { func NewClient(token string) *Client { return &Client{ - HTTPClient: &http.Client{Timeout: rpcClientTimeout}, - OBSClient: &http.Client{}, - AccessToken: token, + HTTPClient: &http.Client{Timeout: rpcClientTimeout}, + OBSClient: &http.Client{}, + AccessToken: token, + OBSTokenCache: &OBSTokenCache{}, } } diff --git a/pkg/line/methods.go b/pkg/line/methods.go index 801e965..a61a2a3 100644 --- a/pkg/line/methods.go +++ b/pkg/line/methods.go @@ -11,26 +11,39 @@ import ( "time" ) -var ( - obsTokenMu sync.Mutex - obsTokenCache string - obsTokenExpiry time.Time -) - const obsTokenBuffer = 30 * time.Second const defaultChannelTokenLifetime = 5 * time.Minute -// InvalidateOBSTokenCache clears the cached OBS access token. The OBS token is -// derived from the main LINE access token; when the latter is rotated (refresh -// or re-login) any previously-issued OBS token is invalidated server-side, but -// the cache here would keep handing it out until its original TTL expires. -// Callers must invoke this after any successful re-authentication. -func InvalidateOBSTokenCache() { - obsTokenMu.Lock() - obsTokenCache = "" - obsTokenExpiry = time.Time{} - obsTokenMu.Unlock() +type OBSTokenCache struct { + mu sync.Mutex + token string + expiry time.Time +} + +func (c *OBSTokenCache) get() (string, bool) { + c.mu.Lock() + defer c.mu.Unlock() + if c.token != "" && time.Now().Before(c.expiry) { + return c.token, true + } + return "", false +} + +func (c *OBSTokenCache) set(token string, expiry time.Time) { + c.mu.Lock() + c.token = token + c.expiry = expiry + c.mu.Unlock() +} + +// Invalidate clears the cached OBS access token. The OBS token is derived from +// the main LINE access token; when the latter is rotated (refresh or re-login) +// any previously-issued OBS token is invalidated server-side, but the cache +// would keep handing it out until its original TTL expires. Callers must invoke +// this after any successful re-authentication. +func (c *OBSTokenCache) Invalidate() { + c.set("", time.Time{}) } // LoginV2 performs the loginV2 RPC call to authenticate a user @@ -691,13 +704,9 @@ func (c *Client) GetLastOpRevisionContext(ctx context.Context) (int64, error) { // this token is used to encrypt images, videos, and files uploaded to LINE's OBS storage func (c *Client) AcquireEncryptedAccessToken() (string, error) { - obsTokenMu.Lock() - if obsTokenCache != "" && time.Now().Before(obsTokenExpiry) { - cached := obsTokenCache - obsTokenMu.Unlock() + if cached, ok := c.OBSTokenCache.get(); ok { return cached, nil } - obsTokenMu.Unlock() // 2 = FeatureType::OBS_Authorization. resp, err := c.callRPC("TalkService", "acquireEncryptedAccessToken", 2) @@ -726,10 +735,7 @@ func (c *Client) AcquireEncryptedAccessToken() (string, error) { token := parts[1] if expirySec, err := strconv.Atoi(parts[0]); err == nil && expirySec > 0 { - obsTokenMu.Lock() - obsTokenCache = token - obsTokenExpiry = time.Now().Add(time.Duration(expirySec)*time.Second - obsTokenBuffer) - obsTokenMu.Unlock() + c.OBSTokenCache.set(token, time.Now().Add(time.Duration(expirySec)*time.Second-obsTokenBuffer)) } return token, nil diff --git a/pkg/line/obs_test.go b/pkg/line/obs_test.go index 5716da2..2c1ec95 100644 --- a/pkg/line/obs_test.go +++ b/pkg/line/obs_test.go @@ -16,20 +16,10 @@ type observedOBSRequest struct { headers http.Header } -func installCachedOBSToken(t *testing.T) { - t.Helper() - obsTokenMu.Lock() - oldToken := obsTokenCache - oldExpiry := obsTokenExpiry - obsTokenCache = "obs-token" - obsTokenExpiry = time.Now().Add(time.Hour) - obsTokenMu.Unlock() - t.Cleanup(func() { - obsTokenMu.Lock() - obsTokenCache = oldToken - obsTokenExpiry = oldExpiry - obsTokenMu.Unlock() - }) +func newClientWithCachedOBSToken(token string) *Client { + client := NewClient(token) + client.OBSTokenCache.set("obs-token", time.Now().Add(time.Hour)) + return client } func obsResponse(status int, body string) *http.Response { @@ -41,10 +31,8 @@ func obsResponse(status int, body string) *http.Response { } func TestDownloadOBSPlainMatchesChromeRequestFlow(t *testing.T) { - installCachedOBSToken(t) - var requests []observedOBSRequest - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { requests = append(requests, observedOBSRequest{ @@ -156,10 +144,8 @@ func TestDownloadOBSPublicResourceClassifiesMissingObject(t *testing.T) { } func TestDownloadOBSResourceUsesReceiveServiceAndSID(t *testing.T) { - installCachedOBSToken(t) - var requests []observedOBSRequest - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { requests = append(requests, observedOBSRequest{ @@ -200,10 +186,8 @@ func TestDownloadOBSResourceUsesReceiveServiceAndSID(t *testing.T) { } func TestDownloadOBSPreflightsBaseObjectBeforeTID(t *testing.T) { - installCachedOBSToken(t) - var requests []observedOBSRequest - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { requests = append(requests, observedOBSRequest{path: req.URL.Path, query: req.URL.RawQuery}) @@ -238,10 +222,8 @@ func TestDownloadOBSPreflightsBaseObjectBeforeTID(t *testing.T) { } func TestDownloadOBSEncryptedIncludesTalkMeta(t *testing.T) { - installCachedOBSToken(t) - var talkMetaHeaders []string - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { talkMetaHeaders = append(talkMetaHeaders, req.Header.Get("X-Talk-Meta")) @@ -267,13 +249,12 @@ func TestDownloadOBSEncryptedIncludesTalkMeta(t *testing.T) { } func TestDownloadOBSWaitsForObjectEncoding(t *testing.T) { - installCachedOBSToken(t) oldDelay := obsRetryDelay obsRetryDelay = 0 t.Cleanup(func() { obsRetryDelay = oldDelay }) var paths []string - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { paths = append(paths, req.URL.Path) @@ -306,9 +287,7 @@ func TestDownloadOBSWaitsForObjectEncoding(t *testing.T) { } func TestDownloadOBSClassifiesMissingObject(t *testing.T) { - installCachedOBSToken(t) - - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { return obsResponse(http.StatusOK, `{"status":"notexist"}`), nil @@ -322,9 +301,7 @@ func TestDownloadOBSClassifiesMissingObject(t *testing.T) { } func TestDownloadOBSDoesNotCallObjectInfoHTTP404Expired(t *testing.T) { - installCachedOBSToken(t) - - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { return obsResponse(http.StatusNotFound, "not found"), nil @@ -341,9 +318,7 @@ func TestDownloadOBSDoesNotCallObjectInfoHTTP404Expired(t *testing.T) { } func TestDownloadOBSClassifiesObjectInfoUnauthorized(t *testing.T) { - installCachedOBSToken(t) - - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { return obsResponse(http.StatusUnauthorized, "unauthorized"), nil @@ -357,10 +332,8 @@ func TestDownloadOBSClassifiesObjectInfoUnauthorized(t *testing.T) { } func TestDownloadOBSDoesNotCallPostPreflight404Expired(t *testing.T) { - installCachedOBSToken(t) - var requests int - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { requests++ @@ -381,13 +354,12 @@ func TestDownloadOBSDoesNotCallPostPreflight404Expired(t *testing.T) { } func TestDownloadOBSClassifiesEncodingRetryExhaustion(t *testing.T) { - installCachedOBSToken(t) oldDelay := obsRetryDelay obsRetryDelay = 0 t.Cleanup(func() { obsRetryDelay = oldDelay }) var requests int - client := NewClient("line-token") + client := newClientWithCachedOBSToken("line-token") client.OBSClient = &http.Client{ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { requests++