From 3daa48e6a38ccbc37ddb2accd294a042ae58567f Mon Sep 17 00:00:00 2001 From: Rustem Kamalov Date: Wed, 1 Apr 2026 00:54:21 +0300 Subject: [PATCH] Proxy policy wiring and fail-closed execution - replace proxy wiring with global and per-engine tag policies - split stats endpoints and lock fail-closed proxy behavior with tests --- cmd/proxy_policy.go | 30 ++ cmd/root.go | 184 ++++++++---- cmd/search.go | 102 +++++-- cmd/serve.go | 320 +++++++++++++++++---- cmd/serve_test.go | 105 +++++++ config.yaml | 53 ++-- core/browser.go | 110 +++++-- core/proxy.go | 505 +++++++++++++++++++++++++-------- core/proxy_integration_test.go | 28 +- core/proxy_test.go | 194 ++++++++++--- core/resilient.go | 368 ++++++++++++++---------- core/retry.go | 13 +- core/server.go | 68 ++++- core/server_test.go | 423 ++++++++++++++++++++------- testutil/integration.go | 25 ++ 15 files changed, 1925 insertions(+), 603 deletions(-) create mode 100644 cmd/proxy_policy.go create mode 100644 cmd/serve_test.go create mode 100644 testutil/integration.go diff --git a/cmd/proxy_policy.go b/cmd/proxy_policy.go new file mode 100644 index 0000000..42aa88b --- /dev/null +++ b/cmd/proxy_policy.go @@ -0,0 +1,30 @@ +package cmd + +import ( + "strings" + + "github.com/karust/openserp/core" +) + +func buildEngineProxyPolicyMap() map[string]string { + return map[string]string{ + "google": config.GoogleConfig.Proxy, + "yandex": config.YandexConfig.Proxy, + "baidu": config.BaiduConfig.Proxy, + "bing": config.BingConfig.Proxy, + "duckduckgo": config.DuckDuckGoConfig.Proxy, + } +} + +func buildNormalizedProxyConfig(runtime string) (core.ProxyConfig, error) { + return core.NormalizeProxyConfig(core.ProxyConfig{ + Runtime: runtime, + Proxies: config.Proxies, + EnginePolicies: buildEngineProxyPolicyMap(), + }) +} + +func resolveEngineProxyPolicy(proxyCfg core.ProxyConfig, engineName string) core.ProxyPolicy { + engineKey := strings.ToLower(strings.TrimSpace(engineName)) + return core.ResolveEffectiveProxyPolicy(proxyCfg.Proxies.Global, proxyCfg.EnginePolicies[engineKey]) +} diff --git a/cmd/root.go b/cmd/root.go index a0afc98..12c2730 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -19,44 +19,47 @@ const ( ) type Config struct { - App AppConfig `mapstructure:"app"` - ProxyPool ProxyPoolConfig `mapstructure:"proxy_pool"` - Cache CacheConfig `mapstructure:"cache"` - Resilience ResilienceConfig `mapstructure:"resilience"` - CircuitBreaker CircuitBreakerConfig `mapstructure:"circuit_breaker"` - CORS CORSConfig `mapstructure:"cors"` - Config2Capcha Config2Captcha `mapstructure:"2captcha"` - GoogleConfig core.SearchEngineOptions `mapstructure:"google"` - YandexConfig core.SearchEngineOptions `mapstructure:"yandex"` - BaiduConfig core.SearchEngineOptions `mapstructure:"baidu"` - BingConfig core.SearchEngineOptions `mapstructure:"bing"` - DuckDuckGoConfig core.SearchEngineOptions `mapstructure:"duckduckgo"` + Server ServerConfig `mapstructure:"server"` + App AppConfig `mapstructure:"app"` + Proxies core.ProxiesConfig `mapstructure:"proxies"` + Cache CacheConfig `mapstructure:"cache"` + Resilience ResilienceConfig `mapstructure:"resilience"` + CircuitBreaker CircuitBreakerConfig `mapstructure:"circuit_breaker"` + CORS CORSConfig `mapstructure:"cors"` + Config2Capcha Config2Captcha `mapstructure:"2captcha"` + GoogleConfig EngineConfig `mapstructure:"google"` + YandexConfig EngineConfig `mapstructure:"yandex"` + BaiduConfig EngineConfig `mapstructure:"baidu"` + BingConfig EngineConfig `mapstructure:"bing"` + DuckDuckGoConfig EngineConfig `mapstructure:"duckduckgo"` } type Config2Captcha struct { ApiKey string `mapstructure:"apikey"` } -type AppConfig struct { +type ServerConfig struct { Host string `mapstructure:"host"` Port int `mapstructure:"port"` - Timeout int `mapstructure:"timeout"` ConfigPath string `mapstructure:"config_path"` + IsDebug bool `mapstructure:"debug"` + IsVerbose bool `mapstructure:"verbose"` + IsRawRequests bool `mapstructure:"raw_requests"` + Insecure bool `mapstructure:"insecure"` +} + +type AppConfig struct { + Timeout int `mapstructure:"timeout"` BrowserPath string `mapstructure:"browser_path"` IsBrowserHead bool `mapstructure:"head"` IsLeaveHead bool `mapstructure:"leave_head"` IsLeakless bool `mapstructure:"leakless"` - IsDebug bool `mapstructure:"debug"` - IsVerbose bool `mapstructure:"verbose"` - IsRawRequests bool `mapstructure:"raw_requests"` - ProxyURL string `mapstructure:"proxy"` - Insecure bool `mapstructure:"insecure"` IsStealth bool `mapstructure:"stealth"` } -type ProxyPoolConfig struct { - URLs []string `mapstructure:"urls"` - FailureThreshold int `mapstructure:"failure_threshold"` +type EngineConfig struct { + core.SearchEngineOptions `mapstructure:",squash"` + Proxy string `mapstructure:"proxy"` } type CacheConfig struct { @@ -86,11 +89,21 @@ type CORSConfig struct { var config = Config{} var flagToConfigKey = map[string]string{ - "config": "app.config_path", + "host": "server.host", + "port": "server.port", + "timeout": "app.timeout", + "config": "server.config_path", "browser-path": "app.browser_path", + "verbose": "server.verbose", + "debug": "server.debug", + "head": "app.head", + "leakless": "app.leakless", + "raw": "server.raw_requests", "leave": "app.leave_head", - "raw": "app.raw_requests", "2captcha_key": "2captcha.apikey", + "proxy": "proxies.global", + "stealth": "app.stealth", + "insecure": "server.insecure", "cache_ttl": "cache.ttl_seconds", "cache_max_size": "cache.max_size", "max_retries": "resilience.max_retries", @@ -107,22 +120,15 @@ var RootCmd = &cobra.Command{ Version: version, SilenceUsage: true, PersistentPreRunE: func(cmd *cobra.Command, args []string) error { - core.InitLogger(config.App.IsVerbose, config.App.IsDebug) - err := initializeConfig(cmd) if err != nil { return err } + core.InitLogger(config.Server.IsVerbose, config.Server.IsDebug) logrus.Debugf("Final config: %+v", config) return nil }, - - // Run: func(cmd *cobra.Command, args []string) { - // // Working with OutOrStdout/OutOrStderr allows us to unit test our command easier - // //out := cmd.OutOrStdout() - // logrus.Trace("Config:", config) - // }, } // Bind each cobra flag to its associated viper configuration (config file and environment variable) @@ -189,35 +195,107 @@ func initializeConfig(cmd *cobra.Command) error { // 3. Command flags (highest priority). Bind the current command's flags to viper bindFlags(cmd, v) + if err := validateRemovedConfigPaths(v); err != nil { + return err + } + // Dump Viper values to config struct + if err := validateEngineProxyTags(v); err != nil { + return err + } + err = v.Unmarshal(&config) if err != nil { return fmt.Errorf("cannot unmarshall config: %v", err) } - config.App.ProxyURL, err = core.NormalizeProxyURL(config.App.ProxyURL) + config.Proxies, err = core.NormalizeProxiesConfig(config.Proxies) if err != nil { - return fmt.Errorf("invalid app.proxy: %w", err) + return fmt.Errorf("invalid proxies config: %w", err) } - config.ProxyPool.URLs, err = core.NormalizeProxyURLs(config.ProxyPool.URLs) - if err != nil { - return fmt.Errorf("invalid proxy_pool.urls: %w", err) - } - if config.ProxyPool.FailureThreshold <= 0 { - config.ProxyPool.FailureThreshold = core.DefaultProxyPoolFailureThreshold - } - - if config.App.IsDebug { + if config.Server.IsDebug { logrus.Debug("Viper config:") v.Debug() } return nil } +func validateEngineProxyTags(v *viper.Viper) error { + for _, engineName := range []string{"google", "yandex", "baidu", "bing", "duckduckgo"} { + key := engineName + ".proxy" + if !v.IsSet(key) { + continue + } + + raw := v.Get(key) + tag, ok := raw.(string) + if !ok { + return fmt.Errorf("invalid %s.proxy config: proxy must be a string tag", engineName) + } + + if _, err := core.NormalizeProxyTag(tag); err != nil { + return fmt.Errorf("invalid %s.proxy config: %w", engineName, err) + } + } + + return nil +} + +func validateRemovedConfigPaths(v *viper.Viper) error { + legacyKeys := map[string]string{ + "app.proxy": "use proxies.global or proxies.entries with per-engine proxy tags instead", + "proxy_pool": "use proxies.entries and proxies.health.failure_threshold instead", + "proxy_pool.urls": "use proxies.entries instead", + "proxy_pool.failure_threshold": "use proxies.health.failure_threshold instead", + "app.host": "move to server.host", + "app.port": "move to server.port", + "app.debug": "move to server.debug", + "app.verbose": "move to server.verbose", + "app.raw_requests": "move to server.raw_requests", + "app.insecure": "move to server.insecure", + "proxies.defaults": "use proxies.global or per-engine proxy tags instead", + "proxies.defaults.mode": "use proxies.global or per-engine proxy tags instead", + "proxies.defaults.tag": "use per-engine proxy tags on each engine instead", + "google.proxy.mode": "use google.proxy: or omit it for direct mode", + "google.proxy.tag": "use google.proxy: ", + "yandex.proxy.mode": "use yandex.proxy: or omit it for direct mode", + "yandex.proxy.tag": "use yandex.proxy: ", + "baidu.proxy.mode": "use baidu.proxy: or omit it for direct mode", + "baidu.proxy.tag": "use baidu.proxy: ", + "bing.proxy.mode": "use bing.proxy: or omit it for direct mode", + "bing.proxy.tag": "use bing.proxy: ", + "duckduckgo.proxy.mode": "use duckduckgo.proxy: or omit it for direct mode", + "duckduckgo.proxy.tag": "use duckduckgo.proxy: ", + } + + for key, hint := range legacyKeys { + if v.IsSet(key) { + return fmt.Errorf("config key %q is removed in proxy v2: %s", key, hint) + } + } + return nil +} + func setConfigDefaults(v *viper.Viper) { - v.SetDefault("proxy_pool.urls", []string{}) - v.SetDefault("proxy_pool.failure_threshold", core.DefaultProxyPoolFailureThreshold) + v.SetDefault("server.host", "127.0.0.1") + v.SetDefault("server.port", 7070) + v.SetDefault("server.debug", false) + v.SetDefault("server.verbose", false) + v.SetDefault("server.raw_requests", false) + v.SetDefault("server.insecure", false) + + v.SetDefault("app.timeout", 30) + v.SetDefault("app.browser_path", "") + v.SetDefault("app.head", false) + v.SetDefault("app.leave_head", false) + v.SetDefault("app.leakless", false) + v.SetDefault("app.stealth", false) + + v.SetDefault("proxies.entries", []interface{}{}) + v.SetDefault("proxies.global", "") + v.SetDefault("proxies.health.failure_threshold", core.DefaultProxyFailureThreshold) + v.SetDefault("cache.ttl_seconds", 300) v.SetDefault("cache.max_size", 1000) // Keep stage2 defaults stable even when config file is absent. @@ -234,21 +312,21 @@ func setConfigDefaults(v *viper.Viper) { } func init() { - RootCmd.PersistentFlags().IntVarP(&config.App.Port, "port", "p", 7070, "Port number to run server") - RootCmd.PersistentFlags().StringVarP(&config.App.Host, "host", "a", "127.0.0.1", "Host address to run server") + RootCmd.PersistentFlags().IntVarP(&config.Server.Port, "port", "p", 7070, "Port number to run server") + RootCmd.PersistentFlags().StringVarP(&config.Server.Host, "host", "a", "127.0.0.1", "Host address to run server") RootCmd.PersistentFlags().IntVarP(&config.App.Timeout, "timeout", "t", 30, "Timeout to fail request") - RootCmd.PersistentFlags().StringVarP(&config.App.ConfigPath, "config", "c", "", "Configuration file path") + RootCmd.PersistentFlags().StringVarP(&config.Server.ConfigPath, "config", "c", "", "Configuration file path") RootCmd.PersistentFlags().StringVarP(&config.App.BrowserPath, "browser-path", "", "", "Custom browser binary path (Chrome/Chromium/Edge/Brave..)") - RootCmd.PersistentFlags().BoolVarP(&config.App.IsVerbose, "verbose", "v", false, "Use verbose output") - RootCmd.PersistentFlags().BoolVarP(&config.App.IsDebug, "debug", "d", false, "Use debug output. Disable headless browser") + RootCmd.PersistentFlags().BoolVarP(&config.Server.IsVerbose, "verbose", "v", false, "Use verbose output") + RootCmd.PersistentFlags().BoolVarP(&config.Server.IsDebug, "debug", "d", false, "Use debug output. Disable headless browser") RootCmd.PersistentFlags().BoolVarP(&config.App.IsBrowserHead, "head", "", false, "Enable browser UI") RootCmd.PersistentFlags().BoolVarP(&config.App.IsLeakless, "leakless", "l", false, "Use leakless mode to insure browser instances are closed after search") - RootCmd.PersistentFlags().BoolVarP(&config.App.IsRawRequests, "raw", "r", false, "Disable browser usage, use HTTP requests") + RootCmd.PersistentFlags().BoolVarP(&config.Server.IsRawRequests, "raw", "r", false, "Disable browser usage, use HTTP requests") RootCmd.PersistentFlags().BoolVarP(&config.App.IsLeaveHead, "leave", "", false, "Leave browser and tabs opened after search is made") RootCmd.PersistentFlags().StringVarP(&config.Config2Capcha.ApiKey, "2captcha_key", "", "", "2 captcha api key") - RootCmd.PersistentFlags().StringVarP(&config.App.ProxyURL, "proxy", "x", "", "HTTP/HTTPS/SOCKS5/SOCKS5H proxy URL (e.g. socks5h://127.0.0.1:1080)") + RootCmd.PersistentFlags().StringVarP(&config.Proxies.Global, "proxy", "x", "", "Force a single proxy for all engines (same as proxies.global)") RootCmd.PersistentFlags().BoolVarP(&config.App.IsStealth, "stealth", "s", false, "Use stealth browser plugin") - RootCmd.PersistentFlags().BoolVarP(&config.App.Insecure, "insecure", "k", false, "Allow insecure TLS connections") + RootCmd.PersistentFlags().BoolVarP(&config.Server.Insecure, "insecure", "k", false, "Allow insecure TLS connections") RootCmd.PersistentFlags().IntVar(&config.Cache.TTLSeconds, "cache_ttl", 300, "Cache TTL in seconds (0 to disable)") RootCmd.PersistentFlags().IntVar(&config.Cache.MaxSize, "cache_max_size", 1000, "Maximum number of cached responses") RootCmd.PersistentFlags().IntVar(&config.Resilience.MaxRetries, "max_retries", 3, "Max retry attempts per search engine (0 to disable)") diff --git a/cmd/search.go b/cmd/search.go index cf25624..9082b5d 100644 --- a/cmd/search.go +++ b/cmd/search.go @@ -25,25 +25,46 @@ var searchCMD = &cobra.Command{ } func search(cmd *cobra.Command, args []string) { - var err error - engineType := args[0] + engineType := normalizeEngineArg(args[0]) query := core.Query{ Text: args[1], Limit: 10, Filter: true, - ProxyURL: config.App.ProxyURL, - Insecure: config.App.Insecure, + Insecure: config.Server.Insecure, + } + + proxyRuntime := core.ProxyRuntimeBrowser + if config.Server.IsRawRequests { + proxyRuntime = core.ProxyRuntimeRaw + } + + proxyCfg, err := buildNormalizedProxyConfig(proxyRuntime) + if err != nil { + logrus.Errorf("Error validating proxy config: %v", err) + return + } + + policy := resolveEngineProxyPolicy(proxyCfg, engineType) + + selectedProxy, err := selectCLIProxy(proxyCfg, policy) + if err != nil { + logrus.Errorf("Error selecting proxy for %s: %v", engineType, err) + return + } + + if config.Server.IsRawRequests { + query.ProxyURL = selectedProxy } logrus.Infof("Starting SERP search request using %s engine for query: %s", engineType, query.Text) var results []core.SearchResult - if config.App.IsRawRequests { + if config.Server.IsRawRequests { logrus.Infof("Using raw requests mode for %s search", engineType) results, err = searchRaw(engineType, query) } else { logrus.Infof("Using browser mode for %s search", engineType) - results, err = searchBrowser(engineType, query) + results, err = searchBrowser(engineType, query, selectedProxy) } if err != nil { @@ -61,43 +82,51 @@ func search(cmd *cobra.Command, args []string) { fmt.Println(string(b)) } -func searchBrowser(engineType string, query core.Query) ([]core.SearchResult, error) { + +func searchBrowser(engineType string, query core.Query, browserProxyURL string) ([]core.SearchResult, error) { var engine core.SearchEngine + if core.IsAuthenticatedSocksProxyURL(browserProxyURL) { + return nil, fmt.Errorf( + "%w: browser runtime does not support authenticated SOCKS proxy %s", + core.ErrProxyUnavailable, + core.MaskProxyURL(browserProxyURL), + ) + } opts := core.BrowserOpts{ - IsHeadless: !config.App.IsBrowserHead, // Disable headless if browser head mode is set + IsHeadless: !config.App.IsBrowserHead, IsLeakless: config.App.IsLeakless, Timeout: time.Second * time.Duration(config.App.Timeout), LeavePageOpen: config.App.IsLeaveHead, CaptchaSolverApiKey: config.Config2Capcha.ApiKey, BrowserPath: config.App.BrowserPath, - ProxyURL: config.App.ProxyURL, - Insecure: config.App.Insecure, + ProxyURL: browserProxyURL, + Insecure: config.Server.Insecure, UseStealth: config.App.IsStealth, } - if config.App.IsDebug { + if config.Server.IsDebug { opts.IsHeadless = false } browser, err := core.NewBrowser(opts) if err != nil { - logrus.Error(err) + return nil, err } switch strings.ToLower(engineType) { case "yandex": - engine = yandex.New(*browser, config.YandexConfig) + engine = yandex.New(*browser, config.YandexConfig.SearchEngineOptions) case "google": - engine = google.New(*browser, config.GoogleConfig) + engine = google.New(*browser, config.GoogleConfig.SearchEngineOptions) case "baidu": - engine = baidu.New(*browser, config.BaiduConfig) + engine = baidu.New(*browser, config.BaiduConfig.SearchEngineOptions) case "bing": - engine = bing.New(*browser, config.BingConfig) - case "duck": - engine = duckduckgo.New(*browser, config.DuckDuckGoConfig) + engine = bing.New(*browser, config.BingConfig.SearchEngineOptions) + case "duckduckgo": + engine = duckduckgo.New(*browser, config.DuckDuckGoConfig.SearchEngineOptions) default: - logrus.Infof("No `%s` search engine found", engineType) + return nil, fmt.Errorf("no %q search engine found", engineType) } return engine.Search(query) @@ -116,13 +145,42 @@ func searchRaw(engineType string, query core.Query) ([]core.SearchResult, error) case "bing": logrus.Warn("Bing does not support raw HTTP requests mode. Please use browser mode instead.") return nil, fmt.Errorf("bing does not support raw requests mode") - case "duck": + case "duckduckgo": logrus.Warn("DuckDuckGo does not support raw HTTP requests mode. Please use browser mode instead.") return nil, fmt.Errorf("duckduckgo does not support raw requests mode") default: - logrus.Infof("No `%s` search engine found", engineType) + return nil, fmt.Errorf("no %q search engine found", engineType) + } +} + +func selectCLIProxy(proxyCfg core.ProxyConfig, policy core.ProxyPolicy) (string, error) { + if policy.Mode == core.ProxyModeOff { + return "", nil + } + + if global := strings.TrimSpace(proxyCfg.Proxies.Global); global != "" { + return global, nil + } + + if proxyCfg.Registry == nil { + return "", fmt.Errorf("%w: no proxy registry configured", core.ErrProxyUnavailable) + } + + selected := proxyCfg.Registry.NextByTag(policy.Tag) + if selected == "" { + return "", fmt.Errorf("%w: no healthy proxy available for tag %q", core.ErrProxyUnavailable, policy.Tag) + } + + return selected, nil +} + +func normalizeEngineArg(raw string) string { + switch strings.ToLower(strings.TrimSpace(raw)) { + case "duck": + return "duckduckgo" + default: + return strings.ToLower(strings.TrimSpace(raw)) } - return nil, nil } func init() { diff --git a/cmd/serve.go b/cmd/serve.go index 54e0079..4ee3a2d 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -2,6 +2,8 @@ package cmd import ( "fmt" + "strings" + "sync" "time" "github.com/karust/openserp/baidu" @@ -21,7 +23,7 @@ type rawEngine struct { } func (r *rawEngine) Search(q core.Query) ([]core.SearchResult, error) { - q.Insecure = config.App.Insecure + q.Insecure = config.Server.Insecure switch r.name { case "google": @@ -67,17 +69,60 @@ func serve(cmd *cobra.Command, args []string) { corsCfg.AllowHeaders = config.CORS.AllowHeaders corsCfg.MaxAge = config.CORS.MaxAge - proxyCfg := core.ProxyConfig{ - Runtime: core.ProxyRuntimeBrowser, - StaticURL: config.App.ProxyURL, - PoolURLs: config.ProxyPool.URLs, - PoolFailureThreshold: config.ProxyPool.FailureThreshold, - } - if config.App.IsRawRequests { - proxyCfg.Runtime = core.ProxyRuntimeRaw + proxyRuntime := core.ProxyRuntimeBrowser + if config.Server.IsRawRequests { + proxyRuntime = core.ProxyRuntimeRaw } - serverOpts := core.ServerOptions{ + proxyCfg, err := buildNormalizedProxyConfig(proxyRuntime) + if err != nil { + logrus.Errorf("invalid proxy configuration: %v", err) + return + } + + if config.Server.IsRawRequests { + logrus.Warn("Browserless results are very inconsistent or may not even work!") + serverOpts := buildServerOptions(corsCfg, proxyCfg) + serv := core.NewServerWithOptions(config.Server.Host, config.Server.Port, serverOpts, + &rawEngine{name: "google"}, + &rawEngine{name: "yandex"}, + &rawEngine{name: "baidu"}, + ) + if err := serv.Listen(); err != nil { + logrus.Error(err) + } + return + } + + baseOpts := core.BrowserOpts{ + IsHeadless: !config.App.IsBrowserHead, + IsLeakless: config.App.IsLeakless, + Timeout: time.Second * time.Duration(config.App.Timeout), + LeavePageOpen: config.App.IsLeaveHead, + CaptchaSolverApiKey: config.Config2Capcha.ApiKey, + BrowserPath: config.App.BrowserPath, + Insecure: config.Server.Insecure, + UseStealth: config.App.IsStealth, + } + if config.Server.IsDebug { + baseOpts.IsHeadless = false + } + + engines, err := buildBrowserEngines(baseOpts, proxyCfg) + if err != nil { + logrus.Error(err) + return + } + + serverOpts := buildServerOptions(corsCfg, proxyCfg) + serv := core.NewServerWithOptions(config.Server.Host, config.Server.Port, serverOpts, engines...) + if err := serv.Listen(); err != nil { + logrus.Error(err) + } +} + +func buildServerOptions(corsCfg core.CORSConfig, proxyCfg core.ProxyConfig) core.ServerOptions { + return core.ServerOptions{ CacheTTL: time.Duration(config.Cache.TTLSeconds) * time.Second, CacheMaxSize: config.Cache.MaxSize, EnableCORS: config.CORS.Enabled, @@ -98,52 +143,227 @@ func serve(cmd *cobra.Command, args []string) { Proxy: proxyCfg, }, } +} - if config.App.IsRawRequests { - logrus.Warn("Browserless results are very inconsistent or may not even work!") - serv := core.NewServerWithOptions(config.App.Host, config.App.Port, serverOpts, - &rawEngine{name: "google"}, - &rawEngine{name: "yandex"}, - &rawEngine{name: "baidu"}, +type browserPool struct { + mu sync.Mutex + base core.BrowserOpts + browser map[string]*core.Browser +} + +func newBrowserPool(base core.BrowserOpts) *browserPool { + return &browserPool{ + base: base, + browser: map[string]*core.Browser{}, + } +} + +func (p *browserPool) get(proxyURL string) (*core.Browser, error) { + key := strings.TrimSpace(proxyURL) + if key == "" { + key = "direct" + } + + p.mu.Lock() + defer p.mu.Unlock() + + if b, ok := p.browser[key]; ok { + return b, nil + } + + opts := p.base + opts.ProxyURL = proxyURL + b, err := core.NewBrowser(opts) + if err != nil { + return nil, err + } + + // Reuse one launched browser per unique effective proxy so startup stays lazy + // and engines with identical proxy policy don't spawn duplicate browser processes. + p.browser[key] = b + return b, nil +} + +type pooledBrowserEngine struct { + name string + limiter *rate.Limiter + opts core.SearchEngineOptions + factory func(core.Browser, core.SearchEngineOptions) core.SearchEngine + pool *browserPool + + mu sync.Mutex + engines map[string]core.SearchEngine +} + +func (e *pooledBrowserEngine) Search(q core.Query) ([]core.SearchResult, error) { + engine, err := e.getOrCreate(q.ProxyURL) + if err != nil { + return nil, err + } + return engine.Search(q) +} + +func (e *pooledBrowserEngine) SearchImage(q core.Query) ([]core.SearchResult, error) { + engine, err := e.getOrCreate(q.ProxyURL) + if err != nil { + return nil, err + } + return engine.SearchImage(q) +} + +func (e *pooledBrowserEngine) IsInitialized() bool { + return true +} + +func (e *pooledBrowserEngine) Name() string { + return e.name +} + +func (e *pooledBrowserEngine) GetRateLimiter() *rate.Limiter { + return e.limiter +} + +func (e *pooledBrowserEngine) getOrCreate(proxyURL string) (core.SearchEngine, error) { + key := strings.TrimSpace(proxyURL) + if key == "" { + key = "direct" + } + + e.mu.Lock() + defer e.mu.Unlock() + + if engine, ok := e.engines[key]; ok { + return engine, nil + } + + browser, err := e.pool.get(proxyURL) + if err != nil { + return nil, err + } + + engine := e.factory(*browser, e.opts) + e.engines[key] = engine + return engine, nil +} + +type browserEngineSpec struct { + name string + opts core.SearchEngineOptions + factory func(core.Browser, core.SearchEngineOptions) core.SearchEngine +} + +func browserEngineSpecs() []browserEngineSpec { + return []browserEngineSpec{ + { + name: "google", + opts: config.GoogleConfig.SearchEngineOptions, + factory: func(browser core.Browser, opts core.SearchEngineOptions) core.SearchEngine { + return google.New(browser, opts) + }, + }, + { + name: "yandex", + opts: config.YandexConfig.SearchEngineOptions, + factory: func(browser core.Browser, opts core.SearchEngineOptions) core.SearchEngine { + return yandex.New(browser, opts) + }, + }, + { + name: "baidu", + opts: config.BaiduConfig.SearchEngineOptions, + factory: func(browser core.Browser, opts core.SearchEngineOptions) core.SearchEngine { + return baidu.New(browser, opts) + }, + }, + { + name: "bing", + opts: config.BingConfig.SearchEngineOptions, + factory: func(browser core.Browser, opts core.SearchEngineOptions) core.SearchEngine { + return bing.New(browser, opts) + }, + }, + { + name: "duckduckgo", + opts: config.DuckDuckGoConfig.SearchEngineOptions, + factory: func(browser core.Browser, opts core.SearchEngineOptions) core.SearchEngine { + return duckduckgo.New(browser, opts) + }, + }, + } +} + +func buildBrowserEngines(baseOpts core.BrowserOpts, proxyCfg core.ProxyConfig) ([]core.SearchEngine, error) { + pool := newBrowserPool(baseOpts) + specs := browserEngineSpecs() + + engines := make([]core.SearchEngine, 0, len(specs)) + for _, spec := range specs { + policy := resolveEngineProxyPolicy(proxyCfg, spec.name) + if err := validateBrowserProxyPolicy(proxyCfg, policy); err != nil { + return nil, fmt.Errorf("browser proxy validation failed for engine %s: %w", spec.name, err) + } + + opts := spec.opts + opts.Init() + engines = append(engines, &pooledBrowserEngine{ + name: spec.name, + limiter: rate.NewLimiter(rate.Every(opts.GetRatelimit()), opts.RateBurst), + opts: opts, + factory: spec.factory, + pool: pool, + engines: map[string]core.SearchEngine{}, + }) + } + + return engines, nil +} + +func validateBrowserProxyPolicy(proxyCfg core.ProxyConfig, policy core.ProxyPolicy) error { + if policy.Mode != core.ProxyModeTagPool { + return nil + } + + proxyURL := strings.TrimSpace(proxyCfg.Proxies.Global) + if proxyURL != "" { + return validateBrowserProxyURL(proxyURL) + } + + for _, entry := range proxyCfg.Proxies.Entries { + if !entryHasTag(entry, policy.Tag) { + continue + } + if err := validateBrowserProxyURL(entry.URL); err != nil { + return err + } + } + + return nil +} + +func validateBrowserProxyURL(proxyURL string) error { + // Browser startup must stop immediately on authenticated SOCKS because Chrome + // cannot use that proxy shape reliably and retrying a different proxy hides the misconfiguration. + if core.IsAuthenticatedSocksProxyURL(proxyURL) { + return fmt.Errorf( + "%w: browser runtime does not support authenticated SOCKS proxy %s", + core.ErrProxyUnavailable, + core.MaskProxyURL(proxyURL), ) - serv.Listen() - return } + return nil +} - opts := core.BrowserOpts{ - IsHeadless: !config.App.IsBrowserHead, // Disable headless if browser head mode is set - IsLeakless: config.App.IsLeakless, - Timeout: time.Second * time.Duration(config.App.Timeout), - LeavePageOpen: config.App.IsLeaveHead, - CaptchaSolverApiKey: config.Config2Capcha.ApiKey, - BrowserPath: config.App.BrowserPath, - ProxyURL: config.App.ProxyURL, - Insecure: config.App.Insecure, - UseStealth: config.App.IsStealth, +func entryHasTag(entry core.ProxyEntryConfig, tag string) bool { + tag = strings.TrimSpace(strings.ToLower(tag)) + if tag == "" { + return false } - - if config.App.IsDebug { - opts.IsHeadless = false - } - - browser, err := core.NewBrowser(opts) - if err != nil { - logrus.Error(err) - return - } - - yand := yandex.New(*browser, config.YandexConfig) - gogl := google.New(*browser, config.GoogleConfig) - baidu := baidu.New(*browser, config.BaiduConfig) - bing := bing.New(*browser, config.BingConfig) - ddg := duckduckgo.New(*browser, config.DuckDuckGoConfig) - - serv := core.NewServerWithOptions(config.App.Host, config.App.Port, serverOpts, gogl, yand, baidu, bing, ddg) - - err = serv.Listen() - if err != nil { - logrus.Error(err) + for _, entryTag := range entry.Tags { + if strings.TrimSpace(strings.ToLower(entryTag)) == tag { + return true + } } + return false } func init() { diff --git a/cmd/serve_test.go b/cmd/serve_test.go new file mode 100644 index 0000000..0f6a639 --- /dev/null +++ b/cmd/serve_test.go @@ -0,0 +1,105 @@ +package cmd + +import ( + "strings" + "testing" + + "github.com/karust/openserp/core" +) + +func TestValidateBrowserProxyPolicyRejectsAuthenticatedSocks(t *testing.T) { + tests := []struct { + name string + proxyCfg core.ProxyConfig + policy core.ProxyPolicy + }{ + { + name: "global authenticated socks", + proxyCfg: core.ProxyConfig{ + Proxies: core.ProxiesConfig{ + Global: "socks5h://user:pass@127.0.0.1:1080", + }, + }, + policy: core.ProxyPolicy{Mode: core.ProxyModeTagPool}, + }, + { + name: "tag pool authenticated socks", + proxyCfg: core.ProxyConfig{ + Proxies: core.ProxiesConfig{ + Entries: []core.ProxyEntryConfig{ + {URL: "socks5://user:pass@127.0.0.1:1080", Tags: []string{"us"}}, + }, + }, + }, + policy: core.ProxyPolicy{Mode: core.ProxyModeTagPool, Tag: "us"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validateBrowserProxyPolicy(tt.proxyCfg, tt.policy) + if err == nil { + t.Fatal("expected browser proxy validation to fail") + } + if !strings.Contains(err.Error(), "authenticated SOCKS proxy") { + t.Fatalf("expected explicit authenticated SOCKS error, got %v", err) + } + }) + } +} + +func TestValidateBrowserProxyPolicyAllowsHTTPAuthAndPlainSocks(t *testing.T) { + tests := []struct { + name string + proxyCfg core.ProxyConfig + policy core.ProxyPolicy + }{ + { + name: "global http auth", + proxyCfg: core.ProxyConfig{ + Proxies: core.ProxiesConfig{ + Global: "http://user:pass@127.0.0.1:8080", + }, + }, + policy: core.ProxyPolicy{Mode: core.ProxyModeTagPool}, + }, + { + name: "tag pool plain socks", + proxyCfg: core.ProxyConfig{ + Proxies: core.ProxiesConfig{ + Entries: []core.ProxyEntryConfig{ + {URL: "socks5://127.0.0.1:1080", Tags: []string{"eu"}}, + }, + }, + }, + policy: core.ProxyPolicy{Mode: core.ProxyModeTagPool, Tag: "eu"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := validateBrowserProxyPolicy(tt.proxyCfg, tt.policy); err != nil { + t.Fatalf("expected browser proxy validation to succeed, got %v", err) + } + }) + } +} + +func TestValidateBrowserProxyPolicyRejectsTaggedAuthenticatedSocksInPool(t *testing.T) { + proxyCfg := core.ProxyConfig{ + Proxies: core.ProxiesConfig{ + Entries: []core.ProxyEntryConfig{ + {URL: "http://127.0.0.1:8080", Tags: []string{"default"}}, + {URL: "socks5://user:pass@127.0.0.1:1080", Tags: []string{"default"}}, + }, + }, + } + + err := validateBrowserProxyPolicy(proxyCfg, core.ProxyPolicy{Mode: core.ProxyModeTagPool, Tag: "default"}) + if err == nil { + t.Fatal("expected browser proxy validation to fail for tag pool") + } + if !strings.Contains(err.Error(), "authenticated SOCKS proxy") { + t.Fatalf("expected explicit authenticated SOCKS error, got %v", err) + } +} diff --git a/config.yaml b/config.yaml index 9efc33e..ae9f7d3 100644 --- a/config.yaml +++ b/config.yaml @@ -1,67 +1,78 @@ -app: +server: host: 0.0.0.0 # API host to bind port: 7000 # API port to bind debug: false # Enable debug logs and force browser UI mode verbose: true # Enable info-level request logs + raw_requests: false # true = raw HTTP mode, false = browser mode + insecure: true # Allow insecure TLS connections + +app: timeout: 15 # Browser/search timeout in seconds + browser_path: "" # Custom browser binary path (chrome/chromium/edge..) head: false # Show browser UI (headful mode) leakless: false # Force browser process cleanup after request leave_head: false # Keep tabs open after request for debugging stealth: false # Enable stealth browser plugin - insecure: true # Allow insecure TLS connections - # HTTP/SOCKS5 proxy URL - #proxy: "http://127.0.0.1:1" - # Custom browser binary path (chrome/chromium/edge..) - #browser_path: "C:/Program Files/BraveSoftware/Brave-Browser/Application/brave.exe" - -proxy_pool: - # List of proxy URLs for raw-mode resilient rotation. - # Browser mode still uses only app.proxy in stage 4. - #urls: - # - "socks5h://127.0.0.1:1080" - failure_threshold: 3 # Disable a proxy after this many consecutive request failures +#proxies: +# # Force a single proxy for all engines. +# # Same behavior as passing --proxy on the CLI. +# global: http://test:test@127.0.0.1:18888 +# +# # Advanced mode: define tagged proxy pools and opt engines in with `proxy: `. +# entries: +# - url: http://test:test@127.0.0.1:18888 +# tags: [default, us] +# - url: socks5://127.0.0.1:19080 # Chrome doesn't support authenticated SOCKS proxies and hostname resolution +# tags: [default, eu] +# health: +# failure_threshold: 3 # Disable proxy after this many consecutive failures cache: ttl_seconds: 60 # Dedicated endpoint cache TTL in seconds (0 disables cache) max_size: 1000 # Maximum cached dedicated responses before oldest-entry eviction resilience: - max_retries: 3 # Retry attempts per engine request (0 disables retries) + max_retries: 2 # Retry attempts per engine request (0 disables retries) allow_endpoint_fallback: false # Keep dedicated endpoints engine-pure by default -circuit_breaker: - failures: 5 # Consecutive failures required to open circuit - recovery_seconds: 60 # Wait time before moving open circuit to half-open - successes: 2 # Consecutive half-open successes required to close circuit +# circuit_breaker: +# failures: 5 # Consecutive failures required to open circuit +# recovery_seconds: 60 # Wait time before moving open circuit to half-open +# successes: 2 # Consecutive half-open successes required to close circuit cors: enabled: true allow_origins: "*" allow_methods: "GET, POST, OPTIONS" allow_headers: "Origin, Content-Type, Accept, Authorization" - max_age: 86400 # Browser preflight cache in seconds + max_age: 86400 -2captcha: - apikey: "123123123123123" +# 2captcha: +# apikey: "123123123123123" google: rate_requests: 4 # Allowed average requests per minute rate_burst: 2 # Burst requests before limiter applies captcha: true # Enable captcha solver path + #proxy: us yandex: rate_requests: 4 rate_burst: 2 + #proxy: eu baidu: rate_requests: 4 rate_burst: 2 + # No proxy tag means direct traffic bing: rate_requests: 4 rate_burst: 2 + # No proxy tag means direct traffic duckduckgo: rate_requests: 4 rate_burst: 2 + #proxy: default diff --git a/core/browser.go b/core/browser.go index 9f05182..70fb5e3 100644 --- a/core/browser.go +++ b/core/browser.go @@ -1,6 +1,8 @@ package core import ( + "context" + "errors" "fmt" "net/url" "os" @@ -79,17 +81,16 @@ func NewBrowser(opts BrowserOpts) (*Browser, error) { return nil, fmt.Errorf("invalid proxy URL: %v", err) } - // Make sure the proxy URL includes the scheme when passed to launcher - // This ensures proper handling of SOCKS5 proxies - proxyStr := proxyUrl.String() - logrus.Debugf("Setting up proxy: %s", proxyStr) + // Chrome's proxy-server flag must not contain credentials. + // Auth (if needed) is handled separately via DevTools auth callbacks. + proxyStr := proxyURLForBrowserLaunch(proxyUrl) + logrus.Debugf("Setting up proxy: %s", MaskProxyURL(proxyStr)) l = l.Proxy(proxyStr) // Check if proxy has auth credentials if proxyUrl.User != nil { username := proxyUrl.User.Username() - logrus.Debugf("Using proxy authentication: %s:****", username) - // We'll handle auth in the Navigate method + logrus.Debugf("Proxy credentials configured for %s proxy: %s:****", proxyUrl.Scheme, username) } } @@ -104,6 +105,23 @@ func NewBrowser(opts BrowserOpts) (*Browser, error) { return &b, err } +func proxyURLForBrowserLaunch(u *url.URL) string { + if u == nil { + return "" + } + clone := *u + // Chrome expects socks5 scheme in --proxy-server; socks5h is not accepted. + if clone.Scheme == "socks5h" { + clone.Scheme = "socks5" + } + clone.User = nil + clone.Path = "" + clone.RawPath = "" + clone.RawQuery = "" + clone.Fragment = "" + return clone.String() +} + func validateBrowserBinaryPath(path string) error { info, err := os.Stat(path) if err != nil { @@ -146,9 +164,14 @@ func (b *Browser) IsInitialized() bool { func (b *Browser) Navigate(URL string) (*rod.Page, error) { logrus.Debug("Navigate to: ", URL) - b.browser = rod.New().ControlURL(b.browserAddr) - b.browser.MustConnect() - b.browser.SetCookies(nil) + browser := rod.New().ControlURL(b.browserAddr).Timeout(b.Timeout) + if err := browser.Connect(); err != nil { + return nil, fmt.Errorf("browser connect failed: %w", err) + } + b.browser = browser + if err := b.browser.SetCookies(nil); err != nil { + return nil, fmt.Errorf("browser cookie reset failed: %w", err) + } // Handle proxy authentication before any navigations if b.ProxyURL != "" { @@ -156,39 +179,66 @@ func (b *Browser) Navigate(URL string) (*rod.Page, error) { // Always ignore certificate errors when using proxies // This fixes the ERR_CERT_AUTHORITY_INVALID error for SOCKS5 proxies - b.browser.MustIgnoreCertErrors(true) + if err := b.browser.IgnoreCertErrors(true); err != nil { + return nil, fmt.Errorf("configure proxy cert handling failed: %w", err) + } - if proxyUrl.User != nil { + if proxyUrl.User != nil && (proxyUrl.Scheme == "http" || proxyUrl.Scheme == "https") { username := proxyUrl.User.Username() password, _ := proxyUrl.User.Password() // Launch auth handler before any navigation occurs - go b.browser.MustHandleAuth(username, password)() + go func() { + if err := b.browser.HandleAuth(username, password)(); err != nil { + logrus.Debugf("Proxy auth handler stopped: %v", err) + } + }() + } else if proxyUrl.User != nil && (proxyUrl.Scheme == "socks5" || proxyUrl.Scheme == "socks5h") { + // This callback handles HTTP proxy auth challenges; it doesn't authenticate SOCKS proxies. + logrus.Debug("SOCKS proxy credentials are not handled by browser auth callback") } } else if b.Insecure { // Still respect the insecure flag if no proxy is used - b.browser.MustIgnoreCertErrors(true) + if err := b.browser.IgnoreCertErrors(true); err != nil { + return nil, fmt.Errorf("configure insecure mode failed: %w", err) + } } - ua := strings.ReplaceAll(b.browser.MustVersion().UserAgent, "HeadlessChrome/", "Chrome/") + version, err := b.browser.Version() + if err != nil { + return nil, fmt.Errorf("read browser version failed: %w", err) + } + ua := strings.ReplaceAll(version.UserAgent, "HeadlessChrome/", "Chrome/") var page *rod.Page if b.UseStealth { - page = stealth.MustPage(b.browser) - page.MustEmulate(devices.Device{ + page, err = stealth.Page(b.browser) + if err != nil { + return nil, fmt.Errorf("create stealth page failed: %w", err) + } + err = page.Emulate(devices.Device{ AcceptLanguage: b.LanguageCode, UserAgent: ua, }) + if err != nil { + return nil, fmt.Errorf("emulate stealth page failed: %w", err) + } } else { - page = b.browser.MustPage("about:blank") + page, err = b.browser.Page(proto.TargetCreateTarget{URL: "about:blank"}) + if err != nil { + return nil, fmt.Errorf("create page failed: %w", err) + } - page.MustEmulate(devices.Device{ + err = page.Emulate(devices.Device{ AcceptLanguage: b.LanguageCode, UserAgent: ua, }) + if err != nil { + return nil, fmt.Errorf("emulate page failed: %w", err) + } - proto.EmulationSetDeviceMetricsOverride{ + err = proto.EmulationSetDeviceMetricsOverride{ Width: 1920, Height: 1080, DeviceScaleFactor: 1, @@ -196,25 +246,37 @@ func (b *Browser) Navigate(URL string) (*rod.Page, error) { ScreenWidth: &[]int{1920}[0], ScreenHeight: &[]int{1080}[0], }.Call(page) + if err != nil { + return nil, fmt.Errorf("set device metrics failed: %w", err) + } } //EnableCustomStealth(page) - err := page.Navigate(URL) + timedPage := page.Timeout(b.Timeout) + + err = timedPage.Navigate(URL) if err != nil { return nil, err } // Avoid panics from MustWaitLoad when the target navigates/closes mid-wait - if werr := page.WaitLoad(); werr != nil { - logrus.Debugf("WaitLoad returned early: %v", werr) + if werr := timedPage.WaitLoad(); werr != nil { + if errors.Is(werr, context.DeadlineExceeded) { + // Some engines keep loading background resources while the DOM is already usable. + // Treat load timeout as non-fatal and let engine-specific selector timeouts decide. + logrus.Debugf("WaitLoad timed out after %s; continuing with partial page state", b.Timeout) + } else { + logrus.Debugf("WaitLoad returned early: %v", werr) + } } - wait := page.MustWaitRequestIdle() + // may cause bugs with google if b.WaitRequests { + wait := timedPage.WaitRequestIdle(300*time.Millisecond, nil, nil, nil) wait() } - time.Sleep(2 * time.Second) + time.Sleep(b.WaitLoadTime) return page, nil } diff --git a/core/proxy.go b/core/proxy.go index 01e3e20..987b8dd 100644 --- a/core/proxy.go +++ b/core/proxy.go @@ -1,8 +1,10 @@ package core import ( + "errors" "fmt" "net/url" + "sort" "strings" "sync" @@ -10,12 +12,11 @@ import ( ) const ( - ProxyRuntimeBrowser = "browser" - ProxyRuntimeRaw = "raw" - ProxyModeDisabled = "disabled" - ProxyModeStatic = "static" - ProxyModePool = "pool" - DefaultProxyPoolFailureThreshold = 3 + ProxyRuntimeBrowser = "browser" + ProxyRuntimeRaw = "raw" + ProxyModeOff = "off" + ProxyModeTagPool = "tag_pool" + DefaultProxyFailureThreshold = 3 ) var supportedProxySchemes = map[string]struct{}{ @@ -25,57 +26,176 @@ var supportedProxySchemes = map[string]struct{}{ "socks5h": {}, } -type ProxyConfig struct { - Runtime string - StaticURL string - PoolURLs []string - PoolFailureThreshold int +var ErrProxyUnavailable = errors.New("proxy unavailable") + +type ProxyPolicy struct { + Mode string `json:"mode" mapstructure:"mode"` + Tag string `json:"tag,omitempty" mapstructure:"tag"` } -type ProxyPool struct { +type ProxyEntryConfig struct { + URL string `json:"url" mapstructure:"url"` + Tags []string `json:"tags" mapstructure:"tags"` +} + +type ProxiesHealthConfig struct { + FailureThreshold int `json:"failure_threshold" mapstructure:"failure_threshold"` +} + +type ProxiesConfig struct { + Global string `json:"global,omitempty" mapstructure:"global"` + Entries []ProxyEntryConfig `json:"entries" mapstructure:"entries"` + Health ProxiesHealthConfig `json:"health" mapstructure:"health"` +} + +type ProxyConfig struct { + Runtime string // raw or browser runtime behavior + Proxies ProxiesConfig // canonical proxy inventory + EnginePolicies map[string]string // engine-specific proxy tags + Registry *ProxyRegistry // optional shared registry from caller +} + +type ProxyTagSummary struct { + Configured int `json:"configured"` + Healthy int `json:"healthy"` +} + +type ProxyStatsEntry struct { + Proxy string `json:"proxy"` + Tags []string `json:"tags"` + Healthy bool `json:"healthy"` + Failures int `json:"failures"` + Disabled bool `json:"disabled"` +} + +type ProxyEngineStats struct { + Tag string `json:"tag,omitempty"` + SelectedProxy string `json:"selected_proxy"` +} + +type ProxyStats struct { + ConfiguredCount int `json:"configured_count"` + HealthyCount int `json:"healthy_count"` + UnhealthyCount int `json:"unhealthy_count"` + Tags map[string]ProxyTagSummary `json:"tags"` + Entries []ProxyStatsEntry `json:"entries"` + Engines map[string]ProxyEngineStats `json:"engines,omitempty"` +} + +type proxyState struct { + url string + tags []string + failures int + disabled bool +} + +type ProxyRegistry struct { mu sync.Mutex - proxies []ProxyEntry - next int + states map[string]*proxyState + order []string + tagIndex map[string][]string + nextByTag map[string]int failureThreshold int } -type ProxyEntry struct { - URL string - FailCount int - IsDisabled bool -} - -type ProxyPoolStats struct { - FailureThreshold int - Total int - Active int - Disabled int +func DefaultProxiesConfig() ProxiesConfig { + return ProxiesConfig{ + Global: "", + Entries: []ProxyEntryConfig{}, + Health: ProxiesHealthConfig{FailureThreshold: DefaultProxyFailureThreshold}, + } } func DefaultProxyConfig() ProxyConfig { return ProxyConfig{ - Runtime: ProxyRuntimeBrowser, - PoolFailureThreshold: DefaultProxyPoolFailureThreshold, + Runtime: ProxyRuntimeBrowser, + Proxies: DefaultProxiesConfig(), + EnginePolicies: map[string]string{}, } } func NormalizeProxyConfig(cfg ProxyConfig) (ProxyConfig, error) { cfg.Runtime = normalizeProxyRuntime(cfg.Runtime) - if cfg.PoolFailureThreshold <= 0 { - cfg.PoolFailureThreshold = DefaultProxyPoolFailureThreshold + + var err error + cfg.Proxies, err = NormalizeProxiesConfig(cfg.Proxies) + if err != nil { + return cfg, err } - staticURL, err := NormalizeProxyURL(cfg.StaticURL) - if err != nil { - return cfg, fmt.Errorf("invalid static proxy: %w", err) + if cfg.EnginePolicies == nil { + cfg.EnginePolicies = map[string]string{} } - poolURLs, err := NormalizeProxyURLs(cfg.PoolURLs) - if err != nil { - return cfg, fmt.Errorf("invalid proxy pool: %w", err) + normalizedEnginePolicies := make(map[string]string, len(cfg.EnginePolicies)) + for rawEngine, rawTag := range cfg.EnginePolicies { + engine := normalizeEngineName(rawEngine) + if engine == "" { + continue + } + tag := normalizeTag(rawTag) + if tag == "" { + continue + } + normalizedEnginePolicies[engine] = tag + } + cfg.EnginePolicies = normalizedEnginePolicies + + if cfg.Registry == nil { + if len(cfg.Proxies.Entries) > 0 { + registry, err := NewProxyRegistry(cfg.Proxies.Entries, cfg.Proxies.Health.FailureThreshold) + if err != nil { + return cfg, err + } + cfg.Registry = registry + } } - cfg.StaticURL = staticURL - cfg.PoolURLs = poolURLs + return cfg, nil +} + +func NormalizeProxiesConfig(cfg ProxiesConfig) (ProxiesConfig, error) { + global, err := NormalizeProxyURL(cfg.Global) + if err != nil { + return cfg, fmt.Errorf("invalid proxies.global: %w", err) + } + cfg.Global = global + + failureThreshold := cfg.Health.FailureThreshold + if failureThreshold <= 0 { + failureThreshold = DefaultProxyFailureThreshold + } + + normalizedEntries := make([]ProxyEntryConfig, 0, len(cfg.Entries)) + entryByURL := make(map[string]int, len(cfg.Entries)) + + for i, rawEntry := range cfg.Entries { + proxyURL, err := NormalizeProxyURL(rawEntry.URL) + if err != nil { + return cfg, fmt.Errorf("invalid proxies.entries[%d].url: %w", i, err) + } + if proxyURL == "" { + return cfg, fmt.Errorf("invalid proxies.entries[%d].url: value is required", i) + } + + tags, err := normalizeProxyTags(rawEntry.Tags) + if err != nil { + return cfg, fmt.Errorf("invalid proxies.entries[%d].tags: %w", i, err) + } + + if idx, ok := entryByURL[proxyURL]; ok { + normalizedEntries[idx].Tags = mergeTags(normalizedEntries[idx].Tags, tags) + continue + } + + normalizedEntries = append(normalizedEntries, ProxyEntryConfig{ + URL: proxyURL, + Tags: tags, + }) + entryByURL[proxyURL] = len(normalizedEntries) - 1 + } + + cfg.Entries = normalizedEntries + cfg.Health = ProxiesHealthConfig{FailureThreshold: failureThreshold} return cfg, nil } @@ -140,126 +260,227 @@ func MaskProxyURL(raw string) string { return fmt.Sprintf("%s://%s", parsed.Scheme, parsed.Host) } -func NewProxyPool(proxyURLs []string, failureThreshold int) (*ProxyPool, error) { - normalizedURLs, err := NormalizeProxyURLs(proxyURLs) +func ResolveEffectiveProxyPolicy(globalProxyURL string, engineTag string) ProxyPolicy { + if strings.TrimSpace(globalProxyURL) != "" { + return ProxyPolicy{Mode: ProxyModeTagPool} + } + + tag := normalizeTag(engineTag) + if tag == "" { + return ProxyPolicy{Mode: ProxyModeOff} + } + + return ProxyPolicy{Mode: ProxyModeTagPool, Tag: tag} +} + +func NormalizeProxyTag(raw string) (string, error) { + tag := normalizeTag(raw) + if tag == "" { + return "", fmt.Errorf("value is required") + } + return tag, nil +} + +func IsAuthenticatedSocksProxyURL(raw string) bool { + normalized, err := NormalizeProxyURL(raw) + if err != nil || normalized == "" { + return false + } + + parsed, err := url.Parse(normalized) if err != nil { - return nil, err + return false } + + if (parsed.Scheme == "socks5" || parsed.Scheme == "socks5h") && parsed.User != nil { + return true + } + + return false +} + +func NewProxyRegistry(entries []ProxyEntryConfig, failureThreshold int) (*ProxyRegistry, error) { if failureThreshold <= 0 { - failureThreshold = DefaultProxyPoolFailureThreshold + failureThreshold = DefaultProxyFailureThreshold } - entries := make([]ProxyEntry, 0, len(normalizedURLs)) - for _, proxyURL := range normalizedURLs { - entries = append(entries, ProxyEntry{URL: proxyURL}) + states := make(map[string]*proxyState, len(entries)) + order := make([]string, 0, len(entries)) + tagIndex := make(map[string][]string) + + for idx, entry := range entries { + proxyURL, err := NormalizeProxyURL(entry.URL) + if err != nil { + return nil, fmt.Errorf("invalid proxy registry entry[%d] url: %w", idx, err) + } + if proxyURL == "" { + return nil, fmt.Errorf("invalid proxy registry entry[%d] url: value is required", idx) + } + + tags, err := normalizeProxyTags(entry.Tags) + if err != nil { + return nil, fmt.Errorf("invalid proxy registry entry[%d] tags: %w", idx, err) + } + + states[proxyURL] = &proxyState{url: proxyURL, tags: tags} + order = append(order, proxyURL) + for _, tag := range tags { + tagIndex[tag] = append(tagIndex[tag], proxyURL) + } } - return &ProxyPool{ - proxies: entries, + return &ProxyRegistry{ + states: states, + order: order, + tagIndex: tagIndex, + nextByTag: make(map[string]int, len(tagIndex)), failureThreshold: failureThreshold, }, nil } -func (p *ProxyPool) Next() string { - p.mu.Lock() - defer p.mu.Unlock() - - if len(p.proxies) == 0 { +func (r *ProxyRegistry) NextByTag(tag string) string { + tag = normalizeTag(tag) + if tag == "" { return "" } - if p.allDisabledLocked() { - logrus.Warn("Proxy pool exhausted, re-enabling all configured proxies") - for i := range p.proxies { - p.proxies[i].IsDisabled = false - p.proxies[i].FailCount = 0 + r.mu.Lock() + defer r.mu.Unlock() + + urls := r.tagIndex[tag] + if len(urls) == 0 { + return "" + } + + if r.allDisabledLocked(urls) { + logrus.Warnf("Proxy tag pool exhausted for %q, re-enabling tagged proxies", tag) + for _, proxyURL := range urls { + state := r.states[proxyURL] + state.disabled = false + state.failures = 0 } } - for i := 0; i < len(p.proxies); i++ { - idx := (p.next + i) % len(p.proxies) - if p.proxies[idx].IsDisabled { + start := r.nextByTag[tag] + for i := 0; i < len(urls); i++ { + idx := (start + i) % len(urls) + proxyURL := urls[idx] + state := r.states[proxyURL] + if state.disabled { continue } - p.next = (idx + 1) % len(p.proxies) - selected := p.proxies[idx].URL - logrus.Debugf("Selected proxy from pool: %s", MaskProxyURL(selected)) - return selected + r.nextByTag[tag] = (idx + 1) % len(urls) + logrus.Debugf("Selected proxy for tag=%s: %s", tag, MaskProxyURL(proxyURL)) + return proxyURL } return "" } -func (p *ProxyPool) ReportFailure(proxyURL string) { - p.mu.Lock() - defer p.mu.Unlock() - - for i := range p.proxies { - if p.proxies[i].URL != proxyURL { - continue - } - - p.proxies[i].FailCount++ - if p.proxies[i].FailCount >= p.failureThreshold { - p.proxies[i].IsDisabled = true - logrus.Warnf( - "Disabled proxy after %d failures: %s", - p.proxies[i].FailCount, - MaskProxyURL(proxyURL), - ) - } +func (r *ProxyRegistry) ReportFailure(proxyURL string) { + proxyURL, err := NormalizeProxyURL(proxyURL) + if err != nil || proxyURL == "" { return } -} -func (p *ProxyPool) ReportSuccess(proxyURL string) { - p.mu.Lock() - defer p.mu.Unlock() + r.mu.Lock() + defer r.mu.Unlock() - for i := range p.proxies { - if p.proxies[i].URL != proxyURL { - continue - } - - p.proxies[i].FailCount = 0 - p.proxies[i].IsDisabled = false + state, ok := r.states[proxyURL] + if !ok { return } -} -func (p *ProxyPool) Size() int { - p.mu.Lock() - defer p.mu.Unlock() - - return len(p.proxies) -} - -func (p *ProxyPool) Stats() ProxyPoolStats { - p.mu.Lock() - defer p.mu.Unlock() - - stats := ProxyPoolStats{ - FailureThreshold: p.failureThreshold, - Total: len(p.proxies), + state.failures++ + if state.failures >= r.failureThreshold { + state.disabled = true + logrus.Warnf("Disabled proxy after %d failures: %s", state.failures, MaskProxyURL(proxyURL)) } - for _, proxy := range p.proxies { - if proxy.IsDisabled { - stats.Disabled++ - continue +} + +func (r *ProxyRegistry) ReportSuccess(proxyURL string) { + proxyURL, err := NormalizeProxyURL(proxyURL) + if err != nil || proxyURL == "" { + return + } + + r.mu.Lock() + defer r.mu.Unlock() + + state, ok := r.states[proxyURL] + if !ok { + return + } + + state.failures = 0 + state.disabled = false +} + +func (r *ProxyRegistry) HasHealthyProxyForTag(tag string) bool { + tag = normalizeTag(tag) + if tag == "" { + return false + } + + r.mu.Lock() + defer r.mu.Unlock() + + for _, proxyURL := range r.tagIndex[tag] { + if state, ok := r.states[proxyURL]; ok && !state.disabled { + return true + } + } + + return false +} + +func (r *ProxyRegistry) BuildStats() ProxyStats { + r.mu.Lock() + defer r.mu.Unlock() + + stats := ProxyStats{ + Tags: map[string]ProxyTagSummary{}, + Entries: make([]ProxyStatsEntry, 0, len(r.order)), + } + + for _, proxyURL := range r.order { + state := r.states[proxyURL] + healthy := !state.disabled + if healthy { + stats.HealthyCount++ + } else { + stats.UnhealthyCount++ + } + stats.ConfiguredCount++ + + stats.Entries = append(stats.Entries, ProxyStatsEntry{ + Proxy: MaskProxyURL(state.url), + Tags: append([]string(nil), state.tags...), + Healthy: healthy, + Failures: state.failures, + Disabled: state.disabled, + }) + + for _, tag := range state.tags { + summary := stats.Tags[tag] + summary.Configured++ + if healthy { + summary.Healthy++ + } + stats.Tags[tag] = summary } - stats.Active++ } return stats } -func (p *ProxyPool) allDisabledLocked() bool { - if len(p.proxies) == 0 { +func (r *ProxyRegistry) allDisabledLocked(urls []string) bool { + if len(urls) == 0 { return false } - for _, proxy := range p.proxies { - if !proxy.IsDisabled { + for _, proxyURL := range urls { + if state, ok := r.states[proxyURL]; ok && !state.disabled { return false } } @@ -274,3 +495,55 @@ func normalizeProxyRuntime(runtime string) string { return ProxyRuntimeBrowser } } + +func normalizeProxyTags(tags []string) ([]string, error) { + if len(tags) == 0 { + return nil, fmt.Errorf("at least one tag is required") + } + + seen := make(map[string]struct{}, len(tags)) + normalized := make([]string, 0, len(tags)) + for _, rawTag := range tags { + tag := normalizeTag(rawTag) + if tag == "" { + continue + } + if _, ok := seen[tag]; ok { + continue + } + seen[tag] = struct{}{} + normalized = append(normalized, tag) + } + + if len(normalized) == 0 { + return nil, fmt.Errorf("at least one non-empty tag is required") + } + + sort.Strings(normalized) + return normalized, nil +} + +func normalizeTag(raw string) string { + return strings.ToLower(strings.TrimSpace(raw)) +} + +func mergeTags(base []string, additional []string) []string { + combined := make(map[string]struct{}, len(base)+len(additional)) + for _, tag := range base { + combined[tag] = struct{}{} + } + for _, tag := range additional { + combined[tag] = struct{}{} + } + + merged := make([]string, 0, len(combined)) + for tag := range combined { + merged = append(merged, tag) + } + sort.Strings(merged) + return merged +} + +func normalizeEngineName(raw string) string { + return strings.ToLower(strings.TrimSpace(raw)) +} diff --git a/core/proxy_integration_test.go b/core/proxy_integration_test.go index 620b299..bf21cfb 100644 --- a/core/proxy_integration_test.go +++ b/core/proxy_integration_test.go @@ -14,6 +14,7 @@ import ( "strings" "testing" + "github.com/karust/openserp/testutil" "golang.org/x/time/rate" ) @@ -99,6 +100,7 @@ func (e *proxyIntegrationEngine) SearchImage(q Query) ([]SearchResult, error) { func proxyIntegrationConfig(t *testing.T) proxyIntegrationURLs { t.Helper() + testutil.RequireIntegration(t) if os.Getenv(proxyIntegrationEnabledEnv) != "1" { t.Skipf("set %s=1 to run proxy integration tests", proxyIntegrationEnabledEnv) @@ -144,9 +146,17 @@ func assertProxyPoolRotation(t *testing.T, targetURL string, pool []string) { opts.Resilience.Retry.MaxBackoff = 0 opts.Resilience.Retry.BackoffFactor = 1 opts.Resilience.Proxy = ProxyConfig{ - Runtime: ProxyRuntimeRaw, - PoolURLs: pool, - PoolFailureThreshold: 1, + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + Health: ProxiesHealthConfig{FailureThreshold: 1}, + }, + EnginePolicies: map[string]string{"google": "default"}, + } + for _, proxyURL := range pool { + opts.Resilience.Proxy.Proxies.Entries = append(opts.Resilience.Proxy.Proxies.Entries, ProxyEntryConfig{ + URL: proxyURL, + Tags: []string{"default"}, + }) } srv := NewServerWithOptions("127.0.0.1", 7190, opts, engine) @@ -162,19 +172,17 @@ func assertProxyPoolRotation(t *testing.T, targetURL string, pool []string) { t.Fatalf("unexpected proxy rotation order: %#v", engine.proxies) } - statsResp := request(t, srv, "/resilience/stats") + statsResp := request(t, srv, "/stats/proxy") var stats map[string]interface{} if err := json.NewDecoder(statsResp.Body).Decode(&stats); err != nil { t.Fatalf("decode stats: %v", err) } - proxyStats := stats["proxy"].(map[string]interface{}) - poolStats := proxyStats["pool"].(map[string]interface{}) - if got := poolStats["active"].(float64); got != 1 { - t.Fatalf("expected 1 active proxy, got %v", got) + if got := stats["healthy_count"].(float64); got != 1 { + t.Fatalf("expected healthy_count=1, got %v", got) } - if got := poolStats["disabled"].(float64); got != 1 { - t.Fatalf("expected 1 disabled proxy, got %v", got) + if got := stats["unhealthy_count"].(float64); got != 1 { + t.Fatalf("expected unhealthy_count=1, got %v", got) } } diff --git a/core/proxy_test.go b/core/proxy_test.go index 6fd5bf8..63b69b6 100644 --- a/core/proxy_test.go +++ b/core/proxy_test.go @@ -7,6 +7,7 @@ import ( "net" "net/http" "net/http/httptest" + "net/url" "testing" "time" @@ -50,68 +51,116 @@ func TestNormalizeProxyURL(t *testing.T) { } } -func TestNormalizeProxyConfigDefaultsAndDeduplicates(t *testing.T) { - cfg, err := NormalizeProxyConfig(ProxyConfig{ - Runtime: "RAW", - StaticURL: " socks5h://127.0.0.1:1080 ", - PoolURLs: []string{ - "", - "http://proxy-one:8080", - "http://proxy-one:8080", - "socks5://proxy-two:1080", +func TestNormalizeProxiesConfigDefaultsAndDeduplicates(t *testing.T) { + cfg, err := NormalizeProxiesConfig(ProxiesConfig{ + Global: " HTTP://proxy-global:8080 ", + Entries: []ProxyEntryConfig{ + {URL: " http://proxy-one:8080 ", Tags: []string{"default", "us"}}, + {URL: "http://proxy-one:8080", Tags: []string{"de", "us"}}, + {URL: "socks5://proxy-two:1080", Tags: []string{"default"}}, }, }) if err != nil { - t.Fatalf("normalize config: %v", err) + t.Fatalf("normalize proxies config: %v", err) } - if cfg.Runtime != ProxyRuntimeRaw { - t.Fatalf("expected raw runtime, got %s", cfg.Runtime) + if cfg.Global != "http://proxy-global:8080" { + t.Fatalf("expected normalized global proxy, got %q", cfg.Global) } - if cfg.StaticURL != "socks5h://127.0.0.1:1080" { - t.Fatalf("unexpected static proxy: %s", cfg.StaticURL) + if cfg.Health.FailureThreshold != DefaultProxyFailureThreshold { + t.Fatalf("expected default failure threshold %d, got %d", DefaultProxyFailureThreshold, cfg.Health.FailureThreshold) } - if cfg.PoolFailureThreshold != DefaultProxyPoolFailureThreshold { - t.Fatalf("expected default threshold %d, got %d", DefaultProxyPoolFailureThreshold, cfg.PoolFailureThreshold) + if len(cfg.Entries) != 2 { + t.Fatalf("expected 2 deduplicated entries, got %d", len(cfg.Entries)) } - if len(cfg.PoolURLs) != 2 { - t.Fatalf("expected 2 deduplicated pool URLs, got %d", len(cfg.PoolURLs)) + if cfg.Entries[0].URL != "http://proxy-one:8080" { + t.Fatalf("unexpected normalized URL for first entry: %s", cfg.Entries[0].URL) + } + if len(cfg.Entries[0].Tags) != 3 { + t.Fatalf("expected merged tags in first entry, got %#v", cfg.Entries[0].Tags) } } -func TestProxyPoolRoundRobinAndFailureRecovery(t *testing.T) { - pool, err := NewProxyPool([]string{"http://proxy1:8080", "http://proxy2:8080"}, 2) - if err != nil { - t.Fatalf("new proxy pool: %v", err) +func TestNormalizeProxiesConfigRejectsInvalidEntries(t *testing.T) { + _, err := NormalizeProxiesConfig(ProxiesConfig{ + Entries: []ProxyEntryConfig{{URL: "ftp://proxy:21", Tags: []string{"default"}}}, + }) + if err == nil { + t.Fatal("expected invalid scheme error") } - if got := pool.Next(); got != "http://proxy1:8080" { + _, err = NormalizeProxiesConfig(ProxiesConfig{ + Entries: []ProxyEntryConfig{{URL: "http://proxy:8080", Tags: []string{" "}}}, + }) + if err == nil { + t.Fatal("expected empty tags error") + } + + _, err = NormalizeProxiesConfig(ProxiesConfig{ + Global: "ftp://proxy:21", + Entries: []ProxyEntryConfig{{URL: "http://proxy:8080", Tags: []string{"default"}}}, + }) + if err == nil { + t.Fatal("expected invalid global proxy scheme error") + } +} + +func TestResolveEffectiveProxyPolicy(t *testing.T) { + offPolicy := ResolveEffectiveProxyPolicy("", "") + if offPolicy.Mode != ProxyModeOff { + t.Fatalf("expected mode off, got %s", offPolicy.Mode) + } + if offPolicy.Tag != "" { + t.Fatalf("expected empty tag for off mode, got %q", offPolicy.Tag) + } + + tagOnlyPolicy := ResolveEffectiveProxyPolicy("", "US") + if tagOnlyPolicy.Mode != ProxyModeTagPool || tagOnlyPolicy.Tag != "us" { + t.Fatalf("unexpected effective policy with tag override: %#v", tagOnlyPolicy) + } + + globalPolicy := ResolveEffectiveProxyPolicy("http://proxy-global:8080", "eu") + if globalPolicy.Mode != ProxyModeTagPool || globalPolicy.Tag != "" { + t.Fatalf("expected global proxy to ignore engine tags, got %#v", globalPolicy) + } +} + +func TestProxyRegistryRoundRobinAndFailureRecovery(t *testing.T) { + registry, err := NewProxyRegistry([]ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + {URL: "http://proxy2:8080", Tags: []string{"default"}}, + }, 2) + if err != nil { + t.Fatalf("new proxy registry: %v", err) + } + + if got := registry.NextByTag("default"); got != "http://proxy1:8080" { t.Fatalf("expected first proxy1, got %s", got) } - if got := pool.Next(); got != "http://proxy2:8080" { + if got := registry.NextByTag("default"); got != "http://proxy2:8080" { t.Fatalf("expected second proxy2, got %s", got) } - pool.ReportFailure("http://proxy1:8080") - pool.ReportFailure("http://proxy1:8080") - if got := pool.Next(); got != "http://proxy2:8080" { + registry.ReportFailure("http://proxy1:8080") + registry.ReportFailure("http://proxy1:8080") + if got := registry.NextByTag("default"); got != "http://proxy2:8080" { t.Fatalf("expected proxy2 while proxy1 disabled, got %s", got) } - pool.ReportFailure("http://proxy2:8080") - pool.ReportFailure("http://proxy2:8080") - if got := pool.Next(); got != "http://proxy1:8080" { - t.Fatalf("expected pool reset to proxy1 after exhaustion, got %s", got) + registry.ReportFailure("http://proxy2:8080") + registry.ReportFailure("http://proxy2:8080") + if got := registry.NextByTag("default"); got != "http://proxy1:8080" { + t.Fatalf("expected tag pool reset to proxy1 after exhaustion, got %s", got) } - pool.ReportFailure("http://proxy1:8080") - pool.ReportSuccess("http://proxy1:8080") - stats := pool.Stats() - if stats.Disabled != 0 { - t.Fatalf("expected no disabled proxies after recovery, got %d", stats.Disabled) + registry.ReportFailure("http://proxy1:8080") + registry.ReportSuccess("http://proxy1:8080") + stats := registry.BuildStats() + if stats.UnhealthyCount != 0 { + t.Fatalf("expected no unhealthy proxies after success recovery, got %d", stats.UnhealthyCount) } - if stats.Active != 2 { - t.Fatalf("expected both proxies active, got %d", stats.Active) + if stats.HealthyCount != 2 { + t.Fatalf("expected two healthy proxies, got %d", stats.HealthyCount) } } @@ -121,6 +170,75 @@ func TestMaskProxyURLRedactsCredentials(t *testing.T) { } } +func TestProxyURLForBrowserLaunchStripsCredentials(t *testing.T) { + u, err := url.Parse("http://user:pass@127.0.0.1:18888") + if err != nil { + t.Fatalf("parse proxy URL: %v", err) + } + + got := proxyURLForBrowserLaunch(u) + want := "http://127.0.0.1:18888" + if got != want { + t.Fatalf("expected %q, got %q", want, got) + } +} + +func TestProxyURLForBrowserLaunchNormalizesSocks5h(t *testing.T) { + u, err := url.Parse("socks5h://test:test@127.0.0.1:19080") + if err != nil { + t.Fatalf("parse proxy URL: %v", err) + } + + got := proxyURLForBrowserLaunch(u) + want := "socks5://127.0.0.1:19080" + if got != want { + t.Fatalf("expected %q, got %q", want, got) + } +} + +func TestProxyStatsMaskCredentials(t *testing.T) { + registry, err := NewProxyRegistry([]ProxyEntryConfig{ + {URL: "http://user:pass@proxy.example:8080", Tags: []string{"default"}}, + }, 1) + if err != nil { + t.Fatalf("new proxy registry: %v", err) + } + + stats := registry.BuildStats() + if len(stats.Entries) != 1 { + t.Fatalf("expected one proxy stats entry, got %d", len(stats.Entries)) + } + if got := stats.Entries[0].Proxy; got != "http://proxy.example:8080" { + t.Fatalf("expected masked proxy in stats, got %q", got) + } +} + +func TestNormalizeProxyTag(t *testing.T) { + tag, err := NormalizeProxyTag(" US ") + if err != nil { + t.Fatalf("normalize proxy tag: %v", err) + } + if tag != "us" { + t.Fatalf("expected normalized tag us, got %q", tag) + } + + if _, err := NormalizeProxyTag(" "); err == nil { + t.Fatal("expected empty proxy tag validation error") + } +} + +func TestIsAuthenticatedSocksProxyURL(t *testing.T) { + if !IsAuthenticatedSocksProxyURL("socks5h://user:pass@127.0.0.1:1080") { + t.Fatal("expected authenticated socks proxy to be detected") + } + if IsAuthenticatedSocksProxyURL("socks5://127.0.0.1:1080") { + t.Fatal("expected plain socks proxy to remain browser-compatible") + } + if IsAuthenticatedSocksProxyURL("http://user:pass@127.0.0.1:8080") { + t.Fatal("expected HTTP auth proxy to remain browser-compatible") + } +} + func TestNewRawHTTPClientSocks5hUsesProxyDNS(t *testing.T) { target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) diff --git a/core/resilient.go b/core/resilient.go index a8e356b..c8bebb2 100644 --- a/core/resilient.go +++ b/core/resilient.go @@ -2,7 +2,9 @@ package core import ( "context" + "errors" "fmt" + "strings" "sync" "github.com/sirupsen/logrus" @@ -13,8 +15,19 @@ type ResilientSearcher struct { engines []SearchEngine cbManager *CircuitBreakerManager retryCfg RetryConfig - proxyCfg ProxyConfig - proxyPool *ProxyPool + + proxyRuntime string + proxyCfg ProxyConfig + proxyRegistry *ProxyRegistry + proxyDefaults ProxyPolicy + + effectivePolicies map[string]ProxyPolicy +} + +type ProxyExecutionMeta struct { + Mode string `json:"mode"` + Tag string `json:"tag,omitempty"` + Used string `json:"used"` } type ResilientConfig struct { @@ -34,103 +47,101 @@ func DefaultResilientConfig() ResilientConfig { func NewResilientSearcher(engines []SearchEngine, cfg ResilientConfig) *ResilientSearcher { proxyCfg, err := NormalizeProxyConfig(cfg.Proxy) if err != nil { - logrus.Errorf("Invalid proxy config, disabling proxy support: %v", err) + logrus.Errorf("Invalid proxy config, using defaults: %v", err) proxyCfg = DefaultProxyConfig() + proxyCfg, _ = NormalizeProxyConfig(proxyCfg) } rs := &ResilientSearcher{ - engines: engines, - cbManager: NewCircuitBreakerManager(cfg.CircuitBreaker), - retryCfg: cfg.Retry, - proxyCfg: proxyCfg, + engines: engines, + cbManager: NewCircuitBreakerManager(cfg.CircuitBreaker), + retryCfg: cfg.Retry, + proxyRuntime: proxyCfg.Runtime, + proxyCfg: proxyCfg, + proxyRegistry: proxyCfg.Registry, + proxyDefaults: ResolveEffectiveProxyPolicy(proxyCfg.Proxies.Global, ""), + effectivePolicies: make(map[string]ProxyPolicy, len(engines)), } - if len(proxyCfg.PoolURLs) > 0 { - pool, err := NewProxyPool(proxyCfg.PoolURLs, proxyCfg.PoolFailureThreshold) - if err != nil { - logrus.Errorf("Invalid proxy pool config, disabling proxy rotation: %v", err) - } else { - rs.proxyPool = pool - if proxyCfg.Runtime == ProxyRuntimeBrowser { - logrus.Warn("Proxy pool configured in browser runtime; pool remains observability-only until browser proxy rotation is implemented") - } else { - logrus.Infof("Proxy rotation enabled with %d proxies", pool.Size()) - } - } + for _, engine := range engines { + engineName := normalizeEngineName(engine.Name()) + override := proxyCfg.EnginePolicies[engineName] + effective := ResolveEffectiveProxyPolicy(proxyCfg.Proxies.Global, override) + rs.effectivePolicies[engineName] = effective } return rs } // SearchPrimary keeps dedicated endpoints engine-pure (no fallback). -func (rs *ResilientSearcher) SearchPrimary(primaryEngine SearchEngine, q Query) ([]SearchResult, string, error) { - results, err := rs.searchWithProtection(primaryEngine, q) +func (rs *ResilientSearcher) SearchPrimary(primaryEngine SearchEngine, q Query) ([]SearchResult, string, ProxyExecutionMeta, error) { + results, proxyMeta, err := rs.searchWithProtection(primaryEngine, q, false) if err != nil { - return nil, primaryEngine.Name(), err + return nil, primaryEngine.Name(), proxyMeta, err } - return results, primaryEngine.Name(), nil + return results, primaryEngine.Name(), proxyMeta, nil } // SearchWithFallback retries primary and then tries other initialized engines. -func (rs *ResilientSearcher) SearchWithFallback(primaryEngine SearchEngine, q Query) ([]SearchResult, string, error) { - results, err := rs.searchWithProtection(primaryEngine, q) - if err == nil { - return results, primaryEngine.Name(), nil - } - - logrus.Warnf("[Resilient] Primary engine %s failed: %s. Trying fallback engines...", primaryEngine.Name(), err) - for _, fallbackEngine := range rs.engines { - if fallbackEngine.Name() == primaryEngine.Name() || !fallbackEngine.IsInitialized() { - continue - } - - results, err := rs.searchWithProtection(fallbackEngine, q) - if err == nil { - logrus.Infof("[Resilient] Fallback to %s succeeded with %d results", fallbackEngine.Name(), len(results)) - return results, fallbackEngine.Name(), nil - } - logrus.Warnf("[Resilient] Fallback engine %s also failed: %s", fallbackEngine.Name(), err) - } - - return nil, primaryEngine.Name(), ErrAllEnginesFailed +func (rs *ResilientSearcher) SearchWithFallback(primaryEngine SearchEngine, q Query) ([]SearchResult, string, ProxyExecutionMeta, error) { + return rs.searchWithFallback(primaryEngine, q, false) } -func (rs *ResilientSearcher) SearchImagePrimary(primaryEngine SearchEngine, q Query) ([]SearchResult, string, error) { - results, err := rs.searchImageWithProtection(primaryEngine, q) +func (rs *ResilientSearcher) SearchImagePrimary(primaryEngine SearchEngine, q Query) ([]SearchResult, string, ProxyExecutionMeta, error) { + results, proxyMeta, err := rs.searchWithProtection(primaryEngine, q, true) if err != nil { - return nil, primaryEngine.Name(), err + return nil, primaryEngine.Name(), proxyMeta, err } - return results, primaryEngine.Name(), nil + return results, primaryEngine.Name(), proxyMeta, nil } -func (rs *ResilientSearcher) SearchImageWithFallback(primaryEngine SearchEngine, q Query) ([]SearchResult, string, error) { - results, err := rs.searchImageWithProtection(primaryEngine, q) +func (rs *ResilientSearcher) SearchImageWithFallback(primaryEngine SearchEngine, q Query) ([]SearchResult, string, ProxyExecutionMeta, error) { + return rs.searchWithFallback(primaryEngine, q, true) +} + +func (rs *ResilientSearcher) searchWithFallback(primaryEngine SearchEngine, q Query, isImage bool) ([]SearchResult, string, ProxyExecutionMeta, error) { + results, proxyMeta, err := rs.searchWithProtection(primaryEngine, q, isImage) if err == nil { - return results, primaryEngine.Name(), nil + return results, primaryEngine.Name(), proxyMeta, nil + } + if errors.Is(err, ErrProxyUnavailable) { + logrus.Warnf("[Resilient] Primary engine %s proxy policy failed closed: %s", primaryEngine.Name(), err) + return nil, primaryEngine.Name(), proxyMeta, err } - logrus.Warnf("[Resilient] Primary engine %s image search failed: %s. Trying fallback engines...", primaryEngine.Name(), err) + action := "failed" + successMessage := "Fallback to %s succeeded with %d results" + if isImage { + action = "image search failed" + successMessage = "Image fallback to %s succeeded with %d results" + } + + logrus.Warnf("[Resilient] Primary engine %s %s: %s. Trying fallback engines...", primaryEngine.Name(), action, err) for _, fallbackEngine := range rs.engines { if fallbackEngine.Name() == primaryEngine.Name() || !fallbackEngine.IsInitialized() { continue } - results, err := rs.searchImageWithProtection(fallbackEngine, q) - if err == nil { - logrus.Infof("[Resilient] Image fallback to %s succeeded with %d results", fallbackEngine.Name(), len(results)) - return results, fallbackEngine.Name(), nil + results, fallbackMeta, fallbackErr := rs.searchWithProtection(fallbackEngine, q, isImage) + if fallbackErr == nil { + logrus.Infof("[Resilient] "+successMessage, fallbackEngine.Name(), len(results)) + return results, fallbackEngine.Name(), fallbackMeta, nil } + logrus.Warnf("[Resilient] Fallback engine %s also failed: %s", fallbackEngine.Name(), fallbackErr) } - return nil, primaryEngine.Name(), ErrAllEnginesFailed + return nil, primaryEngine.Name(), proxyMeta, ErrAllEnginesFailed } -func (rs *ResilientSearcher) searchWithProtection(engine SearchEngine, q Query) ([]SearchResult, error) { +func (rs *ResilientSearcher) searchWithProtection(engine SearchEngine, q Query, isImage bool) ([]SearchResult, ProxyExecutionMeta, error) { cb := rs.cbManager.Get(engine.Name()) if !cb.AllowRequest() { - return nil, ErrCircuitOpen + return nil, ProxyExecutionMeta{}, ErrCircuitOpen } + policy := rs.effectivePolicyForEngine(engine.Name()) + attemptMeta := rs.baseProxyMeta(policy) + result := RetryableSearch(rs.retryCfg, engine.Name(), func() ([]SearchResult, error) { limiter := engine.GetRateLimiter() if limiter != nil { @@ -139,48 +150,51 @@ func (rs *ResilientSearcher) searchWithProtection(engine SearchEngine, q Query) } } - attemptQuery, attemptProxy, reportToPool := rs.prepareAttemptQuery(q) - results, err := engine.Search(attemptQuery) - rs.reportProxyAttempt(attemptProxy, reportToPool, err) - return results, err - }) + attemptQuery := q + proxyURL := "" + reportToRegistry := false + attemptMeta = rs.baseProxyMeta(policy) - if result.Err != nil { - cb.RecordFailure() - return nil, result.Err - } - - cb.RecordSuccess() - return result.Results, nil -} - -func (rs *ResilientSearcher) searchImageWithProtection(engine SearchEngine, q Query) ([]SearchResult, error) { - cb := rs.cbManager.Get(engine.Name()) - if !cb.AllowRequest() { - return nil, ErrCircuitOpen - } - - result := RetryableSearch(rs.retryCfg, engine.Name(), func() ([]SearchResult, error) { - limiter := engine.GetRateLimiter() - if limiter != nil { - if err := limiter.Wait(context.Background()); err != nil { - return nil, err + switch policy.Mode { + case ProxyModeOff: + attemptQuery.ProxyURL = "" + attemptMeta.Used = "direct" + case ProxyModeTagPool: + proxyURL = rs.selectProxyForPolicy(policy) + if proxyURL == "" { + return nil, fmt.Errorf("%w: no healthy proxy available for tag %q", ErrProxyUnavailable, policy.Tag) } + attemptQuery.ProxyURL = proxyURL + reportToRegistry = policy.Tag != "" + attemptMeta.Used = MaskProxyURL(proxyURL) + } + + var ( + results []SearchResult + err error + ) + if isImage { + results, err = engine.SearchImage(attemptQuery) + } else { + results, err = engine.Search(attemptQuery) + } + + if reportToRegistry { + rs.reportProxyAttempt(proxyURL, err) } - attemptQuery, attemptProxy, reportToPool := rs.prepareAttemptQuery(q) - results, err := engine.SearchImage(attemptQuery) - rs.reportProxyAttempt(attemptProxy, reportToPool, err) return results, err }) if result.Err != nil { - cb.RecordFailure() - return nil, result.Err + if !errors.Is(result.Err, ErrProxyUnavailable) { + cb.RecordFailure() + } + return nil, attemptMeta, result.Err } cb.RecordSuccess() - return result.Results, nil + return result.Results, attemptMeta, nil } // SearchAllParallel applies retry/circuit protections per engine for mega search. @@ -202,7 +216,7 @@ func (rs *ResilientSearcher) SearchAllParallel(q Query, engines []SearchEngine) go func(eng SearchEngine) { defer wg.Done() - results, err := rs.searchWithProtection(eng, q) + results, _, err := rs.searchWithProtection(eng, q, false) if err != nil { return } @@ -240,7 +254,7 @@ func (rs *ResilientSearcher) SearchAllImageParallel(q Query, engines []SearchEng go func(eng SearchEngine) { defer wg.Done() - results, err := rs.searchImageWithProtection(eng, q) + results, _, err := rs.searchWithProtection(eng, q, true) if err != nil { return } @@ -264,88 +278,144 @@ func (rs *ResilientSearcher) GetCircuitBreakerStats() []map[string]interface{} { return rs.cbManager.AllStats() } -func (rs *ResilientSearcher) GetProxyPool() *ProxyPool { - return rs.proxyPool -} - -func (rs *ResilientSearcher) GetProxyStats() map[string]interface{} { - mode, source, rotationActive := rs.proxyMode() - - stats := map[string]interface{}{ - "mode": mode, - "runtime": rs.proxyCfg.Runtime, - "source": source, - "rotation_active": rotationActive, +func (rs *ResilientSearcher) GetProxyStats() ProxyStats { + stats := ProxyStats{ + ConfiguredCount: 0, + HealthyCount: 0, + UnhealthyCount: 0, + Tags: map[string]ProxyTagSummary{}, + Entries: []ProxyStatsEntry{}, } - if rs.proxyPool != nil { - poolStats := rs.proxyPool.Stats() - stats["pool"] = map[string]interface{}{ - "failure_threshold": poolStats.FailureThreshold, - "total": poolStats.Total, - "active": poolStats.Active, - "disabled": poolStats.Disabled, + if rs.proxyRegistry != nil { + stats = rs.proxyRegistry.BuildStats() + } + + engines := map[string]ProxyEngineStats{} + for _, engine := range rs.engines { + engineName := normalizeEngineName(engine.Name()) + policy := rs.effectivePolicyForEngine(engineName) + engineStats := ProxyEngineStats{} + + switch policy.Mode { + case ProxyModeOff: + engineStats.SelectedProxy = "direct" + case ProxyModeTagPool: + engineStats.Tag = policy.Tag + if global := strings.TrimSpace(rs.proxyCfg.Proxies.Global); global != "" { + engineStats.SelectedProxy = MaskProxyURL(global) + } else { + engineStats.SelectedProxy = "pooled" + } } + + engines[engineName] = engineStats + } + if len(engines) > 0 { + stats.Engines = engines } return stats } -func (rs *ResilientSearcher) prepareAttemptQuery(q Query) (Query, string, bool) { - attemptQuery := q - if rs.proxyCfg.Runtime != ProxyRuntimeRaw { - return attemptQuery, "", false +func (rs *ResilientSearcher) ResolveMegaProxyMeta(engines []SearchEngine) ProxyExecutionMeta { + if len(engines) == 0 { + return ProxyExecutionMeta{Mode: ProxyModeOff, Used: "direct"} } - if rs.proxyPool != nil { - proxyURL := rs.proxyPool.Next() - if proxyURL != "" { - attemptQuery.ProxyURL = proxyURL - return attemptQuery, proxyURL, true + allOff := true + proxiedTags := map[string]struct{}{} + hasOff := false + + for _, engine := range engines { + policy := rs.effectivePolicyForEngine(engine.Name()) + if policy.Mode == ProxyModeOff { + hasOff = true + continue + } + + allOff = false + if policy.Tag != "" { + proxiedTags[policy.Tag] = struct{}{} } } - attemptQuery.ProxyURL = rs.proxyCfg.StaticURL - return attemptQuery, "", false + if allOff { + return ProxyExecutionMeta{Mode: ProxyModeOff, Used: "direct"} + } + + meta := ProxyExecutionMeta{Mode: ProxyModeTagPool} + if len(proxiedTags) == 1 { + for tag := range proxiedTags { + meta.Tag = tag + } + } + + if global := strings.TrimSpace(rs.proxyCfg.Proxies.Global); global != "" && !hasOff { + meta.Used = MaskProxyURL(global) + return meta + } + + if rs.proxyRuntime == ProxyRuntimeRaw { + meta.Used = "multiple" + if hasOff { + meta.Used = "mixed" + } + return meta + } + meta.Used = "multiple" + if hasOff { + meta.Used = "mixed" + } + return meta } -func (rs *ResilientSearcher) reportProxyAttempt(proxyURL string, reportToPool bool, err error) { - if !reportToPool || rs.proxyPool == nil || proxyURL == "" { +func (rs *ResilientSearcher) baseProxyMeta(policy ProxyPolicy) ProxyExecutionMeta { + meta := ProxyExecutionMeta{Mode: policy.Mode} + if policy.Mode == ProxyModeTagPool { + meta.Tag = policy.Tag + return meta + } + meta.Used = "direct" + return meta +} + +func (rs *ResilientSearcher) effectivePolicyForEngine(engineName string) ProxyPolicy { + engineName = normalizeEngineName(engineName) + if policy, ok := rs.effectivePolicies[engineName]; ok { + return policy + } + return rs.proxyDefaults +} + +func (rs *ResilientSearcher) selectProxyForTag(tag string) string { + if rs.proxyRegistry == nil { + return "" + } + return rs.proxyRegistry.NextByTag(tag) +} + +func (rs *ResilientSearcher) reportProxyAttempt(proxyURL string, err error) { + if rs.proxyRegistry == nil || proxyURL == "" { return } if err != nil { - rs.proxyPool.ReportFailure(proxyURL) + rs.proxyRegistry.ReportFailure(proxyURL) return } - rs.proxyPool.ReportSuccess(proxyURL) + rs.proxyRegistry.ReportSuccess(proxyURL) } -func (rs *ResilientSearcher) proxyMode() (string, string, bool) { - hasStatic := rs.proxyCfg.StaticURL != "" - hasPool := rs.proxyPool != nil - - switch rs.proxyCfg.Runtime { - case ProxyRuntimeRaw: - switch { - case hasPool: - return ProxyModePool, "proxy_pool.urls", true - case hasStatic: - return ProxyModeStatic, "app.proxy", false - default: - return ProxyModeDisabled, "none", false - } - default: - switch { - case hasStatic: - return ProxyModeStatic, "app.proxy", false - case hasPool: - return ProxyModePool, "proxy_pool.urls", false - default: - return ProxyModeDisabled, "none", false - } +func (rs *ResilientSearcher) selectProxyForPolicy(policy ProxyPolicy) string { + if policy.Mode != ProxyModeTagPool { + return "" } + if global := strings.TrimSpace(rs.proxyCfg.Proxies.Global); global != "" { + return global + } + return rs.selectProxyForTag(policy.Tag) } var ErrAllEnginesFailed = fmt.Errorf("all search engines failed") diff --git a/core/retry.go b/core/retry.go index 76109f6..bbf48d9 100644 --- a/core/retry.go +++ b/core/retry.go @@ -1,6 +1,7 @@ package core import ( + "errors" "fmt" "math" "time" @@ -33,7 +34,7 @@ type RetryResult struct { } // RetryableSearch executes searchFn with exponential backoff retries. -// CAPTCHA errors are not retried. +// CAPTCHA and proxy-unavailable errors are not retried. func RetryableSearch(cfg RetryConfig, engineName string, searchFn func() ([]SearchResult, error)) RetryResult { if cfg.BackoffFactor <= 0 { cfg.BackoffFactor = 2.0 @@ -60,7 +61,7 @@ func RetryableSearch(cfg RetryConfig, engineName string, searchFn func() ([]Sear } lastErr = err - if err == ErrCaptcha { + if errors.Is(err, ErrCaptcha) { logrus.Warnf("[%s] CAPTCHA detected, skipping retries", engineName) return RetryResult{ Err: err, @@ -68,6 +69,14 @@ func RetryableSearch(cfg RetryConfig, engineName string, searchFn func() ([]Sear Engine: engineName, } } + if errors.Is(err, ErrProxyUnavailable) { + logrus.Warnf("[%s] Proxy unavailable, skipping retries", engineName) + return RetryResult{ + Err: err, + Attempts: attempt + 1, + Engine: engineName, + } + } logrus.Warnf("[%s] Attempt %d failed: %s", engineName, attempt+1, err) } diff --git a/core/server.go b/core/server.go index eefce58..888573c 100644 --- a/core/server.go +++ b/core/server.go @@ -2,6 +2,7 @@ package core import ( "encoding/json" + "errors" "fmt" "runtime" "sort" @@ -84,8 +85,10 @@ func NewServerWithOptions(host string, port int, opts ServerOptions, searchEngin app.Use(RequestLoggerMiddleware()) app.Get("/health", serv.handleHealthCheck) - app.Get("/cache/stats", serv.handleCacheStats) - app.Get("/resilience/stats", serv.handleResilienceStats) + app.Get("/stats", serv.handleStats) + app.Get("/stats/cache", serv.handleCacheStats) + app.Get("/stats/proxy", serv.handleProxyStats) + app.Get("/stats/cb", serv.handleCircuitBreakerStats) for _, engine := range searchEngines { locEngine := engine @@ -137,22 +140,24 @@ func (s *Server) handleDedicatedEndpoint(c *fiber.Ctx, engine SearchEngine, isIm var ( res []SearchResult usedEngine string + proxyMeta ProxyExecutionMeta searchErr error ) if isImage { if s.opts.AllowEndpointFallback { - res, usedEngine, searchErr = s.resilient.SearchImageWithFallback(engine, q) + res, usedEngine, proxyMeta, searchErr = s.resilient.SearchImageWithFallback(engine, q) } else { - res, usedEngine, searchErr = s.resilient.SearchImagePrimary(engine, q) + res, usedEngine, proxyMeta, searchErr = s.resilient.SearchImagePrimary(engine, q) } } else { if s.opts.AllowEndpointFallback { - res, usedEngine, searchErr = s.resilient.SearchWithFallback(engine, q) + res, usedEngine, proxyMeta, searchErr = s.resilient.SearchWithFallback(engine, q) } else { - res, usedEngine, searchErr = s.resilient.SearchPrimary(engine, q) + res, usedEngine, proxyMeta, searchErr = s.resilient.SearchPrimary(engine, q) } } + s.applyProxyHeaders(c, proxyMeta) if searchErr != nil { errToReturn := searchErr @@ -161,6 +166,10 @@ func (s *Server) handleDedicatedEndpoint(c *fiber.Ctx, engine SearchEngine, isIm errToReturn = fmt.Errorf("captcha found, please stop sending requests for a while: %w", searchErr) case ErrSearchTimeout: errToReturn = fmt.Errorf("%s", searchErr) + default: + if errors.Is(searchErr, ErrProxyUnavailable) { + errToReturn = fmt.Errorf("%s", searchErr) + } } logrus.Errorf("Error during resilient %s %s: %s", engine.Name(), action, searchErr) return fiber.NewError(fiber.StatusServiceUnavailable, errToReturn.Error()) @@ -273,18 +282,26 @@ func (s *Server) handleHealthCheck(c *fiber.Ctx) error { return c.JSON(health) } -func (s *Server) handleResilienceStats(c *fiber.Ctx) error { +func (s *Server) handleStats(c *fiber.Ctx) error { return c.JSON(map[string]interface{}{ - "circuit_breakers": s.resilient.GetCircuitBreakerStats(), + "cache": s.cacheStatsPayload(), "proxy": s.resilient.GetProxyStats(), + "circuit_breakers": s.resilient.GetCircuitBreakerStats(), }) } func (s *Server) handleCacheStats(c *fiber.Ctx) error { - if s.cache == nil { - return c.JSON(map[string]interface{}{"status": false}) - } - return c.JSON(s.cache.Stats()) + return c.JSON(s.cacheStatsPayload()) +} + +func (s *Server) handleProxyStats(c *fiber.Ctx) error { + return c.JSON(s.resilient.GetProxyStats()) +} + +func (s *Server) handleCircuitBreakerStats(c *fiber.Ctx) error { + return c.JSON(map[string]interface{}{ + "circuit_breakers": s.resilient.GetCircuitBreakerStats(), + }) } type MegaSearchResult struct { @@ -317,6 +334,7 @@ func (s *Server) handleMegaEndpoint(c *fiber.Ctx, action string, run func(Query, engineNames[i] = engine.Name() } engineNamesJoined := strings.Join(engineNames, ",") + s.applyProxyHeaders(c, s.resilient.ResolveMegaProxyMeta(enginesToUse)) logrus.Infof("Starting SERP mega %s request using engines: %s for query: %s", action, engineNamesJoined, q.Text) cacheHitCandidates := []cacheHitCandidate{ @@ -519,6 +537,32 @@ func (s *Server) megaCacheableEngines(engines []SearchEngine) []SearchEngine { return cacheable } +func (s *Server) cacheStatsPayload() interface{} { + if s.cache == nil { + return map[string]interface{}{"status": false} + } + return s.cache.Stats() +} + +func (s *Server) applyProxyHeaders(c *fiber.Ctx, meta ProxyExecutionMeta) { + mode := meta.Mode + if mode == "" { + mode = ProxyModeOff + } + + tag := meta.Tag + used := meta.Used + + if mode == ProxyModeOff { + tag = "" + used = "direct" + } + + c.Set("X-Proxy-Mode", mode) + c.Set("X-Proxy-Tag", tag) + c.Set("X-Proxy-Used", used) +} + func (s *Server) Listen() error { return s.app.Listen(s.addr) } diff --git a/core/server_test.go b/core/server_test.go index 213b789..df16edb 100644 --- a/core/server_test.go +++ b/core/server_test.go @@ -216,6 +216,34 @@ func TestDedicatedEndpointFallbackBypassesCache(t *testing.T) { } } +func TestDedicatedEndpointFallbackDoesNotBypassProxyPolicy(t *testing.T) { + primary := &engineMock{name: "google", initialized: true} + fallback := &engineMock{name: "yandex", initialized: true} + + opts := DefaultServerOptions() + opts.AllowEndpointFallback = true + opts.Resilience.Retry.MaxRetries = 0 + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + Entries: []ProxyEntryConfig{}, + }, + EnginePolicies: map[string]string{"google": "missing"}, + } + + srv := NewServerWithOptions("127.0.0.1", 7103, opts, primary, fallback) + resp := request(t, srv, "/google/search?text=golang") + if resp.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("expected fail-closed 503 when primary proxy policy cannot be satisfied, got %d", resp.StatusCode) + } + if got := resp.Header.Get("X-Fallback-Engine"); got != "" { + t.Fatalf("unexpected fallback header when proxy policy fails closed: %q", got) + } + if fallback.searchCalls != 0 { + t.Fatalf("fallback engine should not be called when proxy policy fails closed, got %d calls", fallback.searchCalls) + } +} + func TestMegaSearchCachesWholeQueryWithEngineOrderNormalization(t *testing.T) { google := &engineMock{ name: "google", @@ -374,7 +402,7 @@ func TestResilienceStatsContainsRetryInWhenCircuitOpen(t *testing.T) { srv := NewServerWithOptions("127.0.0.1", 7075, opts, primary) _ = request(t, srv, "/google/search?text=golang") - statsResp := request(t, srv, "/resilience/stats") + statsResp := request(t, srv, "/stats/cb") if statsResp.StatusCode != http.StatusOK { t.Fatalf("expected stats endpoint to return 200, got %d", statsResp.StatusCode) } @@ -402,105 +430,87 @@ func TestResilienceStatsContainsRetryInWhenCircuitOpen(t *testing.T) { } } -func TestResilienceStatsReportProxyModes(t *testing.T) { +func TestStatsEndpointsContract(t *testing.T) { engine := &engineMock{name: "google", initialized: true} + opts := DefaultServerOptions() + srv := NewServerWithOptions("127.0.0.1", 7090, opts, engine) - tests := []struct { - name string - proxyCfg ProxyConfig - wantMode string - wantRuntime string - wantSource string - wantRotation bool - wantPool bool - }{ - { - name: "disabled raw", - proxyCfg: ProxyConfig{Runtime: ProxyRuntimeRaw}, - wantMode: ProxyModeDisabled, - wantRuntime: ProxyRuntimeRaw, - wantSource: "none", - wantRotation: false, - wantPool: false, - }, - { - name: "static raw", - proxyCfg: ProxyConfig{ - Runtime: ProxyRuntimeRaw, - StaticURL: "socks5h://127.0.0.1:1080", - }, - wantMode: ProxyModeStatic, - wantRuntime: ProxyRuntimeRaw, - wantSource: "app.proxy", - wantRotation: false, - wantPool: false, - }, - { - name: "pool raw", - proxyCfg: ProxyConfig{ - Runtime: ProxyRuntimeRaw, - PoolURLs: []string{"http://proxy1:8080", "http://proxy2:8080"}, - PoolFailureThreshold: 2, - }, - wantMode: ProxyModePool, - wantRuntime: ProxyRuntimeRaw, - wantSource: "proxy_pool.urls", - wantRotation: true, - wantPool: true, - }, - { - name: "pool browser inactive", - proxyCfg: ProxyConfig{ - Runtime: ProxyRuntimeBrowser, - PoolURLs: []string{"http://proxy1:8080", "http://proxy2:8080"}, - PoolFailureThreshold: 2, - }, - wantMode: ProxyModePool, - wantRuntime: ProxyRuntimeBrowser, - wantSource: "proxy_pool.urls", - wantRotation: false, - wantPool: true, - }, + for _, path := range []string{"/stats", "/stats/cache", "/stats/proxy", "/stats/cb"} { + resp := request(t, srv, path) + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected %s to return 200, got %d", path, resp.StatusCode) + } } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - opts := DefaultServerOptions() - opts.Resilience.Proxy = tt.proxyCfg + for _, oldPath := range []string{"/cache/stats", "/resilience/stats"} { + resp := request(t, srv, oldPath) + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("expected %s to return 404, got %d", oldPath, resp.StatusCode) + } + } +} - srv := NewServerWithOptions("127.0.0.1", 7090, opts, engine) - resp := request(t, srv, "/resilience/stats") - if resp.StatusCode != http.StatusOK { - t.Fatalf("expected 200, got %d", resp.StatusCode) - } +func TestStatsProxyV2Payload(t *testing.T) { + engine := &engineMock{name: "google", initialized: true} + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + Entries: []ProxyEntryConfig{ + {URL: "http://user:pass@proxy1:8080", Tags: []string{"default", "us"}}, + {URL: "http://proxy2:8080", Tags: []string{"default"}}, + }, + Health: ProxiesHealthConfig{FailureThreshold: 1}, + }, + EnginePolicies: map[string]string{"google": "default"}, + } - var stats map[string]interface{} - if err := json.NewDecoder(resp.Body).Decode(&stats); err != nil { - t.Fatalf("decode stats response: %v", err) - } + srv := NewServerWithOptions("127.0.0.1", 7092, opts, engine) + resp := request(t, srv, "/stats/proxy") + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected /stats/proxy to return 200, got %d", resp.StatusCode) + } - proxy, ok := stats["proxy"].(map[string]interface{}) - if !ok { - t.Fatalf("expected proxy stats object, got %#v", stats["proxy"]) - } - if got := proxy["mode"]; got != tt.wantMode { - t.Fatalf("expected mode %q, got %#v", tt.wantMode, got) - } - if got := proxy["runtime"]; got != tt.wantRuntime { - t.Fatalf("expected runtime %q, got %#v", tt.wantRuntime, got) - } - if got := proxy["source"]; got != tt.wantSource { - t.Fatalf("expected source %q, got %#v", tt.wantSource, got) - } - if got := proxy["rotation_active"]; got != tt.wantRotation { - t.Fatalf("expected rotation_active=%v, got %#v", tt.wantRotation, got) - } + var payload map[string]interface{} + if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil { + t.Fatalf("decode /stats/proxy response: %v", err) + } - _, hasPool := proxy["pool"] - if hasPool != tt.wantPool { - t.Fatalf("expected pool presence=%v, got %v", tt.wantPool, hasPool) - } - }) + if got := payload["configured_count"].(float64); got != 2 { + t.Fatalf("expected configured_count=2, got %v", got) + } + if got := payload["healthy_count"].(float64); got != 2 { + t.Fatalf("expected healthy_count=2, got %v", got) + } + if got := payload["unhealthy_count"].(float64); got != 0 { + t.Fatalf("expected unhealthy_count=0, got %v", got) + } + + if _, exists := payload["defaults"]; exists { + t.Fatalf("defaults must not be exposed in proxy stats payload") + } + + entries := payload["entries"].([]interface{}) + first := entries[0].(map[string]interface{}) + if got := first["proxy"].(string); got == "http://user:pass@proxy1:8080" { + t.Fatalf("expected masked proxy, got %q", got) + } + if _, exists := payload["runtime"]; exists { + t.Fatalf("runtime must not be exposed in V2 stats payload") + } + if _, exists := payload["source"]; exists { + t.Fatalf("source must not be exposed in V2 stats payload") + } + engines := payload["engines"].(map[string]interface{}) + google := engines["google"].(map[string]interface{}) + if _, exists := google["mode"]; exists { + t.Fatalf("engine mode must not be exposed in proxy stats payload") + } + if got := google["tag"]; got != "default" { + t.Fatalf("expected engine tag default, got %#v", got) + } + if got := google["selected_proxy"]; got != "pooled" { + t.Fatalf("expected pooled engine proxy stats, got %#v", got) } } @@ -524,9 +534,15 @@ func TestResilientRawProxyPoolRotatesOnRetry(t *testing.T) { opts.Resilience.Retry.MaxBackoff = 0 opts.Resilience.Retry.BackoffFactor = 1 opts.Resilience.Proxy = ProxyConfig{ - Runtime: ProxyRuntimeRaw, - PoolURLs: []string{"http://bad-proxy:8080", "http://good-proxy:8080"}, - PoolFailureThreshold: 1, + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + Entries: []ProxyEntryConfig{ + {URL: "http://bad-proxy:8080", Tags: []string{"default"}}, + {URL: "http://good-proxy:8080", Tags: []string{"default"}}, + }, + Health: ProxiesHealthConfig{FailureThreshold: 1}, + }, + EnginePolicies: map[string]string{"google": "default"}, } srv := NewServerWithOptions("127.0.0.1", 7091, opts, engine) @@ -541,22 +557,217 @@ func TestResilientRawProxyPoolRotatesOnRetry(t *testing.T) { t.Fatalf("unexpected proxy rotation order: %#v", attemptedProxies) } - statsResp := request(t, srv, "/resilience/stats") + statsResp := request(t, srv, "/stats/proxy") var stats map[string]interface{} if err := json.NewDecoder(statsResp.Body).Decode(&stats); err != nil { t.Fatalf("decode stats response: %v", err) } - proxy := stats["proxy"].(map[string]interface{}) - if got := proxy["mode"]; got != ProxyModePool { - t.Fatalf("expected pool mode, got %#v", got) + if got := stats["healthy_count"].(float64); got != 1 { + t.Fatalf("expected healthy_count=1, got %v", got) + } + if got := stats["unhealthy_count"].(float64); got != 1 { + t.Fatalf("expected unhealthy_count=1, got %v", got) + } +} + +func TestProxyHeadersDirectAndTagPool(t *testing.T) { + directEngine := &engineMock{name: "google", initialized: true} + directSrv := NewServerWithOptions("127.0.0.1", 7093, DefaultServerOptions(), directEngine) + + directResp := request(t, directSrv, "/google/search?text=golang") + if directResp.StatusCode != http.StatusOK { + t.Fatalf("expected direct request to succeed, got %d", directResp.StatusCode) + } + if got := directResp.Header.Get("X-Proxy-Mode"); got != ProxyModeOff { + t.Fatalf("expected X-Proxy-Mode=%s, got %q", ProxyModeOff, got) + } + if got := directResp.Header.Get("X-Proxy-Tag"); got != "" { + t.Fatalf("expected empty X-Proxy-Tag in off mode, got %q", got) + } + if got := directResp.Header.Get("X-Proxy-Used"); got != "direct" { + t.Fatalf("expected X-Proxy-Used=direct, got %q", got) } - pool := proxy["pool"].(map[string]interface{}) - if got := pool["active"].(float64); got != 1 { - t.Fatalf("expected 1 active proxy, got %v", got) + proxiedEngine := &engineMock{name: "google", initialized: true} + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + Entries: []ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + }, + }, + EnginePolicies: map[string]string{"google": "default"}, } - if got := pool["disabled"].(float64); got != 1 { - t.Fatalf("expected 1 disabled proxy, got %v", got) + proxiedSrv := NewServerWithOptions("127.0.0.1", 7094, opts, proxiedEngine) + proxiedResp := request(t, proxiedSrv, "/google/search?text=golang") + if proxiedResp.StatusCode != http.StatusOK { + t.Fatalf("expected proxied request to succeed, got %d", proxiedResp.StatusCode) + } + if got := proxiedResp.Header.Get("X-Proxy-Mode"); got != ProxyModeTagPool { + t.Fatalf("expected X-Proxy-Mode=%s, got %q", ProxyModeTagPool, got) + } + if got := proxiedResp.Header.Get("X-Proxy-Tag"); got != "default" { + t.Fatalf("expected X-Proxy-Tag=default, got %q", got) + } + if got := proxiedResp.Header.Get("X-Proxy-Used"); got != "http://proxy1:8080" { + t.Fatalf("expected masked selected proxy, got %q", got) + } +} + +func TestGlobalProxyForcesAllEnginesRaw(t *testing.T) { + var googleProxy string + var yandexProxy string + + googleEngine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(q Query) ([]SearchResult, error) { + googleProxy = q.ProxyURL + return []SearchResult{{Rank: 1, URL: "https://example.com/google", Title: "google"}}, nil + }, + } + yandexEngine := &engineMock{ + name: "yandex", + initialized: true, + searchFn: func(q Query) ([]SearchResult, error) { + yandexProxy = q.ProxyURL + return []SearchResult{{Rank: 1, URL: "https://example.com/yandex", Title: "yandex"}}, nil + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + Global: "http://global-proxy:8080", + }, + EnginePolicies: map[string]string{ + "google": "us", + }, + } + + srv := NewServerWithOptions("127.0.0.1", 7097, opts, googleEngine, yandexEngine) + if resp := request(t, srv, "/google/search?text=golang"); resp.StatusCode != http.StatusOK { + t.Fatalf("expected google request to succeed, got %d", resp.StatusCode) + } + if resp := request(t, srv, "/yandex/search?text=golang"); resp.StatusCode != http.StatusOK { + t.Fatalf("expected yandex request to succeed, got %d", resp.StatusCode) + } + + if googleProxy != "http://global-proxy:8080" { + t.Fatalf("expected google to use global proxy, got %q", googleProxy) + } + if yandexProxy != "http://global-proxy:8080" { + t.Fatalf("expected yandex to use global proxy, got %q", yandexProxy) + } +} + +func TestBrowserProxyPoolRotatesPerRequest(t *testing.T) { + var attemptedProxies []string + engine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(q Query) ([]SearchResult, error) { + attemptedProxies = append(attemptedProxies, q.ProxyURL) + return []SearchResult{{Rank: 1, URL: "https://example.com/google", Title: "google"}}, nil + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeBrowser, + Proxies: ProxiesConfig{ + Entries: []ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + {URL: "http://proxy2:8080", Tags: []string{"default"}}, + }, + }, + EnginePolicies: map[string]string{"google": "default"}, + } + + srv := NewServerWithOptions("127.0.0.1", 7098, opts, engine) + if resp := request(t, srv, "/google/search?text=golang"); resp.StatusCode != http.StatusOK { + t.Fatalf("expected first browser-style request to succeed, got %d", resp.StatusCode) + } + if resp := request(t, srv, "/google/search?text=golang+2"); resp.StatusCode != http.StatusOK { + t.Fatalf("expected second browser-style request to succeed, got %d", resp.StatusCode) + } + + if len(attemptedProxies) != 2 { + t.Fatalf("expected 2 browser-style proxy attempts, got %d", len(attemptedProxies)) + } + if attemptedProxies[0] != "http://proxy1:8080" || attemptedProxies[1] != "http://proxy2:8080" { + t.Fatalf("expected browser proxy rotation order, got %#v", attemptedProxies) + } +} + +func TestProxyFailClosedWhenNoHealthyProxy(t *testing.T) { + engine := &engineMock{name: "google", initialized: true} + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + Entries: []ProxyEntryConfig{}, + }, + EnginePolicies: map[string]string{"google": "missing"}, + } + + srv := NewServerWithOptions("127.0.0.1", 7095, opts, engine) + resp := request(t, srv, "/google/search?text=golang") + if resp.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("expected fail-closed 503 when no healthy proxy exists, got %d", resp.StatusCode) + } + if got := resp.Header.Get("X-Proxy-Mode"); got != ProxyModeTagPool { + t.Fatalf("expected X-Proxy-Mode=%s on fail-closed response, got %q", ProxyModeTagPool, got) + } +} + +func TestEngineOverrideProxyBehaviorRaw(t *testing.T) { + var googleProxy string + var yandexProxy string + + googleEngine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(q Query) ([]SearchResult, error) { + googleProxy = q.ProxyURL + return []SearchResult{{Rank: 1, URL: "https://example.com/google", Title: "google"}}, nil + }, + } + yandexEngine := &engineMock{ + name: "yandex", + initialized: true, + searchFn: func(q Query) ([]SearchResult, error) { + yandexProxy = q.ProxyURL + return []SearchResult{{Rank: 1, URL: "https://example.com/yandex", Title: "yandex"}}, nil + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + Entries: []ProxyEntryConfig{ + {URL: "http://proxy-us:8080", Tags: []string{"us"}}, + }, + }, + EnginePolicies: map[string]string{"google": "us"}, + } + + srv := NewServerWithOptions("127.0.0.1", 7096, opts, googleEngine, yandexEngine) + if resp := request(t, srv, "/google/search?text=golang"); resp.StatusCode != http.StatusOK { + t.Fatalf("expected google request to succeed, got %d", resp.StatusCode) + } + if resp := request(t, srv, "/yandex/search?text=golang"); resp.StatusCode != http.StatusOK { + t.Fatalf("expected yandex request to succeed, got %d", resp.StatusCode) + } + + if googleProxy == "" { + t.Fatalf("expected proxied google request, got empty proxy") + } + if yandexProxy != "" { + t.Fatalf("expected direct yandex request, got proxy %q", yandexProxy) } } @@ -600,7 +811,7 @@ func TestCacheStatsDisabled(t *testing.T) { opts.CacheMaxSize = 0 srv := NewServerWithOptions("127.0.0.1", 7079, opts, engine) - resp := request(t, srv, "/cache/stats") + resp := request(t, srv, "/stats/cache") if resp.StatusCode != http.StatusOK { t.Fatalf("expected disabled cache stats to return 200, got %d", resp.StatusCode) } @@ -638,7 +849,7 @@ func TestCacheStatsReflectActivity(t *testing.T) { _ = request(t, srv, "/google/search?text=golang") _ = request(t, srv, "/google/search?text=fallback") - resp := request(t, srv, "/cache/stats") + resp := request(t, srv, "/stats/cache") if resp.StatusCode != http.StatusOK { t.Fatalf("expected cache stats endpoint to return 200, got %d", resp.StatusCode) } diff --git a/testutil/integration.go b/testutil/integration.go new file mode 100644 index 0000000..bef55af --- /dev/null +++ b/testutil/integration.go @@ -0,0 +1,25 @@ +package testutil + +import ( + "os" + "strings" + "testing" +) + +const IntegrationEnv = "OPENSERP_INTEGRATION_TESTS" + +func RequireIntegration(t *testing.T) { + t.Helper() + if strings.TrimSpace(os.Getenv(IntegrationEnv)) != "1" { + t.Skipf("set %s=1 to run integration tests", IntegrationEnv) + } +} + +func RequireEnv(t *testing.T, key string) string { + t.Helper() + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + t.Skipf("set %s to run this integration test", key) + } + return value +}