Something went wrong. Try again.
sloproxy
Something went wrong. Try again.
13 kB · 399 lines
Go
at commit e8e39526
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400// Package api provides the HTTP API server implementation for the CLI Proxy API.// It includes the main server struct, routing setup, middleware for CORS and authentication,// and integration with various AI API handlers (OpenAI, Claude, Gemini).// The server supports hot-reloading of clients and configuration.package api
import ( "context" "crypto/tls" "errors" "fmt" "net" "net/http" "os" "strings" "sync" "sync/atomic" "time"
"github.com/gin-gonic/gin" managementHandlers "github.com/router-for-me/CLIProxyAPI/v7/internal/api/handlers/management" "github.com/router-for-me/CLIProxyAPI/v7/internal/api/middleware" codexlive "github.com/router-for-me/CLIProxyAPI/v7/internal/client/codex/live" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" "github.com/router-for-me/CLIProxyAPI/v7/internal/logging" "github.com/router-for-me/CLIProxyAPI/v7/internal/managementasset" "github.com/router-for-me/CLIProxyAPI/v7/internal/pluginhost" "github.com/router-for-me/CLIProxyAPI/v7/internal/redisqueue" sdkaccess "github.com/router-for-me/CLIProxyAPI/v7/sdk/access" "github.com/router-for-me/CLIProxyAPI/v7/sdk/api/handlers" "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" log "github.com/sirupsen/logrus" "golang.org/x/net/http2" "gopkg.in/yaml.v3")
// Server represents the main API server.// It encapsulates the Gin engine, HTTP server, handlers, and configuration.type Server struct { // engine is the Gin web framework engine instance. engine *gin.Engine
// server is the underlying HTTP server. server *http.Server
// muxBaseListener is the shared TCP listener used to serve both HTTP and Redis protocol traffic. muxBaseListener net.Listener
// muxHTTPListener receives HTTP connections selected by the multiplexer. muxHTTPListener *muxListener
// handlers contains the API handlers for processing requests. handlers *handlers.BaseAPIHandler codexLiveHandler *codexlive.Handler
// cfg holds the current server configuration. cfg *config.Config
// oldConfigYaml stores a YAML snapshot of the previous configuration for change detection. // This prevents issues when the config object is modified in place by Management API. oldConfigYaml []byte
// accessManager handles request authentication providers. accessManager *sdkaccess.Manager
// requestLogger is the request logger instance for dynamic configuration updates. requestLogger logging.RequestLogger loggerToggle func(bool)
// configFilePath is the absolute path to the YAML config file for persistence. configFilePath string
// currentPath is the absolute path to the current working directory. currentPath string
// wsRoutes tracks registered websocket upgrade paths. wsRouteMu sync.Mutex wsRoutes map[string]struct{} wsAuthChanged func(bool, bool) wsAuthEnabled atomic.Bool
// management handler mgmt *managementHandlers.Handler
// pluginHost owns dynamic plugin Management API route dispatch. pluginHost *pluginhost.Host
// managementRoutesRegistered tracks whether the management routes have been attached to the engine. managementRoutesRegistered atomic.Bool // managementRoutesEnabled controls whether management endpoints serve real handlers. managementRoutesEnabled atomic.Bool
// envManagementSecret indicates whether MANAGEMENT_PASSWORD is configured. envManagementSecret bool
localPassword string
keepAliveEnabled bool keepAliveTimeout time.Duration keepAliveOnTimeout func() keepAliveHeartbeat chan struct{} keepAliveStop chan struct{}
exampleAPIKeySafeModeEnabled bool exampleAPIKeySafeModeActive atomic.Bool}
// NewServer creates and initializes a new API server instance.// It sets up the Gin engine, middleware, routes, and handlers.//// Parameters:// - cfg: The server configuration// - authManager: core runtime auth manager// - accessManager: request authentication manager//// Returns:// - *Server: A new server instancefunc NewServer(cfg *config.Config, authManager *auth.Manager, accessManager *sdkaccess.Manager, configFilePath string, opts ...ServerOption) *Server { optionState := &serverOptionConfig{ requestLoggerFactory: defaultRequestLoggerFactory, } for i := range opts { opts[i](optionState) } // Set gin mode if !cfg.Debug { gin.SetMode(gin.ReleaseMode) }
// Create gin engine engine := gin.New() if optionState.engineConfigurator != nil { optionState.engineConfigurator(engine) }
// Add middleware engine.Use(logging.GinLogrusLogger()) engine.Use(logging.GinLogrusRecovery()) engine.Use(logging.CPATraceIDMiddleware()) for _, mw := range optionState.extraMiddleware { engine.Use(mw) }
// Add request logging middleware (positioned after recovery, before auth) // Resolve logs directory relative to the configuration file directory. var requestLogger logging.RequestLogger var toggle func(bool) if !cfg.CommercialMode { if optionState.requestLoggerFactory != nil { requestLogger = optionState.requestLoggerFactory(cfg, configFilePath) } if requestLogger != nil { engine.Use(middleware.RequestLoggingMiddleware(requestLogger)) if setter, ok := requestLogger.(interface{ SetEnabled(bool) }); ok { toggle = setter.SetEnabled } } }
engine.Use(corsMiddleware()) wd, err := os.Getwd() if err != nil { wd = configFilePath }
envAdminPassword, envAdminPasswordSet := os.LookupEnv("MANAGEMENT_PASSWORD") envAdminPassword = strings.TrimSpace(envAdminPassword) envManagementSecret := envAdminPasswordSet && envAdminPassword != ""
// Create server instance s := &Server{ engine: engine, handlers: handlers.NewBaseAPIHandlers(effectiveSDKConfig(cfg), authManager), cfg: cfg, accessManager: accessManager, requestLogger: requestLogger, loggerToggle: toggle, configFilePath: configFilePath, currentPath: wd, envManagementSecret: envManagementSecret, wsRoutes: make(map[string]struct{}), pluginHost: optionState.pluginHost,
exampleAPIKeySafeModeEnabled: optionState.exampleAPIKeySafeMode, } s.wsAuthEnabled.Store(cfg.WebsocketAuth) s.exampleAPIKeySafeModeActive.Store(s.exampleAPIKeySafeModeRequired(cfg)) s.handlers.SetPluginHost(optionState.pluginHost) if optionState.pluginHost != nil { optionState.pluginHost.SetModelExecutor(s.handlers) optionState.pluginHost.SetAuthManager(authManager) } // Save initial YAML snapshot s.oldConfigYaml, _ = yaml.Marshal(cfg) s.applyAccessConfig(nil, cfg) if authManager != nil { authManager.SetRetryConfig(cfg.RequestRetry, time.Duration(cfg.MaxRetryInterval)*time.Second, cfg.MaxRetryCredentials) } managementasset.SetCurrentConfig(cfg) auth.SetQuotaCooldownDisabled(cfg.DisableCooling) auth.SetTransientErrorCooldownSeconds(cfg.TransientErrorCooldownSeconds) applySignatureCacheConfig(nil, cfg) // Initialize management handler s.mgmt = managementHandlers.NewHandler(cfg, configFilePath, authManager) s.mgmt.SetPluginHost(optionState.pluginHost) s.mgmt.SetConfigReloadHook(optionState.configReloadHook) if optionState.localPassword != "" { s.mgmt.SetLocalPassword(optionState.localPassword) } logDir := logging.ResolveLogDirectory(cfg) s.mgmt.SetLogDirectory(logDir) if optionState.postAuthHook != nil { s.mgmt.SetPostAuthHook(optionState.postAuthHook) } if optionState.postAuthPersistHook != nil { s.mgmt.SetPostAuthPersistHook(optionState.postAuthPersistHook) } s.localPassword = optionState.localPassword
// Home heartbeat gate: when home is enabled, block all endpoints with 503 until the // subscribe-config heartbeat connection is healthy. engine.Use(s.homeHeartbeatMiddleware()) engine.Use(s.exampleAPIKeySafeModeMiddleware())
// Setup routes s.setupRoutes()
// Apply additional router configurators from options if optionState.routerConfigurator != nil { optionState.routerConfigurator(engine, s.handlers, cfg) }
// Register management routes when configuration or environment secrets are available, // or when a local management password is provided (e.g. TUI mode). hasManagementSecret := cfg.RemoteManagement.SecretKey != "" || envManagementSecret || s.localPassword != "" s.managementRoutesEnabled.Store(hasManagementSecret) redisqueue.SetEnabled(hasManagementSecret || (cfg != nil && cfg.Home.Enabled)) if hasManagementSecret { s.registerManagementRoutes() } s.refreshPluginManagementRoutes() engine.NoRoute(s.pluginManagementNoRoute)
if optionState.keepAliveEnabled { s.enableKeepAlive(optionState.keepAliveTimeout, optionState.keepAliveOnTimeout) }
// Create HTTP server s.server = &http.Server{ Addr: fmt.Sprintf("%s:%d", cfg.Host, cfg.Port), Handler: engine, }
return s}
// Start begins listening for and serving HTTP or HTTPS requests.// It's a blocking call and will only return on an unrecoverable error.//// Returns:// - error: An error if the server fails to startfunc (s *Server) Start() error { if s == nil || s.server == nil { return fmt.Errorf("failed to start HTTP server: server not initialized") }
addr := s.server.Addr listener, errListen := net.Listen("tcp", addr) if errListen != nil { return fmt.Errorf("failed to start HTTP server: %v", errListen) }
useTLS := s.cfg != nil && s.cfg.TLS.Enable if useTLS { certPath := strings.TrimSpace(s.cfg.TLS.Cert) keyPath := strings.TrimSpace(s.cfg.TLS.Key) if certPath == "" || keyPath == "" { if errClose := listener.Close(); errClose != nil { log.Errorf("failed to close listener after TLS validation failure: %v", errClose) } return fmt.Errorf("failed to start HTTPS server: tls.cert or tls.key is empty") } certPair, errLoad := tls.LoadX509KeyPair(certPath, keyPath) if errLoad != nil { if errClose := listener.Close(); errClose != nil { log.Errorf("failed to close listener after TLS key pair load failure: %v", errClose) } return fmt.Errorf("failed to start HTTPS server: %v", errLoad) }
tlsConfig := &tls.Config{ Certificates: []tls.Certificate{certPair}, NextProtos: []string{"h2", "http/1.1"}, } s.server.TLSConfig = tlsConfig if errHTTP2 := http2.ConfigureServer(s.server, &http2.Server{}); errHTTP2 != nil { log.Warnf("failed to configure HTTP/2: %v", errHTTP2) } listener = tls.NewListener(listener, tlsConfig) log.Debugf("Starting API server on %s with TLS", addr) } else { log.Debugf("Starting API server on %s", addr) }
httpListener := newMuxListener(listener.Addr(), 1024) s.muxBaseListener = listener s.muxHTTPListener = httpListener
httpErrCh := make(chan error, 1) acceptErrCh := make(chan error, 1)
go func() { httpErrCh <- s.server.Serve(httpListener) }() go func() { acceptErrCh <- s.acceptMuxConnections(listener, httpListener) }()
select { case errServe := <-httpErrCh: if s.muxBaseListener != nil { if errClose := s.muxBaseListener.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) { log.Debugf("failed to close shared listener after HTTP serve exit: %v", errClose) } } if s.muxHTTPListener != nil { _ = s.muxHTTPListener.Close() } errAccept := <-acceptErrCh errServe = normalizeHTTPServeError(errServe) errAccept = normalizeListenerError(errAccept) if errServe != nil { return fmt.Errorf("failed to start HTTP server: %v", errServe) } if errAccept != nil { return fmt.Errorf("failed to start HTTP server: %v", errAccept) } return nil case errAccept := <-acceptErrCh: if s.muxHTTPListener != nil { _ = s.muxHTTPListener.Close() } if s.muxBaseListener != nil { if errClose := s.muxBaseListener.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) { log.Debugf("failed to close shared listener after accept loop exit: %v", errClose) } } errServe := <-httpErrCh errServe = normalizeHTTPServeError(errServe) errAccept = normalizeListenerError(errAccept) if errAccept != nil { return fmt.Errorf("failed to start HTTP server: %v", errAccept) } if errServe != nil { return fmt.Errorf("failed to start HTTP server: %v", errServe) } return nil }}
// Stop gracefully shuts down the API server without interrupting any// active connections.//// Parameters:// - ctx: The context for graceful shutdown//// Returns:// - error: An error if the server fails to stopfunc (s *Server) Stop(ctx context.Context) error { log.Debug("Stopping API server...")
if s.keepAliveEnabled { select { case s.keepAliveStop <- struct{}{}: default: } }
if s.muxHTTPListener != nil { _ = s.muxHTTPListener.Close() } if s.muxBaseListener != nil { if errClose := s.muxBaseListener.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) { log.Debugf("failed to close shared listener: %v", errClose) } }
// Shutdown the HTTP server. errShutdown := s.server.Shutdown(ctx) if s.codexLiveHandler != nil { s.codexLiveHandler.Close() } if errShutdown != nil { return fmt.Errorf("failed to shutdown HTTP server: %v", errShutdown) }
log.Debug("API server stopped") return nil}