From bd02e194ff62bd15a6f4d043db2240915a786355 Mon Sep 17 00:00:00 2001 From: truewhile <62226914+truewhile@users.noreply.github.com> Date: Tue, 8 Sep 2026 20:52:35 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/service/cloud115/client.go | 23 +++--- internal/service/cloud115/cloud115_test.go | 41 +++++++++++ internal/service/cloud115/open.go | 14 +++- internal/service/strm_115_oauth.go | 17 ++++- internal/service/strm_account_test.go | 85 ++++++++++++++++++++++ internal/service/strm_service.go | 65 ++++++++++++++--- 6 files changed, 219 insertions(+), 26 deletions(-) diff --git a/internal/service/cloud115/client.go b/internal/service/cloud115/client.go index 7abd788..d29e803 100644 --- a/internal/service/cloud115/client.go +++ b/internal/service/cloud115/client.go @@ -150,6 +150,7 @@ func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map if err != nil { return nil, err } + attemptedAccess := strings.TrimPrefix(req.Header.Get("Authorization"), "Bearer ") resp, err := c.HTTP.Do(req) if err != nil { @@ -218,7 +219,7 @@ func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map // refresh_token 刷新后重试一次。刷新失败或重试后仍失败才返回， // 避免长时间同步因 token 过期而整体失败。 if isTokenCode(base.Code) { - if access && c.tryRefreshTokenLocked(ctx) { + if access && c.tryRefreshTokenLocked(ctx, attemptedAccess) { continue } if access { @@ -305,21 +306,23 @@ func (c *OpenClient) doAuthJSONWithUA(ctx context.Context, method, rawURL string // tryRefreshTokenLocked 并发安全地刷新 access_token；成功返回 true（调用方 // 应使用内存中的新 token 重试原请求）。 // -// 拿到写锁后在锁内读取 oldAccess，与持锁期间的当前值对比：若已被其他 -// goroutine 刷新过则直接复用新 token，避免并发请求连环轮转消耗 115 的 -// 一次性 refresh_token。全程持写锁读写 token 字段，无 TOCTOU 窗口。 +// failedAccess 是失败请求实际携带的 token。拿到写锁后与当前 token 对比： +// 若已被其他 goroutine 刷新过则直接复用，避免并发请求连环轮转消耗 115 +// 的一次性 refresh_token。全程持写锁读写 token 字段，无 TOCTOU 窗口。 // // 对"refresh_token 本身已失效/被吊销"（IsRefreshTokenDead，如 40140114/116/119/120） // 这类不可恢复的错误直接放弃并清空内存 token（提示需重新授权）。 // 对其它失败（网络瞬时抖动、刷新接口可重试错误码等）做指数退避重试几次再放弃， // 避免同步长任务中途 token 到期时恰好撞上一个短暂的刷新失败就整体失败。 -func (c *OpenClient) tryRefreshTokenLocked(ctx context.Context) bool { +func (c *OpenClient) tryRefreshTokenLocked(ctx context.Context, failedAccess string) bool { c.tokenMu.Lock() - // 在已持有写锁内读取当前 token 作为"刷新前快照"，消除双重加锁窗口： - // 若在拿锁期间已有其他 goroutine 完成刷新，refreshTokenWhileLocked - // 内的 c.AccessToken != oldAccess 判断会立即命中并返回复用。 - oldAccess := c.AccessToken - newToken, ok := c.refreshTokenWhileLocked(ctx, oldAccess) + // 请求发出后若其他 goroutine 已经刷新完成，直接复用新 token 重试； + // 不能再次轮换一次性的 refresh_token。 + if failedAccess != "" && c.AccessToken != failedAccess { + c.tokenMu.Unlock() + return true + } + newToken, ok := c.refreshTokenWhileLocked(ctx, failedAccess) c.tokenMu.Unlock() // 回调必须在 tokenMu 释放后调用，避免上层在回调内访问客户端时死锁 if ok && newToken != nil && c.OnTokenRefreshed != nil { diff --git a/internal/service/cloud115/cloud115_test.go b/internal/service/cloud115/cloud115_test.go index a9b8019..1a83de3 100644 --- a/internal/service/cloud115/cloud115_test.go +++ b/internal/service/cloud115/cloud115_test.go @@ -140,6 +140,47 @@ func TestRefreshTokenDead(t *testing.T) { } } +func TestConcurrentTokenFailuresShareOneRefresh(t *testing.T) { + var mu sync.Mutex + refreshCalls := 0 + mockAPI(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/open/refreshToken" { + t.Errorf("unexpected path %s", r.URL.Path) + } + mu.Lock() + refreshCalls++ + mu.Unlock() + time.Sleep(50 * time.Millisecond) + _, _ = w.Write([]byte(`{"state":true,"data":{"access_token":"at-new","refresh_token":"rt-new","expires_in":7200}}`)) + }) + + client := NewOpenClient("100195129", "at-old", "rt-old") + var wg sync.WaitGroup + results := make(chan bool, 2) + for range 2 { + wg.Add(1) + go func() { + defer wg.Done() + results <- client.tryRefreshTokenLocked(context.Background(), "at-old") + }() + } + wg.Wait() + close(results) + for ok := range results { + if !ok { + t.Fatal("concurrent refresh should reuse the refreshed token") + } + } + mu.Lock() + defer mu.Unlock() + if refreshCalls != 1 { + t.Fatalf("refresh calls = %d, want 1", refreshCalls) + } + if client.CurrentAccessToken() != "at-new" { + t.Fatalf("access token = %q, want at-new", client.CurrentAccessToken()) + } +} + func TestFsListAndDownload(t *testing.T) { mockAPI(t, func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { diff --git a/internal/service/cloud115/open.go b/internal/service/cloud115/open.go index 0769ad1..cfdeb1c 100644 --- a/internal/service/cloud115/open.go +++ b/internal/service/cloud115/open.go @@ -477,24 +477,32 @@ func (c *OpenClient) GetToken(qrCode *QrCodeDataReturn) (*TokenData, error) { // RefreshToken 刷新访问令牌。 func (c *OpenClient) RefreshToken(refreshToken string) (*TokenData, error) { + c.tokenMu.Lock() if refreshToken == "" { - refreshToken = c.currentRefreshToken() + refreshToken = c.RefreshTokenStr } if refreshToken == "" { + c.tokenMu.Unlock() return nil, fmt.Errorf("没有可用的 refresh_token") } token, err := c.doRefreshToken(refreshToken) if err != nil { // refresh_token 已失效时清空内存令牌（提示需重新授权） if IsRefreshTokenDead(err) { - c.SetAuthToken("", "") + c.setAuthTokenLocked("", "") } + c.tokenMu.Unlock() return nil, err } if token.AccessToken == "" || token.RefreshToken == "" { + c.tokenMu.Unlock() return nil, fmt.Errorf("115: 刷新返回空凭证（access_token/refresh_token 缺失）") } - c.SetAuthToken(token.AccessToken, token.RefreshToken) + c.setAuthTokenLocked(token.AccessToken, token.RefreshToken) + c.tokenMu.Unlock() + if c.OnTokenRefreshed != nil { + c.OnTokenRefreshed(token.AccessToken, token.RefreshToken) + } return token, nil } diff --git a/internal/service/strm_115_oauth.go b/internal/service/strm_115_oauth.go index 78f0abd..7d5735d 100644 --- a/internal/service/strm_115_oauth.go +++ b/internal/service/strm_115_oauth.go @@ -293,7 +293,11 @@ func (s *StrmService) save115OAuthToken(ctx context.Context, session *strm115Aut acct.LastTestAt = &now acct.LastTestResult = "授权成功" acct.LastTestOK = true - return s.repo.StrmAccount.Update(ctx, acct) + if err := s.repo.StrmAccount.Update(ctx, acct); err != nil { + return err + } + s.invalidate115Provider(acct.ID) + return nil } func (s *StrmService) drop115AuthSession(sessionID string) { @@ -382,8 +386,15 @@ func (s *StrmService) refresh115TokensOnce(ctx context.Context) { } } } - client := cloud115.NewOpenClient(cfg["app_id"], cfg["access_token"], cfg["refresh_token"]) - token, err := client.RefreshToken(cfg["refresh_token"]) + provider, err := s.providerFor(ctx, acct) + if err != nil { + continue + } + openProvider, ok := provider.(interface{ OpenClient() *cloud115.OpenClient }) + if !ok || openProvider.OpenClient() == nil { + continue + } + token, err := openProvider.OpenClient().RefreshToken("") if err != nil { msg := "令牌刷新失败：" + err.Error() if cloud115.IsRefreshTokenDead(err) { diff --git a/internal/service/strm_account_test.go b/internal/service/strm_account_test.go index b140319..fb76034 100644 --- a/internal/service/strm_account_test.go +++ b/internal/service/strm_account_test.go @@ -10,6 +10,7 @@ import ( "github.com/truewhile/MeBox/internal/config" "github.com/truewhile/MeBox/internal/model" "github.com/truewhile/MeBox/internal/repository" + "github.com/truewhile/MeBox/internal/service/cloud115" ) func TestStrmAccountConfigPreviewOf(t *testing.T) { @@ -71,6 +72,90 @@ func TestUpdateStrmAccountMergesConfigWithoutClearingSecrets(t *testing.T) { } } +func TestProviderForReuses115ClientUntilCredentialsChange(t *testing.T) { + svc := testStrmService(t) + ctx := context.Background() + acct, err := svc.CreateStrmAccount(ctx, "115", model.StrmProvider115, map[string]string{ + "app_id": "100195129", + "access_token": "at-1", + "refresh_token": "rt-1", + }) + if err != nil { + t.Fatalf("create account: %v", err) + } + + first, err := svc.providerFor(ctx, acct) + if err != nil { + t.Fatalf("first provider: %v", err) + } + second, err := svc.providerFor(ctx, acct) + if err != nil { + t.Fatalf("second provider: %v", err) + } + if first != second { + t.Fatal("115 provider should be shared per account") + } + + updated, err := svc.UpdateStrmAccount(ctx, acct.ID, "", nil, map[string]string{ + "access_token": "at-2", + "refresh_token": "rt-2", + }) + if err != nil { + t.Fatalf("update account: %v", err) + } + third, err := svc.providerFor(ctx, updated) + if err != nil { + t.Fatalf("provider after credential update: %v", err) + } + if first == third { + t.Fatal("credential update must invalidate the shared provider") + } + + oldClient := first.(interface{ OpenClient() *cloud115.OpenClient }).OpenClient() + oldClient.OnTokenRefreshed("at-stale", "rt-stale") + fresh, err := svc.repo.StrmAccount.FindByID(ctx, acct.ID) + if err != nil || fresh == nil { + t.Fatalf("reload account: %v", err) + } + freshCfg, err := svc.strmAccountConfig(fresh) + if err != nil { + t.Fatalf("decode refreshed account: %v", err) + } + if freshCfg["access_token"] != "at-2" || freshCfg["refresh_token"] != "rt-2" { + t.Fatalf("stale client overwrote new credentials: %#v", freshCfg) + } +} + +func TestPersist115TokensKeepsSharedClient(t *testing.T) { + svc := testStrmService(t) + ctx := context.Background() + acct, err := svc.CreateStrmAccount(ctx, "115", model.StrmProvider115, map[string]string{ + "app_id": "100195129", + "access_token": "at-1", + "refresh_token": "rt-1", + }) + if err != nil { + t.Fatalf("create account: %v", err) + } + first, err := svc.providerFor(ctx, acct) + if err != nil { + t.Fatalf("first provider: %v", err) + } + + svc.persist115Tokens(acct.ID, "at-2", "rt-2") + fresh, err := svc.repo.StrmAccount.FindByID(ctx, acct.ID) + if err != nil || fresh == nil { + t.Fatalf("reload account: %v", err) + } + second, err := svc.providerFor(ctx, fresh) + if err != nil { + t.Fatalf("second provider: %v", err) + } + if first != second { + t.Fatal("automatic token persistence must keep the in-memory shared client") + } +} + func TestDeleteStrmAccountCascadesEmbyMounts(t *testing.T) { ctx := context.Background() db := newServiceTestDB(t, &model.StrmAccount{}, &model.EmbyMount{}, &model.StrmSyncPath{}) diff --git a/internal/service/strm_service.go b/internal/service/strm_service.go index fb85946..5bdfddd 100644 --- a/internal/service/strm_service.go +++ b/internal/service/strm_service.go @@ -95,6 +95,9 @@ type StrmService struct { oauthSessions map[string]*strm115AuthSession wafUntil time.Time // 115 风控/限流熔断截止时间（由 mu 保护） + providerMu sync.Mutex + provider115Cache map[string]cloud.Provider // account ID -> shared provider/OpenClient + downloadSem115 chan struct{} // 115 换直链+下载并发上限（风控兜底） downloadSemDAV chan struct{} // WebDAV/OpenList/CloudDrive2 元数据下载并发上限 downloadSemOnce sync.Once @@ -158,15 +161,16 @@ func (s *StrmService) releaseDownloadSlot(provider string) { // NewStrmService constructs the STRM service. func NewStrmService(cfg *config.Config, log *zap.Logger, repos *repository.Container, crypto *CryptoService) *StrmService { return &StrmService{ - log: log, - repo: repos, - cfg: cfg, - crypto: crypto, - http: &http.Client{Timeout: 90 * time.Second}, - stopCh: make(chan struct{}), - baseCtx: context.Background(), - running: map[string]context.CancelFunc{}, - oauthSessions: map[string]*strm115AuthSession{}, + log: log, + repo: repos, + cfg: cfg, + crypto: crypto, + http: &http.Client{Timeout: 90 * time.Second}, + stopCh: make(chan struct{}), + baseCtx: context.Background(), + running: map[string]context.CancelFunc{}, + oauthSessions: map[string]*strm115AuthSession{}, + provider115Cache: map[string]cloud.Provider{}, } } @@ -409,6 +413,9 @@ func (s *StrmService) UpdateStrmAccount(ctx context.Context, id, name string, en if err := s.repo.StrmAccount.Update(ctx, acct); err != nil { return nil, err } + if acct.Provider == model.StrmProvider115 && len(config) > 0 { + s.invalidate115Provider(acct.ID) + } return acct, nil } @@ -426,6 +433,7 @@ func (s *StrmService) DeleteStrmAccount(ctx context.Context, id string) error { if err := s.repo.StrmAccount.Delete(ctx, id); err != nil { return err } + s.invalidate115Provider(id) // 级联清理远程 Emby 挂载：否则留下孤儿挂载，挂载计数/列表仍会显示。 // 账号已删，挂载清理失败只记日志，不让删除请求报错。 if _, err := s.repo.EmbyMount.DeleteByAccountID(ctx, id); err != nil && s.log != nil { @@ -464,6 +472,23 @@ func (s *StrmService) ListAccounts(ctx context.Context) ([]model.StrmAccount, er // providerFor 依据账号配置构建网盘驱动。 func (s *StrmService) providerFor(ctx context.Context, acct *model.StrmAccount) (cloud.Provider, error) { + if acct != nil && acct.Provider == model.StrmProvider115 { + s.providerMu.Lock() + defer s.providerMu.Unlock() + if provider := s.provider115Cache[acct.ID]; provider != nil { + return provider, nil + } + provider, err := s.newProvider(ctx, acct) + if err != nil { + return nil, err + } + s.provider115Cache[acct.ID] = provider + return provider, nil + } + return s.newProvider(ctx, acct) +} + +func (s *StrmService) newProvider(ctx context.Context, acct *model.StrmAccount) (cloud.Provider, error) { cfg, err := s.strmAccountConfig(acct) if err != nil { return nil, err @@ -482,13 +507,33 @@ func (s *StrmService) providerFor(ctx context.Context, acct *model.StrmAccount) // refresh_token 再刷（一次性轮转），两者互相作废，最终把有效账号 // 标成“授权已失效”。 if oc, ok := provider.(interface{ OpenClient() *cloud115.OpenClient }); ok { - oc.OpenClient().OnTokenRefreshed = func(accessToken, refreshToken string) { + client := oc.OpenClient() + client.OnTokenRefreshed = func(accessToken, refreshToken string) { + // 账号重新授权/修改凭据后，旧客户端可能仍有在途请求。旧请求 + // 刷新的令牌不能覆盖新授权写入的凭据。 + if !s.isCurrent115Client(acct.ID, client) { + return + } s.persist115Tokens(acct.ID, accessToken, refreshToken) } } return provider, nil } +func (s *StrmService) invalidate115Provider(accountID string) { + s.providerMu.Lock() + delete(s.provider115Cache, accountID) + s.providerMu.Unlock() +} + +func (s *StrmService) isCurrent115Client(accountID string, client *cloud115.OpenClient) bool { + s.providerMu.Lock() + defer s.providerMu.Unlock() + provider := s.provider115Cache[accountID] + openProvider, ok := provider.(interface{ OpenClient() *cloud115.OpenClient }) + return ok && openProvider.OpenClient() == client +} + // ─── 全局设置 ────────────────────────────────────────────────────────────────── // GetStrmSettings 返回全局 strm 设置（含默认值）。