Merge upstream v6.9.9 (PR #483)

internal/runtime/executor/helps/claude_device_profile.go

@@ -358,6 +358,16 @@ func ApplyClaudeDeviceProfileHeaders(r *http.Request, profile ClaudeDeviceProfil
 	r.Header.Set("X-Stainless-Arch", profile.Arch)
 }
 
+// DefaultClaudeVersion returns the version string (e.g. "2.1.63") from the
+// current baseline device profile. It extracts the version from the User-Agent.
+func DefaultClaudeVersion(cfg *config.Config) string {
+	profile := defaultClaudeDeviceProfile(cfg)
+	if version, ok := parseClaudeCLIVersion(profile.UserAgent); ok {
+		return strconv.Itoa(version.major) + "." + strconv.Itoa(version.minor) + "." + strconv.Itoa(version.patch)
+	}
+	return "2.1.63"
+}
+
 func ApplyClaudeLegacyDeviceHeaders(r *http.Request, ginHeaders http.Header, cfg *config.Config) {
 	if r == nil {
 		return

internal/runtime/executor/helps/logging_helpers.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"html"
 	"net/http"
+	"net/url"
 	"sort"
 	"strings"
 	"time"
@@ -19,9 +20,10 @@ import (
 )
 
 const (
-	apiAttemptsKey = "API_UPSTREAM_ATTEMPTS"
-	apiRequestKey  = "API_REQUEST"
-	apiResponseKey = "API_RESPONSE"
+	apiAttemptsKey          = "API_UPSTREAM_ATTEMPTS"
+	apiRequestKey           = "API_REQUEST"
+	apiResponseKey          = "API_RESPONSE"
+	apiWebsocketTimelineKey = "API_WEBSOCKET_TIMELINE"
 )
 
 // UpstreamRequestLog captures the outbound upstream request details for logging.
@@ -46,6 +48,7 @@ type upstreamAttempt struct {
 	headersWritten       bool
 	bodyStarted          bool
 	bodyHasContent       bool
+	prevWasSSEEvent      bool
 	errorWritten         bool
 }
 
@@ -173,15 +176,157 @@ func AppendAPIResponseChunk(ctx context.Context, cfg *config.Config, chunk []byt
 		attempt.response.WriteString("Body:\n")
 		attempt.bodyStarted = true
 	}
+	currentChunkIsSSEEvent := bytes.HasPrefix(data, []byte("event:"))
+	currentChunkIsSSEData := bytes.HasPrefix(data, []byte("data:"))
 	if attempt.bodyHasContent {
-		attempt.response.WriteString("\n\n")
+		separator := "\n\n"
+		if attempt.prevWasSSEEvent && currentChunkIsSSEData {
+			separator = "\n"
+		}
+		attempt.response.WriteString(separator)
 	}
 	attempt.response.WriteString(string(data))
 	attempt.bodyHasContent = true
+	attempt.prevWasSSEEvent = currentChunkIsSSEEvent
 
 	updateAggregatedResponse(ginCtx, attempts)
 }
 
+// RecordAPIWebsocketRequest stores an upstream websocket request event in Gin context.
+func RecordAPIWebsocketRequest(ctx context.Context, cfg *config.Config, info UpstreamRequestLog) {
+	if cfg == nil || !cfg.RequestLog {
+		return
+	}
+	ginCtx := ginContextFrom(ctx)
+	if ginCtx == nil {
+		return
+	}
+
+	builder := &strings.Builder{}
+	builder.WriteString(fmt.Sprintf("Timestamp: %s\n", time.Now().Format(time.RFC3339Nano)))
+	builder.WriteString("Event: api.websocket.request\n")
+	if info.URL != "" {
+		builder.WriteString(fmt.Sprintf("Upstream URL: %s\n", info.URL))
+	}
+	if auth := formatAuthInfo(info); auth != "" {
+		builder.WriteString(fmt.Sprintf("Auth: %s\n", auth))
+	}
+	builder.WriteString("Headers:\n")
+	writeHeaders(builder, info.Headers)
+	builder.WriteString("\nBody:\n")
+	if len(info.Body) > 0 {
+		builder.Write(info.Body)
+	} else {
+		builder.WriteString("<empty>")
+	}
+	builder.WriteString("\n")
+
+	appendAPIWebsocketTimeline(ginCtx, []byte(builder.String()))
+}
+
+// RecordAPIWebsocketHandshake stores the upstream websocket handshake response metadata.
+func RecordAPIWebsocketHandshake(ctx context.Context, cfg *config.Config, status int, headers http.Header) {
+	if cfg == nil || !cfg.RequestLog {
+		return
+	}
+	ginCtx := ginContextFrom(ctx)
+	if ginCtx == nil {
+		return
+	}
+
+	builder := &strings.Builder{}
+	builder.WriteString(fmt.Sprintf("Timestamp: %s\n", time.Now().Format(time.RFC3339Nano)))
+	builder.WriteString("Event: api.websocket.handshake\n")
+	if status > 0 {
+		builder.WriteString(fmt.Sprintf("Status: %d\n", status))
+	}
+	builder.WriteString("Headers:\n")
+	writeHeaders(builder, headers)
+	builder.WriteString("\n")
+
+	appendAPIWebsocketTimeline(ginCtx, []byte(builder.String()))
+}
+
+// RecordAPIWebsocketUpgradeRejection stores a rejected websocket upgrade as an HTTP attempt.
+func RecordAPIWebsocketUpgradeRejection(ctx context.Context, cfg *config.Config, info UpstreamRequestLog, status int, headers http.Header, body []byte) {
+	if cfg == nil || !cfg.RequestLog {
+		return
+	}
+	ginCtx := ginContextFrom(ctx)
+	if ginCtx == nil {
+		return
+	}
+
+	RecordAPIRequest(ctx, cfg, info)
+	RecordAPIResponseMetadata(ctx, cfg, status, headers)
+	AppendAPIResponseChunk(ctx, cfg, body)
+}
+
+// WebsocketUpgradeRequestURL converts a websocket URL back to its HTTP handshake URL for logging.
+func WebsocketUpgradeRequestURL(rawURL string) string {
+	trimmedURL := strings.TrimSpace(rawURL)
+	if trimmedURL == "" {
+		return ""
+	}
+	parsed, err := url.Parse(trimmedURL)
+	if err != nil {
+		return trimmedURL
+	}
+	switch strings.ToLower(parsed.Scheme) {
+	case "ws":
+		parsed.Scheme = "http"
+	case "wss":
+		parsed.Scheme = "https"
+	}
+	return parsed.String()
+}
+
+// AppendAPIWebsocketResponse stores an upstream websocket response frame in Gin context.
+func AppendAPIWebsocketResponse(ctx context.Context, cfg *config.Config, payload []byte) {
+	if cfg == nil || !cfg.RequestLog {
+		return
+	}
+	data := bytes.TrimSpace(payload)
+	if len(data) == 0 {
+		return
+	}
+	ginCtx := ginContextFrom(ctx)
+	if ginCtx == nil {
+		return
+	}
+	markAPIResponseTimestamp(ginCtx)
+
+	builder := &strings.Builder{}
+	builder.WriteString(fmt.Sprintf("Timestamp: %s\n", time.Now().Format(time.RFC3339Nano)))
+	builder.WriteString("Event: api.websocket.response\n")
+	builder.Write(data)
+	builder.WriteString("\n")
+
+	appendAPIWebsocketTimeline(ginCtx, []byte(builder.String()))
+}
+
+// RecordAPIWebsocketError stores an upstream websocket error event in Gin context.
+func RecordAPIWebsocketError(ctx context.Context, cfg *config.Config, stage string, err error) {
+	if cfg == nil || !cfg.RequestLog || err == nil {
+		return
+	}
+	ginCtx := ginContextFrom(ctx)
+	if ginCtx == nil {
+		return
+	}
+	markAPIResponseTimestamp(ginCtx)
+
+	builder := &strings.Builder{}
+	builder.WriteString(fmt.Sprintf("Timestamp: %s\n", time.Now().Format(time.RFC3339Nano)))
+	builder.WriteString("Event: api.websocket.error\n")
+	if trimmed := strings.TrimSpace(stage); trimmed != "" {
+		builder.WriteString(fmt.Sprintf("Stage: %s\n", trimmed))
+	}
+	builder.WriteString(fmt.Sprintf("Error: %s\n", err.Error()))
+
+	appendAPIWebsocketTimeline(ginCtx, []byte(builder.String()))
+}
+
 func ginContextFrom(ctx context.Context) *gin.Context {
 	ginCtx, _ := ctx.Value("gin").(*gin.Context)
 	return ginCtx
@@ -259,6 +404,40 @@ func updateAggregatedResponse(ginCtx *gin.Context, attempts []*upstreamAttempt)
 	ginCtx.Set(apiResponseKey, []byte(builder.String()))
 }
 
+func appendAPIWebsocketTimeline(ginCtx *gin.Context, chunk []byte) {
+	if ginCtx == nil {
+		return
+	}
+	data := bytes.TrimSpace(chunk)
+	if len(data) == 0 {
+		return
+	}
+	if existing, exists := ginCtx.Get(apiWebsocketTimelineKey); exists {
+		if existingBytes, ok := existing.([]byte); ok && len(existingBytes) > 0 {
+			combined := make([]byte, 0, len(existingBytes)+len(data)+2)
+			combined = append(combined, existingBytes...)
+			if !bytes.HasSuffix(existingBytes, []byte("\n")) {
+				combined = append(combined, '\n')
+			}
+			combined = append(combined, '\n')
+			combined = append(combined, data...)
+			ginCtx.Set(apiWebsocketTimelineKey, combined)
+			return
+		}
+	}
+	ginCtx.Set(apiWebsocketTimelineKey, bytes.Clone(data))
+}
+
+func markAPIResponseTimestamp(ginCtx *gin.Context) {
+	if ginCtx == nil {
+		return
+	}
+	if _, exists := ginCtx.Get("API_RESPONSE_TIMESTAMP"); exists {
+		return
+	}
+	ginCtx.Set("API_RESPONSE_TIMESTAMP", time.Now())
+}
+
 func writeHeaders(builder *strings.Builder, headers http.Header) {
 	if builder == nil {
 		return

internal/runtime/executor/helps/session_id_cache.go

@@ -0,0 +1,92 @@
+package helps
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+type sessionIDCacheEntry struct {
+	value  string
+	expire time.Time
+}
+
+var (
+	sessionIDCache            = make(map[string]sessionIDCacheEntry)
+	sessionIDCacheMu          sync.RWMutex
+	sessionIDCacheCleanupOnce sync.Once
+)
+
+const (
+	sessionIDTTL                = time.Hour
+	sessionIDCacheCleanupPeriod = 15 * time.Minute
+)
+
+func startSessionIDCacheCleanup() {
+	go func() {
+		ticker := time.NewTicker(sessionIDCacheCleanupPeriod)
+		defer ticker.Stop()
+		for range ticker.C {
+			purgeExpiredSessionIDs()
+		}
+	}()
+}
+
+func purgeExpiredSessionIDs() {
+	now := time.Now()
+	sessionIDCacheMu.Lock()
+	for key, entry := range sessionIDCache {
+		if !entry.expire.After(now) {
+			delete(sessionIDCache, key)
+		}
+	}
+	sessionIDCacheMu.Unlock()
+}
+
+func sessionIDCacheKey(apiKey string) string {
+	sum := sha256.Sum256([]byte(apiKey))
+	return hex.EncodeToString(sum[:])
+}
+
+// CachedSessionID returns a stable session UUID per apiKey, refreshing the TTL on each access.
+func CachedSessionID(apiKey string) string {
+	if apiKey == "" {
+		return uuid.New().String()
+	}
+
+	sessionIDCacheCleanupOnce.Do(startSessionIDCacheCleanup)
+
+	key := sessionIDCacheKey(apiKey)
+	now := time.Now()
+
+	sessionIDCacheMu.RLock()
+	entry, ok := sessionIDCache[key]
+	valid := ok && entry.value != "" && entry.expire.After(now)
+	sessionIDCacheMu.RUnlock()
+	if valid {
+		sessionIDCacheMu.Lock()
+		entry = sessionIDCache[key]
+		if entry.value != "" && entry.expire.After(now) {
+			entry.expire = now.Add(sessionIDTTL)
+			sessionIDCache[key] = entry
+			sessionIDCacheMu.Unlock()
+			return entry.value
+		}
+		sessionIDCacheMu.Unlock()
+	}
+
+	newID := uuid.New().String()
+
+	sessionIDCacheMu.Lock()
+	entry, ok = sessionIDCache[key]
+	if !ok || entry.value == "" || !entry.expire.After(now) {
+		entry.value = newID
+	}
+	entry.expire = now.Add(sessionIDTTL)
+	sessionIDCache[key] = entry
+	sessionIDCacheMu.Unlock()
+	return entry.value
+}

internal/runtime/executor/helps/utls_client.go

@@ -0,0 +1,188 @@
+package helps
+
+import (
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	tls "github.com/refraction-networking/utls"
+	"github.com/router-for-me/CLIProxyAPI/v6/internal/config"
+	cliproxyauth "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/auth"
+	"github.com/router-for-me/CLIProxyAPI/v6/sdk/proxyutil"
+	log "github.com/sirupsen/logrus"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/proxy"
+)
+
+// utlsRoundTripper implements http.RoundTripper using utls with Chrome fingerprint
+// to bypass Cloudflare's TLS fingerprinting on Anthropic domains.
+type utlsRoundTripper struct {
+	mu          sync.Mutex
+	connections map[string]*http2.ClientConn
+	pending     map[string]*sync.Cond
+	dialer      proxy.Dialer
+}
+
+func newUtlsRoundTripper(proxyURL string) *utlsRoundTripper {
+	var dialer proxy.Dialer = proxy.Direct
+	if proxyURL != "" {
+		proxyDialer, mode, errBuild := proxyutil.BuildDialer(proxyURL)
+		if errBuild != nil {
+			log.Errorf("utls: failed to configure proxy dialer for %q: %v", proxyURL, errBuild)
+		} else if mode != proxyutil.ModeInherit && proxyDialer != nil {
+			dialer = proxyDialer
+		}
+	}
+	return &utlsRoundTripper{
+		connections: make(map[string]*http2.ClientConn),
+		pending:     make(map[string]*sync.Cond),
+		dialer:      dialer,
+	}
+}
+
+func (t *utlsRoundTripper) getOrCreateConnection(host, addr string) (*http2.ClientConn, error) {
+	t.mu.Lock()
+
+	if h2Conn, ok := t.connections[host]; ok && h2Conn.CanTakeNewRequest() {
+		t.mu.Unlock()
+		return h2Conn, nil
+	}
+
+	if cond, ok := t.pending[host]; ok {
+		cond.Wait()
+		if h2Conn, ok := t.connections[host]; ok && h2Conn.CanTakeNewRequest() {
+			t.mu.Unlock()
+			return h2Conn, nil
+		}
+	}
+
+	cond := sync.NewCond(&t.mu)
+	t.pending[host] = cond
+	t.mu.Unlock()
+
+	h2Conn, err := t.createConnection(host, addr)
+
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	delete(t.pending, host)
+	cond.Broadcast()
+
+	if err != nil {
+		return nil, err
+	}
+
+	t.connections[host] = h2Conn
+	return h2Conn, nil
+}
+
+func (t *utlsRoundTripper) createConnection(host, addr string) (*http2.ClientConn, error) {
+	conn, err := t.dialer.Dial("tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+
+	tlsConfig := &tls.Config{ServerName: host}
+	tlsConn := tls.UClient(conn, tlsConfig, tls.HelloChrome_Auto)
+
+	if err := tlsConn.Handshake(); err != nil {
+		conn.Close()
+		return nil, err
+	}
+
+	tr := &http2.Transport{}
+	h2Conn, err := tr.NewClientConn(tlsConn)
+	if err != nil {
+		tlsConn.Close()
+		return nil, err
+	}
+
+	return h2Conn, nil
+}
+
+func (t *utlsRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	hostname := req.URL.Hostname()
+	port := req.URL.Port()
+	if port == "" {
+		port = "443"
+	}
+	addr := net.JoinHostPort(hostname, port)
+
+	h2Conn, err := t.getOrCreateConnection(hostname, addr)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := h2Conn.RoundTrip(req)
+	if err != nil {
+		t.mu.Lock()
+		if cached, ok := t.connections[hostname]; ok && cached == h2Conn {
+			delete(t.connections, hostname)
+		}
+		t.mu.Unlock()
+		return nil, err
+	}
+
+	return resp, nil
+}
+
+// anthropicHosts contains the hosts that should use utls Chrome TLS fingerprint.
+var anthropicHosts = map[string]struct{}{
+	"api.anthropic.com": {},
+}
+
+// fallbackRoundTripper uses utls for Anthropic HTTPS hosts and falls back to
+// standard transport for all other requests (non-HTTPS or non-Anthropic hosts).
+type fallbackRoundTripper struct {
+	utls     *utlsRoundTripper
+	fallback http.RoundTripper
+}
+
+func (f *fallbackRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	if req.URL.Scheme == "https" {
+		if _, ok := anthropicHosts[strings.ToLower(req.URL.Hostname())]; ok {
+			return f.utls.RoundTrip(req)
+		}
+	}
+	return f.fallback.RoundTrip(req)
+}
+
+// NewUtlsHTTPClient creates an HTTP client using utls Chrome TLS fingerprint.
+// Use this for Claude API requests to match real Claude Code's TLS behavior.
+// Falls back to standard transport for non-HTTPS requests.
+func NewUtlsHTTPClient(cfg *config.Config, auth *cliproxyauth.Auth, timeout time.Duration) *http.Client {
+	var proxyURL string
+	if auth != nil {
+		proxyURL = strings.TrimSpace(auth.ProxyURL)
+	}
+	if proxyURL == "" && cfg != nil {
+		proxyURL = strings.TrimSpace(cfg.ProxyURL)
+	}
+
+	utlsRT := newUtlsRoundTripper(proxyURL)
+
+	var standardTransport http.RoundTripper = &http.Transport{
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext,
+	}
+	if proxyURL != "" {
+		if transport := buildProxyTransport(proxyURL); transport != nil {
+			standardTransport = transport
+		}
+	}
+
+	client := &http.Client{
+		Transport: &fallbackRoundTripper{
+			utls:     utlsRT,
+			fallback: standardTransport,
+		},
+	}
+	if timeout > 0 {
+		client.Timeout = timeout
+	}
+	return client
+}