package executor import ( "bufio" "bytes" "context" "crypto/hmac" "crypto/sha256" "encoding/hex" "fmt" "io" "net/http" "strings" "time" "github.com/google/uuid" iflowauth "github.com/router-for-me/CLIProxyAPI/v6/internal/auth/iflow" "github.com/router-for-me/CLIProxyAPI/v6/internal/config" "github.com/router-for-me/CLIProxyAPI/v6/internal/runtime/executor/helps" "github.com/router-for-me/CLIProxyAPI/v6/internal/thinking" "github.com/router-for-me/CLIProxyAPI/v6/internal/util" cliproxyauth "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/auth" cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/executor" sdktranslator "github.com/router-for-me/CLIProxyAPI/v6/sdk/translator" log "github.com/sirupsen/logrus" "github.com/tidwall/gjson" "github.com/tidwall/sjson") const ( iflowDefaultEndpoint = "/chat/completions" iflowUserAgent = "iFlow-Cli") // IFlowExecutor executes OpenAI-compatible chat completions against the iFlow API using API keys derived from OAuth.type IFlowExecutor struct { cfg *config.Config} // NewIFlowExecutor constructs a new executor instance.func NewIFlowExecutor(cfg *config.Config) *IFlowExecutor { return &IFlowExecutor{cfg: cfg} } // Identifier returns the provider key.func (e *IFlowExecutor) Identifier() string { return "iflow" } // PrepareRequest injects iFlow credentials into the outgoing HTTP request.func (e *IFlowExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error { if req == nil { return nil } apiKey, _ := iflowCreds(auth) if strings.TrimSpace(apiKey) != "" { req.Header.Set("Authorization", "Bearer "+apiKey) } return nil} // HttpRequest injects iFlow credentials into the request and executes it.func (e *IFlowExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) { if req == nil { return nil, fmt.Errorf("iflow executor: request is nil") } if ctx == nil { ctx = req.Context() } httpReq := req.WithContext(ctx) if err := e.PrepareRequest(httpReq, auth); err != nil { return nil, err } httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) return httpClient.Do(httpReq)} // Execute performs a non-streaming chat completion request.func (e *IFlowExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { if opts.Alt == "responses/compact" { return resp, statusErr{code: http.StatusNotImplemented, msg: "/responses/compact not supported"} } baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := iflowCreds(auth) if strings.TrimSpace(apiKey) == "" { err = fmt.Errorf("iflow executor: missing api key") return resp, err } if baseURL == "" { baseURL = iflowauth.DefaultAPIBaseURL } reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth) defer reporter.TrackFailure(ctx, &err) from := opts.SourceFormat to := sdktranslator.FromString("openai") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, false) body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false) body, _ = sjson.SetBytes(body, "model", baseModel) body, err = thinking.ApplyThinking(body, req.Model, from.String(), "iflow", e.Identifier()) if err != nil { return resp, err } body = preserveReasoningContentInMessages(body) requestedModel := helps.PayloadRequestedModel(opts, req.Model) body = helps.ApplyPayloadConfigWithRoot(e.cfg, baseModel, to.String(), "", body, originalTranslated, requestedModel) endpoint := strings.TrimSuffix(baseURL, "/") + iflowDefaultEndpoint httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) if err != nil { return resp, err } applyIFlowHeaders(httpReq, apiKey, false) var attrs map[string]string if auth != nil { attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(httpReq, attrs) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: endpoint, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("iflow executor: close response body error: %v", errClose) } }() helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { b, _ := io.ReadAll(httpResp.Body) helps.AppendAPIResponseChunk(ctx, e.cfg, b) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) err = statusErr{code: httpResp.StatusCode, msg: string(b)} return resp, err } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) reporter.Publish(ctx, helps.ParseOpenAIUsage(data)) // Ensure usage is recorded even if upstream omits usage metadata. reporter.EnsurePublished(ctx) var param any // Note: TranslateNonStream uses req.Model (original with suffix) to preserve // the original model name in the response for client compatibility. out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, opts.OriginalRequest, body, data, ¶m) resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil} // ExecuteStream performs a streaming chat completion request.func (e *IFlowExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) { if opts.Alt == "responses/compact" { return nil, statusErr{code: http.StatusNotImplemented, msg: "/responses/compact not supported"} } baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := iflowCreds(auth) if strings.TrimSpace(apiKey) == "" { err = fmt.Errorf("iflow executor: missing api key") return nil, err } if baseURL == "" { baseURL = iflowauth.DefaultAPIBaseURL } reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth) defer reporter.TrackFailure(ctx, &err) from := opts.SourceFormat to := sdktranslator.FromString("openai") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, true) body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, true) body, _ = sjson.SetBytes(body, "model", baseModel) body, err = thinking.ApplyThinking(body, req.Model, from.String(), "iflow", e.Identifier()) if err != nil { return nil, err } body = preserveReasoningContentInMessages(body) // Ensure tools array exists to avoid provider quirks observed in some upstreams. toolsResult := gjson.GetBytes(body, "tools") if toolsResult.Exists() && toolsResult.IsArray() && len(toolsResult.Array()) == 0 { body = ensureToolsArray(body) } requestedModel := helps.PayloadRequestedModel(opts, req.Model) body = helps.ApplyPayloadConfigWithRoot(e.cfg, baseModel, to.String(), "", body, originalTranslated, requestedModel) endpoint := strings.TrimSuffix(baseURL, "/") + iflowDefaultEndpoint httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) if err != nil { return nil, err } applyIFlowHeaders(httpReq, apiKey, true) var attrs map[string]string if auth != nil { attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(httpReq, attrs) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: endpoint, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return nil, err } helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { data, _ := io.ReadAll(httpResp.Body) if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("iflow executor: close response body error: %v", errClose) } helps.AppendAPIResponseChunk(ctx, e.cfg, data) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), data)) err = statusErr{code: httpResp.StatusCode, msg: string(data)} return nil, err } out := make(chan cliproxyexecutor.StreamChunk) go func() { defer close(out) defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("iflow executor: close response body error: %v", errClose) } }() scanner := bufio.NewScanner(httpResp.Body) scanner.Buffer(nil, 52_428_800) // 50MB var param any for scanner.Scan() { line := scanner.Bytes() helps.AppendAPIResponseChunk(ctx, e.cfg, line) if detail, ok := helps.ParseOpenAIStreamUsage(line); ok { reporter.Publish(ctx, detail) } chunks := sdktranslator.TranslateStream(ctx, to, from, req.Model, opts.OriginalRequest, body, bytes.Clone(line), ¶m) for i := range chunks { out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]} } } if errScan := scanner.Err(); errScan != nil { helps.RecordAPIResponseError(ctx, e.cfg, errScan) reporter.PublishFailure(ctx) out <- cliproxyexecutor.StreamChunk{Err: errScan} } // Guarantee a usage record exists even if the stream never emitted usage data. reporter.EnsurePublished(ctx) }() return &cliproxyexecutor.StreamResult{Headers: httpResp.Header.Clone(), Chunks: out}, nil} func (e *IFlowExecutor) CountTokens(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (cliproxyexecutor.Response, error) { baseModel := thinking.ParseSuffix(req.Model).ModelName from := opts.SourceFormat to := sdktranslator.FromString("openai") body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false) enc, err := helps.TokenizerForModel(baseModel) if err != nil { return cliproxyexecutor.Response{}, fmt.Errorf("iflow executor: tokenizer init failed: %w", err) } count, err := helps.CountOpenAIChatTokens(enc, body) if err != nil { return cliproxyexecutor.Response{}, fmt.Errorf("iflow executor: token counting failed: %w", err) } usageJSON := helps.BuildOpenAIUsageJSON(count) translated := sdktranslator.TranslateTokenCount(ctx, to, from, count, usageJSON) return cliproxyexecutor.Response{Payload: translated}, nil} // Refresh refreshes OAuth tokens or cookie-based API keys and updates the stored API key.func (e *IFlowExecutor) Refresh(ctx context.Context, auth *cliproxyauth.Auth) (*cliproxyauth.Auth, error) { log.Debugf("iflow executor: refresh called") if auth == nil { return nil, fmt.Errorf("iflow executor: auth is nil") } // Check if this is cookie-based authentication var cookie string var email string if auth.Metadata != nil { if v, ok := auth.Metadata["cookie"].(string); ok { cookie = strings.TrimSpace(v) } if v, ok := auth.Metadata["email"].(string); ok { email = strings.TrimSpace(v) } } // If cookie is present, use cookie-based refresh if cookie != "" && email != "" { return e.refreshCookieBased(ctx, auth, cookie, email) } // Otherwise, use OAuth-based refresh return e.refreshOAuthBased(ctx, auth)} // refreshCookieBased refreshes API key using browser cookiefunc (e *IFlowExecutor) refreshCookieBased(ctx context.Context, auth *cliproxyauth.Auth, cookie, email string) (*cliproxyauth.Auth, error) { log.Debugf("iflow executor: checking refresh need for cookie-based API key for user: %s", email) // Get current expiry time from metadata var currentExpire string if auth.Metadata != nil { if v, ok := auth.Metadata["expired"].(string); ok { currentExpire = strings.TrimSpace(v) } } // Check if refresh is needed needsRefresh, _, err := iflowauth.ShouldRefreshAPIKey(currentExpire) if err != nil { log.Warnf("iflow executor: failed to check refresh need: %v", err) // If we can't check, continue with refresh anyway as a safety measure } else if !needsRefresh { log.Debugf("iflow executor: no refresh needed for user: %s", email) return auth, nil } log.Infof("iflow executor: refreshing cookie-based API key for user: %s", email) svc := iflowauth.NewIFlowAuth(e.cfg) keyData, err := svc.RefreshAPIKey(ctx, cookie, email) if err != nil { log.Errorf("iflow executor: cookie-based API key refresh failed: %v", err) return nil, err } if auth.Metadata == nil { auth.Metadata = make(map[string]any) } auth.Metadata["api_key"] = keyData.APIKey auth.Metadata["expired"] = keyData.ExpireTime auth.Metadata["type"] = "iflow" auth.Metadata["last_refresh"] = time.Now().Format(time.RFC3339) auth.Metadata["cookie"] = cookie auth.Metadata["email"] = email log.Infof("iflow executor: cookie-based API key refreshed successfully, new expiry: %s", keyData.ExpireTime) if auth.Attributes == nil { auth.Attributes = make(map[string]string) } auth.Attributes["api_key"] = keyData.APIKey return auth, nil} // refreshOAuthBased refreshes tokens using OAuth refresh tokenfunc (e *IFlowExecutor) refreshOAuthBased(ctx context.Context, auth *cliproxyauth.Auth) (*cliproxyauth.Auth, error) { refreshToken := "" oldAccessToken := "" if auth.Metadata != nil { if v, ok := auth.Metadata["refresh_token"].(string); ok { refreshToken = strings.TrimSpace(v) } if v, ok := auth.Metadata["access_token"].(string); ok { oldAccessToken = strings.TrimSpace(v) } } if refreshToken == "" { return auth, nil } // Log the old access token (masked) before refresh if oldAccessToken != "" { log.Debugf("iflow executor: refreshing access token, old: %s", util.HideAPIKey(oldAccessToken)) } svc := iflowauth.NewIFlowAuth(e.cfg) tokenData, err := svc.RefreshTokens(ctx, refreshToken) if err != nil { log.Errorf("iflow executor: token refresh failed: %v", err) return nil, err } if auth.Metadata == nil { auth.Metadata = make(map[string]any) } auth.Metadata["access_token"] = tokenData.AccessToken if tokenData.RefreshToken != "" { auth.Metadata["refresh_token"] = tokenData.RefreshToken } if tokenData.APIKey != "" { auth.Metadata["api_key"] = tokenData.APIKey } auth.Metadata["expired"] = tokenData.Expire auth.Metadata["type"] = "iflow" auth.Metadata["last_refresh"] = time.Now().Format(time.RFC3339) // Log the new access token (masked) after successful refresh log.Debugf("iflow executor: token refresh successful, new: %s", util.HideAPIKey(tokenData.AccessToken)) if auth.Attributes == nil { auth.Attributes = make(map[string]string) } if tokenData.APIKey != "" { auth.Attributes["api_key"] = tokenData.APIKey } return auth, nil} func applyIFlowHeaders(r *http.Request, apiKey string, stream bool) { r.Header.Set("Content-Type", "application/json") r.Header.Set("Authorization", "Bearer "+apiKey) r.Header.Set("User-Agent", iflowUserAgent) // Generate session-id sessionID := "session-" + generateUUID() r.Header.Set("session-id", sessionID) // Generate timestamp and signature timestamp := time.Now().UnixMilli() r.Header.Set("x-iflow-timestamp", fmt.Sprintf("%d", timestamp)) signature := createIFlowSignature(iflowUserAgent, sessionID, timestamp, apiKey) if signature != "" { r.Header.Set("x-iflow-signature", signature) } if stream { r.Header.Set("Accept", "text/event-stream") } else { r.Header.Set("Accept", "application/json") }} // createIFlowSignature generates HMAC-SHA256 signature for iFlow API requests.// The signature payload format is: userAgent:sessionId:timestampfunc createIFlowSignature(userAgent, sessionID string, timestamp int64, apiKey string) string { if apiKey == "" { return "" } payload := fmt.Sprintf("%s:%s:%d", userAgent, sessionID, timestamp) h := hmac.New(sha256.New, []byte(apiKey)) h.Write([]byte(payload)) return hex.EncodeToString(h.Sum(nil))Lines 1–500 of 585. Load more
package executor import ( "bufio" "bytes" "context" "crypto/hmac" "crypto/sha256" "encoding/hex" "fmt" "io" "net/http" "strings" "time" "github.com/google/uuid" iflowauth "github.com/router-for-me/CLIProxyAPI/v6/internal/auth/iflow" "github.com/router-for-me/CLIProxyAPI/v6/internal/config" "github.com/router-for-me/CLIProxyAPI/v6/internal/runtime/executor/helps" "github.com/router-for-me/CLIProxyAPI/v6/internal/thinking" "github.com/router-for-me/CLIProxyAPI/v6/internal/util" cliproxyauth "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/auth" cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/executor" sdktranslator "github.com/router-for-me/CLIProxyAPI/v6/sdk/translator" log "github.com/sirupsen/logrus" "github.com/tidwall/gjson" "github.com/tidwall/sjson" ) const ( iflowDefaultEndpoint = "/chat/completions" iflowUserAgent = "iFlow-Cli" ) // IFlowExecutor executes OpenAI-compatible chat completions against the iFlow API using API keys derived from OAuth. type IFlowExecutor struct { cfg *config.Config } // NewIFlowExecutor constructs a new executor instance. func NewIFlowExecutor(cfg *config.Config) *IFlowExecutor { return &IFlowExecutor{cfg: cfg} } // Identifier returns the provider key. func (e *IFlowExecutor) Identifier() string { return "iflow" } // PrepareRequest injects iFlow credentials into the outgoing HTTP request. func (e *IFlowExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error { if req == nil { return nil } apiKey, _ := iflowCreds(auth) if strings.TrimSpace(apiKey) != "" { req.Header.Set("Authorization", "Bearer "+apiKey) } return nil } // HttpRequest injects iFlow credentials into the request and executes it. func (e *IFlowExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) { if req == nil { return nil, fmt.Errorf("iflow executor: request is nil") } if ctx == nil { ctx = req.Context() } httpReq := req.WithContext(ctx) if err := e.PrepareRequest(httpReq, auth); err != nil { return nil, err } httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) return httpClient.Do(httpReq) } // Execute performs a non-streaming chat completion request. func (e *IFlowExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { if opts.Alt == "responses/compact" { return resp, statusErr{code: http.StatusNotImplemented, msg: "/responses/compact not supported"} } baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := iflowCreds(auth) if strings.TrimSpace(apiKey) == "" { err = fmt.Errorf("iflow executor: missing api key") return resp, err } if baseURL == "" { baseURL = iflowauth.DefaultAPIBaseURL } reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth) defer reporter.TrackFailure(ctx, &err) from := opts.SourceFormat to := sdktranslator.FromString("openai") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, false) body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false) body, _ = sjson.SetBytes(body, "model", baseModel) body, err = thinking.ApplyThinking(body, req.Model, from.String(), "iflow", e.Identifier()) if err != nil { return resp, err } body = preserveReasoningContentInMessages(body) requestedModel := helps.PayloadRequestedModel(opts, req.Model) body = helps.ApplyPayloadConfigWithRoot(e.cfg, baseModel, to.String(), "", body, originalTranslated, requestedModel) endpoint := strings.TrimSuffix(baseURL, "/") + iflowDefaultEndpoint httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) if err != nil { return resp, err } applyIFlowHeaders(httpReq, apiKey, false) var attrs map[string]string if auth != nil { attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(httpReq, attrs) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: endpoint, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("iflow executor: close response body error: %v", errClose) } }() helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { b, _ := io.ReadAll(httpResp.Body) helps.AppendAPIResponseChunk(ctx, e.cfg, b) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) err = statusErr{code: httpResp.StatusCode, msg: string(b)} return resp, err } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) reporter.Publish(ctx, helps.ParseOpenAIUsage(data)) // Ensure usage is recorded even if upstream omits usage metadata. reporter.EnsurePublished(ctx) var param any // Note: TranslateNonStream uses req.Model (original with suffix) to preserve // the original model name in the response for client compatibility. out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, opts.OriginalRequest, body, data, ¶m) resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil } // ExecuteStream performs a streaming chat completion request. func (e *IFlowExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) { if opts.Alt == "responses/compact" { return nil, statusErr{code: http.StatusNotImplemented, msg: "/responses/compact not supported"} } baseModel := thinking.ParseSuffix(req.Model).ModelName apiKey, baseURL := iflowCreds(auth) if strings.TrimSpace(apiKey) == "" { err = fmt.Errorf("iflow executor: missing api key") return nil, err } if baseURL == "" { baseURL = iflowauth.DefaultAPIBaseURL } reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth) defer reporter.TrackFailure(ctx, &err) from := opts.SourceFormat to := sdktranslator.FromString("openai") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := originalPayloadSource originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, true) body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, true) body, _ = sjson.SetBytes(body, "model", baseModel) body, err = thinking.ApplyThinking(body, req.Model, from.String(), "iflow", e.Identifier()) if err != nil { return nil, err } body = preserveReasoningContentInMessages(body) // Ensure tools array exists to avoid provider quirks observed in some upstreams. toolsResult := gjson.GetBytes(body, "tools") if toolsResult.Exists() && toolsResult.IsArray() && len(toolsResult.Array()) == 0 { body = ensureToolsArray(body) } requestedModel := helps.PayloadRequestedModel(opts, req.Model) body = helps.ApplyPayloadConfigWithRoot(e.cfg, baseModel, to.String(), "", body, originalTranslated, requestedModel) endpoint := strings.TrimSuffix(baseURL, "/") + iflowDefaultEndpoint httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) if err != nil { return nil, err } applyIFlowHeaders(httpReq, apiKey, true) var attrs map[string]string if auth != nil { attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(httpReq, attrs) var authID, authLabel, authType, authValue string if auth != nil { authID = auth.ID authLabel = auth.Label authType, authValue = auth.AccountInfo() } helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ URL: endpoint, Method: http.MethodPost, Headers: httpReq.Header.Clone(), Body: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return nil, err } helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { data, _ := io.ReadAll(httpResp.Body) if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("iflow executor: close response body error: %v", errClose) } helps.AppendAPIResponseChunk(ctx, e.cfg, data) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), data)) err = statusErr{code: httpResp.StatusCode, msg: string(data)} return nil, err } out := make(chan cliproxyexecutor.StreamChunk) go func() { defer close(out) defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("iflow executor: close response body error: %v", errClose) } }() scanner := bufio.NewScanner(httpResp.Body) scanner.Buffer(nil, 52_428_800) // 50MB var param any for scanner.Scan() { line := scanner.Bytes() helps.AppendAPIResponseChunk(ctx, e.cfg, line) if detail, ok := helps.ParseOpenAIStreamUsage(line); ok { reporter.Publish(ctx, detail) } chunks := sdktranslator.TranslateStream(ctx, to, from, req.Model, opts.OriginalRequest, body, bytes.Clone(line), ¶m) for i := range chunks { out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]} } } if errScan := scanner.Err(); errScan != nil { helps.RecordAPIResponseError(ctx, e.cfg, errScan) reporter.PublishFailure(ctx) out <- cliproxyexecutor.StreamChunk{Err: errScan} } // Guarantee a usage record exists even if the stream never emitted usage data. reporter.EnsurePublished(ctx) }() return &cliproxyexecutor.StreamResult{Headers: httpResp.Header.Clone(), Chunks: out}, nil } func (e *IFlowExecutor) CountTokens(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (cliproxyexecutor.Response, error) { baseModel := thinking.ParseSuffix(req.Model).ModelName from := opts.SourceFormat to := sdktranslator.FromString("openai") body := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false) enc, err := helps.TokenizerForModel(baseModel) if err != nil { return cliproxyexecutor.Response{}, fmt.Errorf("iflow executor: tokenizer init failed: %w", err) } count, err := helps.CountOpenAIChatTokens(enc, body) if err != nil { return cliproxyexecutor.Response{}, fmt.Errorf("iflow executor: token counting failed: %w", err) } usageJSON := helps.BuildOpenAIUsageJSON(count) translated := sdktranslator.TranslateTokenCount(ctx, to, from, count, usageJSON) return cliproxyexecutor.Response{Payload: translated}, nil } // Refresh refreshes OAuth tokens or cookie-based API keys and updates the stored API key. func (e *IFlowExecutor) Refresh(ctx context.Context, auth *cliproxyauth.Auth) (*cliproxyauth.Auth, error) { log.Debugf("iflow executor: refresh called") if auth == nil { return nil, fmt.Errorf("iflow executor: auth is nil") } // Check if this is cookie-based authentication var cookie string var email string if auth.Metadata != nil { if v, ok := auth.Metadata["cookie"].(string); ok { cookie = strings.TrimSpace(v) } if v, ok := auth.Metadata["email"].(string); ok { email = strings.TrimSpace(v) } } // If cookie is present, use cookie-based refresh if cookie != "" && email != "" { return e.refreshCookieBased(ctx, auth, cookie, email) } // Otherwise, use OAuth-based refresh return e.refreshOAuthBased(ctx, auth) } // refreshCookieBased refreshes API key using browser cookie func (e *IFlowExecutor) refreshCookieBased(ctx context.Context, auth *cliproxyauth.Auth, cookie, email string) (*cliproxyauth.Auth, error) { log.Debugf("iflow executor: checking refresh need for cookie-based API key for user: %s", email) // Get current expiry time from metadata var currentExpire string if auth.Metadata != nil { if v, ok := auth.Metadata["expired"].(string); ok { currentExpire = strings.TrimSpace(v) } } // Check if refresh is needed needsRefresh, _, err := iflowauth.ShouldRefreshAPIKey(currentExpire) if err != nil { log.Warnf("iflow executor: failed to check refresh need: %v", err) // If we can't check, continue with refresh anyway as a safety measure } else if !needsRefresh { log.Debugf("iflow executor: no refresh needed for user: %s", email) return auth, nil } log.Infof("iflow executor: refreshing cookie-based API key for user: %s", email) svc := iflowauth.NewIFlowAuth(e.cfg) keyData, err := svc.RefreshAPIKey(ctx, cookie, email) if err != nil { log.Errorf("iflow executor: cookie-based API key refresh failed: %v", err) return nil, err } if auth.Metadata == nil { auth.Metadata = make(map[string]any) } auth.Metadata["api_key"] = keyData.APIKey auth.Metadata["expired"] = keyData.ExpireTime auth.Metadata["type"] = "iflow" auth.Metadata["last_refresh"] = time.Now().Format(time.RFC3339) auth.Metadata["cookie"] = cookie auth.Metadata["email"] = email log.Infof("iflow executor: cookie-based API key refreshed successfully, new expiry: %s", keyData.ExpireTime) if auth.Attributes == nil { auth.Attributes = make(map[string]string) } auth.Attributes["api_key"] = keyData.APIKey return auth, nil } // refreshOAuthBased refreshes tokens using OAuth refresh token func (e *IFlowExecutor) refreshOAuthBased(ctx context.Context, auth *cliproxyauth.Auth) (*cliproxyauth.Auth, error) { refreshToken := "" oldAccessToken := "" if auth.Metadata != nil { if v, ok := auth.Metadata["refresh_token"].(string); ok { refreshToken = strings.TrimSpace(v) } if v, ok := auth.Metadata["access_token"].(string); ok { oldAccessToken = strings.TrimSpace(v) } } if refreshToken == "" { return auth, nil } // Log the old access token (masked) before refresh if oldAccessToken != "" { log.Debugf("iflow executor: refreshing access token, old: %s", util.HideAPIKey(oldAccessToken)) } svc := iflowauth.NewIFlowAuth(e.cfg) tokenData, err := svc.RefreshTokens(ctx, refreshToken) if err != nil { log.Errorf("iflow executor: token refresh failed: %v", err) return nil, err } if auth.Metadata == nil { auth.Metadata = make(map[string]any) } auth.Metadata["access_token"] = tokenData.AccessToken if tokenData.RefreshToken != "" { auth.Metadata["refresh_token"] = tokenData.RefreshToken } if tokenData.APIKey != "" { auth.Metadata["api_key"] = tokenData.APIKey } auth.Metadata["expired"] = tokenData.Expire auth.Metadata["type"] = "iflow" auth.Metadata["last_refresh"] = time.Now().Format(time.RFC3339) // Log the new access token (masked) after successful refresh log.Debugf("iflow executor: token refresh successful, new: %s", util.HideAPIKey(tokenData.AccessToken)) if auth.Attributes == nil { auth.Attributes = make(map[string]string) } if tokenData.APIKey != "" { auth.Attributes["api_key"] = tokenData.APIKey } return auth, nil } func applyIFlowHeaders(r *http.Request, apiKey string, stream bool) { r.Header.Set("Content-Type", "application/json") r.Header.Set("Authorization", "Bearer "+apiKey) r.Header.Set("User-Agent", iflowUserAgent) // Generate session-id sessionID := "session-" + generateUUID() r.Header.Set("session-id", sessionID) // Generate timestamp and signature timestamp := time.Now().UnixMilli() r.Header.Set("x-iflow-timestamp", fmt.Sprintf("%d", timestamp)) signature := createIFlowSignature(iflowUserAgent, sessionID, timestamp, apiKey) if signature != "" { r.Header.Set("x-iflow-signature", signature) } if stream { r.Header.Set("Accept", "text/event-stream") } else { r.Header.Set("Accept", "application/json") } } // createIFlowSignature generates HMAC-SHA256 signature for iFlow API requests. // The signature payload format is: userAgent:sessionId:timestamp func createIFlowSignature(userAgent, sessionID string, timestamp int64, apiKey string) string { if apiKey == "" { return "" } payload := fmt.Sprintf("%s:%s:%d", userAgent, sessionID, timestamp) h := hmac.New(sha256.New, []byte(apiKey)) h.Write([]byte(payload)) return hex.EncodeToString(h.Sum(nil))