diff --git a/internal/auth/claude/utls_transport.go b/internal/auth/claude/utls_transport.go index 1df71f1e..0686c509 100644 --- a/internal/auth/claude/utls_transport.go +++ b/internal/auth/claude/utls_transport.go @@ -6,10 +6,10 @@ import ( "net" "net/http" "strings" - "sync" "time" tls "github.com/refraction-networking/utls" + internalcache "github.com/router-for-me/CLIProxyAPI/v7/internal/cache" "github.com/router-for-me/CLIProxyAPI/v7/internal/httpwire" "github.com/router-for-me/CLIProxyAPI/v7/sdk/config" "github.com/router-for-me/CLIProxyAPI/v7/sdk/proxyutil" @@ -64,7 +64,10 @@ func claudeOAuthRequestHeaderOrder(method, requestTarget string) []string { // claudeOAuthSessionCacheCapacity bounds one proxy's TLS session cache. The // OAuth control plane only talks to platform.claude.com and api.anthropic.com, // so a small cache covers every reachable server. -const claudeOAuthSessionCacheCapacity = 8 +const ( + claudeOAuthSessionCacheCapacity = 8 + claudeOAuthProxySessionCacheCapacity = 64 +) // claudeOAuthSessionCaches keys one session cache per effective proxy URL. // @@ -76,15 +79,15 @@ const claudeOAuthSessionCacheCapacity = 8 // server rather than a credential, and connections are already pooled per proxy // on the inference plane, so this adds no new cross-credential linkage. -var claudeOAuthSessionCaches sync.Map +var claudeOAuthSessionCaches = internalcache.NewBoundedLRU[string, tls.ClientSessionCache]( + claudeOAuthProxySessionCacheCapacity, + nil, +) func claudeOAuthSessionCache(proxyURL string) tls.ClientSessionCache { - if cached, ok := claudeOAuthSessionCaches.Load(proxyURL); ok { - return cached.(tls.ClientSessionCache) - } - created := tls.NewLRUClientSessionCache(claudeOAuthSessionCacheCapacity) - actual, _ := claudeOAuthSessionCaches.LoadOrStore(proxyURL, created) - return actual.(tls.ClientSessionCache) + return claudeOAuthSessionCaches.GetOrAdd(proxyURL, func() tls.ClientSessionCache { + return tls.NewLRUClientSessionCache(claudeOAuthSessionCacheCapacity) + }) } // newClaudeOAuthTLSConfig builds the uTLS config for one control-plane dial. diff --git a/internal/auth/claude/utls_transport_test.go b/internal/auth/claude/utls_transport_test.go index cda48f6e..b125c655 100644 --- a/internal/auth/claude/utls_transport_test.go +++ b/internal/auth/claude/utls_transport_test.go @@ -145,6 +145,20 @@ func TestClaudeOAuthTLSResumptionIsWireSafe(t *testing.T) { } } +func TestClaudeOAuthSessionCacheBoundsProxyCardinality(t *testing.T) { + firstProxy := "http://127.0.0.1:31000" + first := claudeOAuthSessionCache(firstProxy) + for index := 1; index <= claudeOAuthProxySessionCacheCapacity; index++ { + claudeOAuthSessionCache("http://127.0.0.1:" + strconv.Itoa(31000+index)) + } + if got := claudeOAuthSessionCaches.Len(); got > claudeOAuthProxySessionCacheCapacity { + t.Fatalf("OAuth session caches = %d, want at most %d", got, claudeOAuthProxySessionCacheCapacity) + } + if recreated := claudeOAuthSessionCache(firstProxy); recreated == first { + t.Fatal("least recently used OAuth proxy session cache was not evicted") + } +} + func TestClaudeOAuthRequestHeaderOrderMatchesNative220Capture(t *testing.T) { t.Parallel() diff --git a/internal/cache/bounded_lru.go b/internal/cache/bounded_lru.go new file mode 100644 index 00000000..458853be --- /dev/null +++ b/internal/cache/bounded_lru.go @@ -0,0 +1,83 @@ +package cache + +import ( + "container/list" + "sync" +) + +type boundedLRUEntry[K comparable, V any] struct { + key K + value V +} + +// BoundedLRU stores at most capacity values and evicts the least recently used +// value when a new key crosses the bound. The optional eviction callback runs +// after the cache lock is released. +type BoundedLRU[K comparable, V any] struct { + mu sync.Mutex + capacity int + entries map[K]*list.Element + order *list.List + onEvict func(K, V) +} + +func NewBoundedLRU[K comparable, V any](capacity int, onEvict func(K, V)) *BoundedLRU[K, V] { + if capacity < 1 { + capacity = 1 + } + return &BoundedLRU[K, V]{ + capacity: capacity, + entries: make(map[K]*list.Element, capacity), + order: list.New(), + onEvict: onEvict, + } +} + +// GetOrAdd returns the cached value or creates and stores one while holding the +// cache lock. The create function must not call back into this cache. +func (cache *BoundedLRU[K, V]) GetOrAdd(key K, create func() V) V { + cache.mu.Lock() + if element, ok := cache.entries[key]; ok { + cache.order.MoveToFront(element) + value := element.Value.(boundedLRUEntry[K, V]).value + cache.mu.Unlock() + return value + } + + value := create() + element := cache.order.PushFront(boundedLRUEntry[K, V]{key: key, value: value}) + cache.entries[key] = element + + var evicted boundedLRUEntry[K, V] + didEvict := false + if cache.order.Len() > cache.capacity { + oldest := cache.order.Back() + evicted = oldest.Value.(boundedLRUEntry[K, V]) + delete(cache.entries, evicted.key) + cache.order.Remove(oldest) + didEvict = true + } + cache.mu.Unlock() + + if didEvict && cache.onEvict != nil { + cache.onEvict(evicted.key, evicted.value) + } + return value +} + +func (cache *BoundedLRU[K, V]) Get(key K) (V, bool) { + cache.mu.Lock() + defer cache.mu.Unlock() + if element, ok := cache.entries[key]; ok { + cache.order.MoveToFront(element) + return element.Value.(boundedLRUEntry[K, V]).value, true + } + var zero V + return zero, false +} + +func (cache *BoundedLRU[K, V]) Len() int { + cache.mu.Lock() + defer cache.mu.Unlock() + return len(cache.entries) +} diff --git a/internal/cache/bounded_lru_test.go b/internal/cache/bounded_lru_test.go new file mode 100644 index 00000000..34d3dfd1 --- /dev/null +++ b/internal/cache/bounded_lru_test.go @@ -0,0 +1,57 @@ +package cache + +import "testing" + +func TestBoundedLRUEvictsLeastRecentlyUsed(t *testing.T) { + var evicted []string + cache := NewBoundedLRU[string, string](2, func(key, value string) { + evicted = append(evicted, key+"="+value) + }) + + if got := cache.GetOrAdd("a", func() string { return "A" }); got != "A" { + t.Fatalf("first value = %q, want A", got) + } + cache.GetOrAdd("b", func() string { return "B" }) + if got, found := cache.Get("a"); !found || got != "A" { + t.Fatalf("Get(a) = %q/%t, want A/true", got, found) + } + cache.GetOrAdd("c", func() string { return "C" }) + + if _, found := cache.Get("b"); found { + t.Fatal("least recently used entry b was not evicted") + } + if got := cache.Len(); got != 2 { + t.Fatalf("Len() = %d, want 2", got) + } + if len(evicted) != 1 || evicted[0] != "b=B" { + t.Fatalf("evicted = %v, want [b=B]", evicted) + } +} + +func TestBoundedLRUCreatesOneValuePerKeyConcurrently(t *testing.T) { + cache := NewBoundedLRU[string, int](2, nil) + started := make(chan struct{}) + release := make(chan struct{}) + results := make(chan int, 2) + creates := make(chan struct{}, 2) + + create := func() int { + creates <- struct{}{} + close(started) + <-release + return 42 + } + go func() { results <- cache.GetOrAdd("key", create) }() + <-started + go func() { results <- cache.GetOrAdd("key", func() int { creates <- struct{}{}; return 7 }) }() + close(release) + + for range 2 { + if got := <-results; got != 42 { + t.Fatalf("cached value = %d, want 42", got) + } + } + if got := len(creates); got != 1 { + t.Fatalf("create calls = %d, want 1", got) + } +} diff --git a/internal/runtime/executor/claude_executor_auth_test.go b/internal/runtime/executor/claude_executor_auth_test.go index f72ddb8a..433cea90 100644 --- a/internal/runtime/executor/claude_executor_auth_test.go +++ b/internal/runtime/executor/claude_executor_auth_test.go @@ -2,14 +2,81 @@ package executor import ( "context" + "errors" "fmt" + "net/http" "testing" claudeauth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/claude" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" cliproxyauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" + cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" + sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" ) +func TestClaudeExecutorDuplicateMetadataIsRequestScoped(t *testing.T) { + testCases := []struct { + name string + run func(context.Context, *ClaudeExecutor, *cliproxyauth.Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) error + }{ + { + name: "execute", + run: func(ctx context.Context, executor *ClaudeExecutor, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) error { + _, errExecute := executor.Execute(ctx, auth, req, opts) + return errExecute + }, + }, + { + name: "stream", + run: func(ctx context.Context, executor *ClaudeExecutor, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) error { + _, errStream := executor.ExecuteStream(ctx, auth, req, opts) + return errStream + }, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + upstreamCalled := false + transport := roundTripperFunc(func(*http.Request) (*http.Response, error) { + upstreamCalled = true + return nil, errors.New("unexpected upstream request") + }) + ctx := context.WithValue(t.Context(), "cliproxy.roundtripper", http.RoundTripper(transport)) + auth := &cliproxyauth.Auth{ + Provider: "claude", + Attributes: map[string]string{"api_key": "sk-ant-oat-duplicate-metadata", "auth_kind": "oauth"}, + Metadata: map[string]any{ + "account_uuid": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + claudeauth.ClaudeDeviceIDsMetadataKey: []string{ + "0000000000000000000000000000000000000000000000000000000000000000", + }, + }, + } + req := cliproxyexecutor.Request{ + Model: "claude-opus-5", + Payload: []byte(`{"model":"claude-opus-5","messages":[{"role":"user","content":"hello"}],` + + `"metadata":{"user_id":"{}"},"metadata":{"user_id":"{}"}}`), + } + errRun := testCase.run(ctx, NewClaudeExecutor(&config.Config{}), auth, req, cliproxyexecutor.Options{SourceFormat: sdktranslator.FormatClaude}) + if errRun == nil { + t.Fatal("duplicate metadata error = nil") + } + if upstreamCalled { + t.Fatal("duplicate metadata reached upstream") + } + var requestErr cliproxyexecutor.RequestScopedError + if !errors.As(errRun, &requestErr) || requestErr == nil || !requestErr.IsRequestScoped() { + t.Fatalf("duplicate metadata error = %T %v, want request-scoped", errRun, errRun) + } + var statusErr interface{ StatusCode() int } + if !errors.As(errRun, &statusErr) || statusErr.StatusCode() != http.StatusBadRequest { + t.Fatalf("duplicate metadata error = %T %v, want HTTP 400", errRun, errRun) + } + }) + } +} + func TestClaudeExecutorPrepareRequestAuthPopulatesCredentialIdentity(t *testing.T) { executor := NewClaudeExecutor(&config.Config{}) executor.oauthProfileFetcher = func(_ context.Context, _ *cliproxyauth.Auth, accessToken string) (*claudeauth.OAuthProfile, error) { diff --git a/internal/runtime/executor/claude_executor_fast_error.go b/internal/runtime/executor/claude_executor_fast_error.go index a3b54060..5b895c8f 100644 --- a/internal/runtime/executor/claude_executor_fast_error.go +++ b/internal/runtime/executor/claude_executor_fast_error.go @@ -33,7 +33,7 @@ func (e *claudeFastRequestError) Unwrap() error { } func (e *claudeFastRequestError) StatusCode() int { - if e == nil { + if e == nil || (e.status >= http.StatusOK && e.status < http.StatusMultipleChoices) { return 0 } return e.status diff --git a/internal/runtime/executor/claude_executor_fast_error_test.go b/internal/runtime/executor/claude_executor_fast_error_test.go index de76e453..3d16943a 100644 --- a/internal/runtime/executor/claude_executor_fast_error_test.go +++ b/internal/runtime/executor/claude_executor_fast_error_test.go @@ -137,6 +137,75 @@ func TestClaudeExecutorFastHTTPErrorPassesThroughWithoutRetry(t *testing.T) { } } +func TestClaudeExecutorFastSuccessfulHTTPDecodeErrorDoesNotExposeSuccessStatus(t *testing.T) { + testCases := []struct { + name string + run func(context.Context, *ClaudeExecutor, *cliproxyauth.Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) error + }{ + { + name: "execute", + run: func(ctx context.Context, executor *ClaudeExecutor, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) error { + _, errExecute := executor.Execute(ctx, auth, req, opts) + return errExecute + }, + }, + { + name: "stream", + run: func(ctx context.Context, executor *ClaudeExecutor, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) error { + _, errStream := executor.ExecuteStream(ctx, auth, req, opts) + return errStream + }, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + var attempts atomic.Int32 + transport := roundTripperFunc(func(req *http.Request) (*http.Response, error) { + attempts.Add(1) + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}, "Content-Encoding": []string{"gzip"}}, + Body: io.NopCloser(strings.NewReader("not-a-gzip-stream")), + Request: req, + }, nil + }) + ctx := context.WithValue(t.Context(), "cliproxy.roundtripper", http.RoundTripper(transport)) + auth := &cliproxyauth.Auth{ + ID: "fast-success-decode-error", + Attributes: map[string]string{"api_key": "sk-ant-oat-fast-success-decode-error"}, + Metadata: claudeOAuthTestMetadata(), + } + request := cliproxyexecutor.Request{ + Model: "claude-opus-5", + Payload: []byte(`{"model":"claude-opus-5","max_tokens":16,"speed":"fast","messages":[{"role":"user","content":"reply OK"}]}`), + } + errRun := testCase.run(ctx, NewClaudeExecutor(&config.Config{}), auth, request, cliproxyexecutor.Options{ + Stream: testCase.name == "stream", + SourceFormat: sdktranslator.FormatClaude, + ResponseFormat: sdktranslator.FormatClaude, + }) + if errRun == nil { + t.Fatal("Fast decode error = nil") + } + if got := attempts.Load(); got != 1 { + t.Fatalf("upstream attempts = %d, want 1", got) + } + var requestErr cliproxyexecutor.RequestScopedError + if !errors.As(errRun, &requestErr) || requestErr == nil || !requestErr.IsRequestScoped() { + t.Fatalf("Fast decode error = %T %v, want request-scoped", errRun, errRun) + } + var statusErr interface{ StatusCode() int } + if !errors.As(errRun, &statusErr) || statusErr == nil { + t.Fatalf("Fast decode error = %T %v, want status provider", errRun, errRun) + } + if got := statusErr.StatusCode(); got != 0 { + t.Fatalf("Fast decode status = %d, want 0 instead of upstream success", got) + } + }) + } +} + func TestClaudeExecutorFastTransportErrorIsRequestScopedWithoutRetry(t *testing.T) { upstreamErr := errors.New("transport unavailable") var attempts atomic.Int32 diff --git a/internal/runtime/executor/claude_executor_request.go b/internal/runtime/executor/claude_executor_request.go index cabdadf3..7f5fe89b 100644 --- a/internal/runtime/executor/claude_executor_request.go +++ b/internal/runtime/executor/claude_executor_request.go @@ -307,11 +307,7 @@ func claudeRequestedBetas(incomingBetas string, extraBetas []string) map[string] // gateways set their own host, yet both delegate to ClaudeExecutor and are // therefore cloaked; a cloak-keyed rule silently rewrites their traffic too. func isAnthropicUpstreamURL(u *url.URL) bool { - if u == nil || u.User != nil || !strings.EqualFold(u.Scheme, "https") || !strings.EqualFold(u.Hostname(), "api.anthropic.com") { - return false - } - port := u.Port() - return port == "" || port == "443" + return helps.IsAnthropicUpstreamURL(u) } // isAnthropicUpstreamBase reports whether a configured base URL targets Anthropic's diff --git a/internal/runtime/executor/helps/claude_credential_identity.go b/internal/runtime/executor/helps/claude_credential_identity.go index c55bc72c..772682c7 100644 --- a/internal/runtime/executor/helps/claude_credential_identity.go +++ b/internal/runtime/executor/helps/claude_credential_identity.go @@ -199,6 +199,42 @@ func ClaudeCredentialAccountUUID(auth *cliproxyauth.Auth) string { return "" } +type claudeCredentialMetadataRequestError struct { + cause error +} + +func (e *claudeCredentialMetadataRequestError) Error() string { + if e == nil || e.cause == nil { + return "" + } + return e.cause.Error() +} + +func (e *claudeCredentialMetadataRequestError) Unwrap() error { + if e == nil { + return nil + } + return e.cause +} + +func (e *claudeCredentialMetadataRequestError) StatusCode() int { + if e == nil { + return 0 + } + return http.StatusBadRequest +} + +func (e *claudeCredentialMetadataRequestError) IsRequestScoped() bool { + return e != nil +} + +func newClaudeCredentialMetadataRequestError(err error) error { + if err == nil { + return nil + } + return &claudeCredentialMetadataRequestError{cause: err} +} + // ApplyClaudeCredentialMetadata rewrites the identity exception shared by native and cloaked OAuth requests. func ApplyClaudeCredentialMetadata(payload []byte, auth *cliproxyauth.Auth, sessionID string) ([]byte, string, error) { if auth == nil { @@ -206,7 +242,7 @@ func ApplyClaudeCredentialMetadata(payload []byte, auth *cliproxyauth.Auth, sess } metadata, metadataPresent, errMetadata := uniqueClaudeJSONObjectMember(payload, "metadata") if errMetadata != nil { - return nil, "", fmt.Errorf("apply Claude credential metadata: %w", errMetadata) + return nil, "", newClaudeCredentialMetadataRequestError(fmt.Errorf("apply Claude credential metadata: %w", errMetadata)) } var existing string if metadataPresent { @@ -214,7 +250,7 @@ func ApplyClaudeCredentialMetadata(payload []byte, auth *cliproxyauth.Auth, sess if len(trimmedMetadata) >= 2 && trimmedMetadata[0] == '{' { userID, userIDPresent, errUserID := uniqueClaudeJSONObjectMember(trimmedMetadata, "user_id") if errUserID != nil { - return nil, "", fmt.Errorf("apply Claude credential metadata: metadata: %w", errUserID) + return nil, "", newClaudeCredentialMetadataRequestError(fmt.Errorf("apply Claude credential metadata: metadata: %w", errUserID)) } if userIDPresent && json.Unmarshal(userID, &existing) != nil { existing = "" @@ -237,7 +273,7 @@ func ApplyClaudeCredentialMetadata(payload []byte, auth *cliproxyauth.Auth, sess encoded, errIdentity := rebuildClaudeMetadataUserID(existing, deviceID, accountUUID, sessionID) if errIdentity != nil { - return nil, "", fmt.Errorf("apply Claude credential metadata: %w", errIdentity) + return nil, "", newClaudeCredentialMetadataRequestError(fmt.Errorf("apply Claude credential metadata: %w", errIdentity)) } updated, errSet := sjson.SetBytes(payload, "metadata.user_id", string(encoded)) if errSet != nil { diff --git a/internal/runtime/executor/helps/claude_credential_identity_test.go b/internal/runtime/executor/helps/claude_credential_identity_test.go index eb65f9c3..6d02cce4 100644 --- a/internal/runtime/executor/helps/claude_credential_identity_test.go +++ b/internal/runtime/executor/helps/claude_credential_identity_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "net/http" "strings" "testing" @@ -179,6 +180,10 @@ func TestApplyClaudeCredentialMetadataRejectsDuplicateIdentityContainers(t *test name string body string }{ + { + name: "invalid request JSON", + body: `{"messages":[],"metadata":`, + }, { name: "duplicate top-level metadata", body: `{"messages":[],"metadata":{"user_id":"{}"},"metadata":{"user_id":"{}"}}`, @@ -194,9 +199,18 @@ func TestApplyClaudeCredentialMetadataRejectsDuplicateIdentityContainers(t *test } for _, test := range tests { t.Run(test.name, func(t *testing.T) { - if _, _, errApply := ApplyClaudeCredentialMetadata([]byte(test.body), auth, sessionID); errApply == nil { + _, _, errApply := ApplyClaudeCredentialMetadata([]byte(test.body), auth, sessionID) + if errApply == nil { t.Fatal("ApplyClaudeCredentialMetadata() error = nil, want duplicate-key rejection") } + var requestErr cliproxyexecutor.RequestScopedError + if !errors.As(errApply, &requestErr) || requestErr == nil || !requestErr.IsRequestScoped() { + t.Fatalf("ApplyClaudeCredentialMetadata() error = %T %v, want request-scoped", errApply, errApply) + } + var statusErr interface{ StatusCode() int } + if !errors.As(errApply, &statusErr) || statusErr.StatusCode() != http.StatusBadRequest { + t.Fatalf("ApplyClaudeCredentialMetadata() error = %T %v, want HTTP 400", errApply, errApply) + } }) } } @@ -207,11 +221,16 @@ func TestApplyClaudeCredentialMetadataRequiresAccountUUID(t *testing.T) { "0000000000000000000000000000000000000000000000000000000000000000", }, }} - if _, _, errApply := ApplyClaudeCredentialMetadata( + _, _, errApply := ApplyClaudeCredentialMetadata( []byte(`{"messages":[]}`), auth, "11111111-2222-4333-8444-555555555555", - ); errApply == nil { + ) + if errApply == nil { t.Fatal("ApplyClaudeCredentialMetadata() error = nil, want missing account UUID rejection") } + var requestErr cliproxyexecutor.RequestScopedError + if errors.As(errApply, &requestErr) && requestErr != nil && requestErr.IsRequestScoped() { + t.Fatalf("missing credential identity error = %T %v, want credential-scoped", errApply, errApply) + } } diff --git a/internal/runtime/executor/helps/claude_diagnostics.go b/internal/runtime/executor/helps/claude_diagnostics.go index 7dc8b836..d1d0a99e 100644 --- a/internal/runtime/executor/helps/claude_diagnostics.go +++ b/internal/runtime/executor/helps/claude_diagnostics.go @@ -3,27 +3,33 @@ package helps import ( "crypto/sha256" "encoding/hex" + "sort" "strings" "sync" "time" ) const ( - claudeDiagnosticsTTL = time.Hour - claudeDiagnosticsCleanupPeriod = 15 * time.Minute + claudeDiagnosticsTTL = time.Hour + claudeDiagnosticsCleanupPeriod = 15 * time.Minute + claudeDiagnosticsMaxEntries = 4096 + claudeDiagnosticsEvictBatchSize = 256 ) type claudeDiagnosticsEntry struct { previousMessageID string - nextSequence uint64 + minimumSequence uint64 committedSequence uint64 + lastAccess uint64 expiresAt time.Time } var claudeDiagnosticsState = struct { sync.Mutex - entries map[string]claudeDiagnosticsEntry - lastCleanup time.Time + entries map[string]claudeDiagnosticsEntry + lastCleanup time.Time + nextSequence uint64 + nextAccess uint64 }{entries: make(map[string]claudeDiagnosticsEntry)} // BeginClaudeDiagnostics starts one request generation for a stable credential @@ -43,27 +49,29 @@ func BeginClaudeDiagnostics(credentialIdentity, sessionID string) (key string, s claudeDiagnosticsState.Lock() defer claudeDiagnosticsState.Unlock() - if claudeDiagnosticsState.lastCleanup.IsZero() || now.Sub(claudeDiagnosticsState.lastCleanup) >= claudeDiagnosticsCleanupPeriod { - for candidateKey, candidate := range claudeDiagnosticsState.entries { - if !candidate.expiresAt.IsZero() && now.After(candidate.expiresAt) { - delete(claudeDiagnosticsState.entries, candidateKey) - } - } - claudeDiagnosticsState.lastCleanup = now + cleanupClaudeDiagnosticsLocked(now) + + entry, found := claudeDiagnosticsState.entries[key] + newGeneration := !found || (!entry.expiresAt.IsZero() && now.After(entry.expiresAt)) + if newGeneration && !found { + evictClaudeDiagnosticsLocked() } - entry := claudeDiagnosticsState.entries[key] - if !entry.expiresAt.IsZero() && now.After(entry.expiresAt) { - entry = claudeDiagnosticsEntry{} + + claudeDiagnosticsState.nextSequence++ + sequence = claudeDiagnosticsState.nextSequence + if newGeneration { + entry = claudeDiagnosticsEntry{minimumSequence: sequence} } - entry.nextSequence++ + claudeDiagnosticsState.nextAccess++ + entry.lastAccess = claudeDiagnosticsState.nextAccess entry.expiresAt = now.Add(claudeDiagnosticsTTL) claudeDiagnosticsState.entries[key] = entry - return key, entry.nextSequence, entry.previousMessageID + return key, sequence, entry.previousMessageID } // CommitClaudeDiagnostics advances continuity only after a response completes. // A response from an older concurrently-started request cannot overwrite a -// newer committed generation. +// newer committed generation, including after TTL expiry or capacity eviction. func CommitClaudeDiagnostics(key string, sequence uint64, messageID string) { key = strings.TrimSpace(key) messageID = strings.TrimSpace(messageID) @@ -75,18 +83,55 @@ func CommitClaudeDiagnostics(key string, sequence uint64, messageID string) { claudeDiagnosticsState.Lock() defer claudeDiagnosticsState.Unlock() entry, ok := claudeDiagnosticsState.entries[key] - if !ok || sequence < entry.committedSequence { + if !ok || sequence < entry.minimumSequence || sequence < entry.committedSequence { return } + claudeDiagnosticsState.nextAccess++ entry.previousMessageID = messageID entry.committedSequence = sequence + entry.lastAccess = claudeDiagnosticsState.nextAccess entry.expiresAt = now.Add(claudeDiagnosticsTTL) claudeDiagnosticsState.entries[key] = entry } +func cleanupClaudeDiagnosticsLocked(now time.Time) { + if !claudeDiagnosticsState.lastCleanup.IsZero() && now.Sub(claudeDiagnosticsState.lastCleanup) < claudeDiagnosticsCleanupPeriod { + return + } + for key, entry := range claudeDiagnosticsState.entries { + if !entry.expiresAt.IsZero() && now.After(entry.expiresAt) { + delete(claudeDiagnosticsState.entries, key) + } + } + claudeDiagnosticsState.lastCleanup = now +} + +func evictClaudeDiagnosticsLocked() { + if len(claudeDiagnosticsState.entries) < claudeDiagnosticsMaxEntries { + return + } + type candidate struct { + key string + lastAccess uint64 + } + candidates := make([]candidate, 0, len(claudeDiagnosticsState.entries)) + for key, entry := range claudeDiagnosticsState.entries { + candidates = append(candidates, candidate{key: key, lastAccess: entry.lastAccess}) + } + sort.Slice(candidates, func(i, j int) bool { + return candidates[i].lastAccess < candidates[j].lastAccess + }) + count := min(claudeDiagnosticsEvictBatchSize, len(candidates)) + for _, candidate := range candidates[:count] { + delete(claudeDiagnosticsState.entries, candidate.key) + } +} + func resetClaudeDiagnosticsForTest() { claudeDiagnosticsState.Lock() defer claudeDiagnosticsState.Unlock() claudeDiagnosticsState.entries = make(map[string]claudeDiagnosticsEntry) claudeDiagnosticsState.lastCleanup = time.Time{} + claudeDiagnosticsState.nextSequence = 0 + claudeDiagnosticsState.nextAccess = 0 } diff --git a/internal/runtime/executor/helps/claude_diagnostics_test.go b/internal/runtime/executor/helps/claude_diagnostics_test.go index 454bf900..09a0e075 100644 --- a/internal/runtime/executor/helps/claude_diagnostics_test.go +++ b/internal/runtime/executor/helps/claude_diagnostics_test.go @@ -1,6 +1,10 @@ package helps -import "testing" +import ( + "fmt" + "testing" + "time" +) func TestClaudeDiagnosticsTracksCompletedMessagePerCredentialSession(t *testing.T) { resetClaudeDiagnosticsForTest() @@ -23,6 +27,66 @@ func TestClaudeDiagnosticsTracksCompletedMessagePerCredentialSession(t *testing. } } +func TestClaudeDiagnosticsRejectsExpiredGenerationCommit(t *testing.T) { + resetClaudeDiagnosticsForTest() + defer resetClaudeDiagnosticsForTest() + + key, expiredSequence, _ := BeginClaudeDiagnostics("credential", "session") + claudeDiagnosticsState.Lock() + entry := claudeDiagnosticsState.entries[key] + entry.expiresAt = time.Now().Add(-time.Second) + claudeDiagnosticsState.entries[key] = entry + claudeDiagnosticsState.Unlock() + + newKey, currentSequence, previous := BeginClaudeDiagnostics("credential", "session") + if newKey != key || currentSequence <= expiredSequence || previous != "" { + t.Fatalf("new generation = %q/%d/%q, want same key/new sequence/empty", newKey, currentSequence, previous) + } + CommitClaudeDiagnostics(newKey, currentSequence, "msg_current") + CommitClaudeDiagnostics(key, expiredSequence, "msg_expired") + _, _, previous = BeginClaudeDiagnostics("credential", "session") + if previous != "msg_current" { + t.Fatalf("previous message = %q, want current generation", previous) + } +} + +func TestClaudeDiagnosticsCacheEvictsOldestEntriesWithinCapacity(t *testing.T) { + resetClaudeDiagnosticsForTest() + defer resetClaudeDiagnosticsForTest() + + firstKey, firstSequence, _ := BeginClaudeDiagnostics("credential", "session-0") + var newestKey string + for index := 1; index <= claudeDiagnosticsMaxEntries; index++ { + newestKey, _, _ = BeginClaudeDiagnostics("credential", fmt.Sprintf("session-%d", index)) + } + + claudeDiagnosticsState.Lock() + entryCount := len(claudeDiagnosticsState.entries) + _, firstFound := claudeDiagnosticsState.entries[firstKey] + _, newestFound := claudeDiagnosticsState.entries[newestKey] + claudeDiagnosticsState.Unlock() + if entryCount > claudeDiagnosticsMaxEntries { + t.Fatalf("cache entries = %d, want at most %d", entryCount, claudeDiagnosticsMaxEntries) + } + if firstFound { + t.Fatal("oldest diagnostics entry was not evicted") + } + if !newestFound { + t.Fatal("newest diagnostics entry was evicted") + } + + newKey, newSequence, _ := BeginClaudeDiagnostics("credential", "session-0") + if newKey != firstKey || newSequence <= firstSequence { + t.Fatalf("recreated generation = %q/%d, want same key after sequence %d", newKey, newSequence, firstSequence) + } + CommitClaudeDiagnostics(newKey, newSequence, "msg_recreated") + CommitClaudeDiagnostics(firstKey, firstSequence, "msg_evicted") + _, _, previous := BeginClaudeDiagnostics("credential", "session-0") + if previous != "msg_recreated" { + t.Fatalf("previous message = %q, want recreated generation", previous) + } +} + func TestClaudeDiagnosticsRejectsLateOlderCommit(t *testing.T) { resetClaudeDiagnosticsForTest() defer resetClaudeDiagnosticsForTest() diff --git a/internal/runtime/executor/helps/claude_upstream.go b/internal/runtime/executor/helps/claude_upstream.go new file mode 100644 index 00000000..bb2b2ef7 --- /dev/null +++ b/internal/runtime/executor/helps/claude_upstream.go @@ -0,0 +1,17 @@ +package helps + +import ( + "net/url" + "strings" +) + +// IsAnthropicUpstreamURL reports whether a resolved request targets Anthropic's +// first-party API origin. Claude-specific body, header, HTTP, and TLS behavior +// must all use this gate so they cannot drift onto custom ports or userinfo URLs. +func IsAnthropicUpstreamURL(u *url.URL) bool { + if u == nil || u.User != nil || !strings.EqualFold(u.Scheme, "https") || !strings.EqualFold(u.Hostname(), "api.anthropic.com") { + return false + } + port := u.Port() + return port == "" || port == "443" +} diff --git a/internal/runtime/executor/helps/claude_upstream_test.go b/internal/runtime/executor/helps/claude_upstream_test.go new file mode 100644 index 00000000..0d345761 --- /dev/null +++ b/internal/runtime/executor/helps/claude_upstream_test.go @@ -0,0 +1,38 @@ +package helps + +import ( + "net/url" + "testing" +) + +func TestIsAnthropicUpstreamURL(t *testing.T) { + testCases := []struct { + name string + targetURL string + want bool + }{ + {name: "default HTTPS port", targetURL: "https://api.anthropic.com/v1/messages", want: true}, + {name: "explicit HTTPS port", targetURL: "https://api.anthropic.com:443/v1/messages", want: true}, + {name: "case insensitive host", targetURL: "https://API.ANTHROPIC.COM/v1/messages", want: true}, + {name: "HTTP", targetURL: "http://api.anthropic.com/v1/messages", want: false}, + {name: "custom port", targetURL: "https://api.anthropic.com:8443/v1/messages", want: false}, + {name: "userinfo", targetURL: "https://caller@api.anthropic.com/v1/messages", want: false}, + {name: "lookalike host", targetURL: "https://api.anthropic.com.example/v1/messages", want: false}, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + parsed, errParse := url.Parse(testCase.targetURL) + if errParse != nil { + t.Fatal(errParse) + } + if got := IsAnthropicUpstreamURL(parsed); got != testCase.want { + t.Fatalf("IsAnthropicUpstreamURL(%q) = %t, want %t", testCase.targetURL, got, testCase.want) + } + }) + } + + if IsAnthropicUpstreamURL(nil) { + t.Fatal("IsAnthropicUpstreamURL(nil) = true") + } +} diff --git a/internal/runtime/executor/helps/utls_client.go b/internal/runtime/executor/helps/utls_client.go index 03067a17..d0e23875 100644 --- a/internal/runtime/executor/helps/utls_client.go +++ b/internal/runtime/executor/helps/utls_client.go @@ -10,6 +10,7 @@ import ( "time" tls "github.com/refraction-networking/utls" + internalcache "github.com/router-for-me/CLIProxyAPI/v7/internal/cache" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" "github.com/router-for-me/CLIProxyAPI/v7/internal/httpwire" cliproxyauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" @@ -208,7 +209,16 @@ func claudeCodeTLSClientHelloSpec() *tls.ClientHelloSpec { } } -var claudeCodeRoundTripperCache sync.Map +const claudeCodeRoundTripperCacheCapacity = 64 + +var claudeCodeRoundTripperCache = internalcache.NewBoundedLRU[string, http.RoundTripper]( + claudeCodeRoundTripperCacheCapacity, + func(_ string, roundTripper http.RoundTripper) { + if transport, ok := roundTripper.(interface{ CloseIdleConnections() }); ok { + transport.CloseIdleConnections() + } + }, +) var claudeCodeMessagesHeaderOrder = []string{ "Accept", @@ -267,18 +277,9 @@ func claudeCodeRequestHeaderOrder(_, requestTarget string) []string { } func cachedClaudeCodeRoundTripper(proxyURL string) http.RoundTripper { - if cached, ok := claudeCodeRoundTripperCache.Load(proxyURL); ok { - return cached.(http.RoundTripper) - } - created := newClaudeCodeRoundTripper(proxyURL) - actual, loaded := claudeCodeRoundTripperCache.LoadOrStore(proxyURL, created) - if loaded { - if transport, ok := created.(*http.Transport); ok { - transport.CloseIdleConnections() - } - return actual.(http.RoundTripper) - } - return created + return claudeCodeRoundTripperCache.GetOrAdd(proxyURL, func() http.RoundTripper { + return newClaudeCodeRoundTripper(proxyURL) + }) } func newClaudeCodeRoundTripper(proxyURL string) http.RoundTripper { @@ -346,13 +347,11 @@ type fallbackRoundTripper struct { } func (f *fallbackRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { - if req.URL.Scheme == "https" { - switch strings.ToLower(req.URL.Hostname()) { - case "api.anthropic.com": - return f.anthropic.RoundTrip(req) - case "chatgpt.com": - return f.chrome.RoundTrip(req) - } + if IsAnthropicUpstreamURL(req.URL) { + return f.anthropic.RoundTrip(req) + } + if req.URL.Scheme == "https" && strings.EqualFold(req.URL.Hostname(), "chatgpt.com") { + return f.chrome.RoundTrip(req) } return f.fallback.RoundTrip(req) } diff --git a/internal/runtime/executor/helps/utls_client_test.go b/internal/runtime/executor/helps/utls_client_test.go index 08321872..30cee33f 100644 --- a/internal/runtime/executor/helps/utls_client_test.go +++ b/internal/runtime/executor/helps/utls_client_test.go @@ -174,6 +174,20 @@ func TestCachedClaudeCodeRoundTripperReusesTransport(t *testing.T) { } } +func TestCachedClaudeCodeRoundTripperBoundsProxyCardinality(t *testing.T) { + firstProxy := fmt.Sprintf("http://127.0.0.1:%d", 30000) + first := cachedClaudeCodeRoundTripper(firstProxy) + for index := 1; index <= claudeCodeRoundTripperCacheCapacity; index++ { + cachedClaudeCodeRoundTripper(fmt.Sprintf("http://127.0.0.1:%d", 30000+index)) + } + if got := claudeCodeRoundTripperCache.Len(); got > claudeCodeRoundTripperCacheCapacity { + t.Fatalf("transport cache entries = %d, want at most %d", got, claudeCodeRoundTripperCacheCapacity) + } + if recreated := cachedClaudeCodeRoundTripper(firstProxy); recreated == first { + t.Fatal("least recently used proxy transport was not evicted") + } +} + func TestClaudeCodeTLSClientHelloCapture(t *testing.T) { proxyURL := os.Getenv("CPA_TLS_FP_PROXY") if proxyURL == "" { @@ -220,6 +234,10 @@ func TestFallbackRoundTripperSelectsProviderFingerprint(t *testing.T) { want string }{ {name: "Anthropic HTTPS", url: "https://api.anthropic.com/v1/messages", want: "anthropic"}, + {name: "Anthropic explicit HTTPS port", url: "https://api.anthropic.com:443/v1/messages", want: "anthropic"}, + {name: "Anthropic custom port", url: "https://api.anthropic.com:8443/v1/messages", want: "fallback"}, + {name: "Anthropic userinfo", url: "https://caller@api.anthropic.com/v1/messages", want: "fallback"}, + {name: "Anthropic lookalike", url: "https://api.anthropic.com.example/v1/messages", want: "fallback"}, {name: "ChatGPT HTTPS", url: "https://chatgpt.com/backend-api/codex/responses", want: "chrome"}, {name: "Other HTTPS", url: "https://example.com/v1/messages", want: "fallback"}, {name: "Anthropic HTTP", url: "http://api.anthropic.com/v1/messages", want: "fallback"}, diff --git a/internal/translator/gemini/claude/gemini_claude_request_test.go b/internal/translator/gemini/claude/gemini_claude_request_test.go index 92f08b05..58b9077c 100644 --- a/internal/translator/gemini/claude/gemini_claude_request_test.go +++ b/internal/translator/gemini/claude/gemini_claude_request_test.go @@ -101,7 +101,7 @@ func TestConvertClaudeRequestToGemini_StripsClaudeCodeAttribution(t *testing.T) "model": "claude-sonnet-4-5", "system": [ {"type": "text", "text": "x-anthropic-billing-header: cc_version=2.1.63.abc; cc_entrypoint=cli; cch=12345;"}, - {"type": "text", "text": "You are Claude Code, Anthropic's official CLI for Claude."}, + {"type": "text", "text": "You are a Claude agent, built on Anthropic's Claude Agent SDK."}, {"type": "text", "text": "User system prompt"} ], "messages": [{"role": "user", "content": [{"type": "text", "text": "hi"}]}] @@ -113,7 +113,7 @@ func TestConvertClaudeRequestToGemini_StripsClaudeCodeAttribution(t *testing.T) if len(parts) != 2 { t.Fatalf("Expected 2 system parts after attribution strip, got %d: %s", len(parts), gjson.GetBytes(output, "systemInstruction.parts").Raw) } - if got := parts[0].Get("text").String(); got != "You are Claude Code, Anthropic's official CLI for Claude." { + if got := parts[0].Get("text").String(); got != "You are a Claude agent, built on Anthropic's Claude Agent SDK." { t.Fatalf("Unexpected first system part: %q", got) } if got := parts[1].Get("text").String(); got != "User system prompt" { diff --git a/sdk/cliproxy/auth/conductor_fast_error_test.go b/sdk/cliproxy/auth/conductor_fast_error_test.go index dc95bda0..7956bdb3 100644 --- a/sdk/cliproxy/auth/conductor_fast_error_test.go +++ b/sdk/cliproxy/auth/conductor_fast_error_test.go @@ -37,6 +37,86 @@ func newFastDirectResponseTestError(status int, body string) error { }} } +func TestManagerFastLocalErrorDoesNotRefreshRetryOrCoolCredential(t *testing.T) { + testCases := []struct { + name string + configure func(*claudeCancellationTestExecutor, *atomic.Int32) + run func(*Manager, string) error + }{ + { + name: "non-stream", + configure: func(executor *claudeCancellationTestExecutor, calls *atomic.Int32) { + executor.executeFn = func(context.Context, *Auth) (cliproxyexecutor.Response, error) { + if calls.Add(1) == 1 { + return cliproxyexecutor.Response{}, &requestScopedStatusError{message: "decode Fast response"} + } + return cliproxyexecutor.Response{Payload: []byte(`{"type":"message","content":[]}`)}, nil + } + }, + run: func(manager *Manager, model string) error { + _, errExecute := manager.Execute(context.Background(), []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{}) + return errExecute + }, + }, + { + name: "stream", + configure: func(executor *claudeCancellationTestExecutor, calls *atomic.Int32) { + executor.streamFn = func(context.Context, *Auth) (*cliproxyexecutor.StreamResult, error) { + if calls.Add(1) == 1 { + return nil, &requestScopedStatusError{message: "decode Fast stream response"} + } + chunks := make(chan cliproxyexecutor.StreamChunk, 1) + chunks <- cliproxyexecutor.StreamChunk{Payload: []byte("ok")} + close(chunks) + return &cliproxyexecutor.StreamResult{Chunks: chunks}, nil + } + }, + run: func(manager *Manager, model string) error { + stream, errStream := manager.ExecuteStream(context.Background(), []string{"claude"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{Stream: true}) + if errStream != nil { + return errStream + } + for range stream.Chunks { + } + return nil + }, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + var calls atomic.Int32 + executor := &claudeCancellationTestExecutor{} + testCase.configure(executor, &calls) + manager, auth, model := newClaudeCancellationTestManager(t, executor, nil) + + errExecute := testCase.run(manager, model) + if errExecute == nil { + t.Fatal("first Fast request error = nil") + } + var direct *cliproxyexecutor.RequestTerminatedError + if errors.As(errExecute, &direct) { + t.Fatalf("local Fast error unexpectedly became a direct HTTP response: %v", errExecute) + } + if got := calls.Load(); got != 1 { + t.Fatalf("first request upstream calls = %d, want 1", got) + } + if got := executor.refreshCalls.Load(); got != 0 { + t.Fatalf("refresh calls = %d, want 0", got) + } + requireClaudeCancellationNeutral(t, manager, auth.ID, model) + + if errFollowUp := testCase.run(manager, model); errFollowUp != nil { + t.Fatalf("follow-up request error = %v", errFollowUp) + } + if got := calls.Load(); got != 2 { + t.Fatalf("total upstream calls = %d, want 2", got) + } + requireClaudeCancellationNeutral(t, manager, auth.ID, model) + }) + } +} + func TestManagerFastDirectErrorDoesNotRefreshRetryOrCoolCredential(t *testing.T) { testCases := []struct { name string