From e8fb0c24fa4dd0f585377ccb6bc99922e3d7389d Mon Sep 17 00:00:00 2001 From: Rustem Kamalov Date: Tue, 28 Apr 2026 03:46:08 +0300 Subject: [PATCH] feat(proxy): support X-Proxy-URL via per-process browser pool --- cmd/root.go | 86 ++++-- cmd/serve.go | 352 +++++++++++++++++++---- cmd/serve_test.go | 109 ++++++++ config.yaml | 10 +- core/browser.go | 353 ++++++++++++++++++----- core/browser_test.go | 157 +++++++++++ core/cache.go | 28 +- core/cache_test.go | 73 +++++ core/common.go | 43 +++ core/errors.go | 19 +- core/logger.go | 5 + core/middleware.go | 46 ++- core/proxy.go | 26 +- core/proxy_context.go | 37 +++ core/proxy_lane.go | 289 +++++++++++++++++++ core/proxy_lane_test.go | 101 +++++++ core/proxy_per_context_spike_test.go | 93 ++++++ core/proxy_test.go | 31 ++ core/resilient.go | 105 ++++++- core/retry.go | 62 ++-- core/server.go | 153 +++++++++- core/server_test.go | 404 ++++++++++++++++++++++++++- docs/openapi.yaml | 386 ++++++++++++++++++++++++- 23 files changed, 2732 insertions(+), 236 deletions(-) create mode 100644 core/proxy_context.go create mode 100644 core/proxy_lane.go create mode 100644 core/proxy_lane_test.go create mode 100644 core/proxy_per_context_spike_test.go diff --git a/cmd/root.go b/cmd/root.go index 40c31ac..d92f04d 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -5,6 +5,7 @@ import ( "os" "strconv" "strings" + "time" "github.com/karust/openserp/core" browserprofile "github.com/karust/openserp/core/browser" @@ -15,7 +16,7 @@ import ( ) const ( - version = "0.7.2" + version = "0.7.3" defaultConfigFilename = "config" envPrefix = "OPENSERP" ) @@ -52,16 +53,18 @@ type ServerConfig struct { } type AppConfig struct { - Timeout int `mapstructure:"timeout"` - BrowserPath string `mapstructure:"browser_path"` - ProfilesJSON string `mapstructure:"profiles"` - IsBrowserHead bool `mapstructure:"head"` - IsLeaveHead bool `mapstructure:"leave_head"` - IsLeakless bool `mapstructure:"leakless"` - BlockResources string `mapstructure:"block_resources"` - BlockTrackers bool `mapstructure:"block_trackers"` - DebugEndpoints bool `mapstructure:"debug_endpoints"` - LogFormat string `mapstructure:"log_format"` + Timeout int `mapstructure:"timeout"` + BrowserPath string `mapstructure:"browser_path"` + ProfilesJSON string `mapstructure:"profiles"` + IsBrowserHead bool `mapstructure:"head"` + IsLeaveHead bool `mapstructure:"leave_head"` + IsLeakless bool `mapstructure:"leakless"` + BlockResources string `mapstructure:"block_resources"` + BlockTrackers bool `mapstructure:"block_trackers"` + DebugEndpoints bool `mapstructure:"debug_endpoints"` + LogFormat string `mapstructure:"log_format"` + MaxProcesses int `mapstructure:"max_processes"` + IdleTTL time.Duration `mapstructure:"idle_ttl"` } type EngineConfig struct { @@ -148,11 +151,58 @@ var RootCmd = &cobra.Command{ config.App.LogFormat = logFormat core.InitLogger(config.Server.IsVerbose, config.Server.IsDebug, config.App.LogFormat) - logrus.WithField("config", fmt.Sprintf("%+v", config)).Debug("Final config") + logrus.WithField("config", sanitizedConfigForLog(config)).Debug("Final config") return nil }, } +func sanitizedConfigForLog(cfg Config) map[string]interface{} { + return map[string]interface{}{ + "server": cfg.Server, + "app": map[string]interface{}{ + "timeout": cfg.App.Timeout, + "browser_path": cfg.App.BrowserPath != "", + "profiles": cfg.App.ProfilesJSON != "", + "head": cfg.App.IsBrowserHead, + "leave_head": cfg.App.IsLeaveHead, + "leakless": cfg.App.IsLeakless, + "block_resources": cfg.App.BlockResources, + "block_trackers": cfg.App.BlockTrackers, + "debug_endpoints": cfg.App.DebugEndpoints, + "log_format": cfg.App.LogFormat, + "max_processes": cfg.App.MaxProcesses, + "idle_ttl": cfg.App.IdleTTL.String(), + }, + "proxies": map[string]interface{}{ + "global": maskedProxyForLog(cfg.Proxies.Global), + "entries": len(cfg.Proxies.Entries), + "allow_request_proxy_url": cfg.Proxies.AllowRequestProxyURL, + "health": cfg.Proxies.Health, + "lanes": cfg.Proxies.Lanes, + }, + "cache": cfg.Cache, + "resilience": cfg.Resilience, + "circuit_breaker": cfg.CircuitBreaker, + "cors": cfg.CORS, + "captcha": cfg.Captcha, + "2captcha": map[string]interface{}{ + "apikey_configured": strings.TrimSpace(cfg.Config2Capcha.ApiKey) != "", + }, + "google": cfg.GoogleConfig, + "yandex": cfg.YandexConfig, + "baidu": cfg.BaiduConfig, + "bing": cfg.BingConfig, + "duckduckgo": cfg.DuckDuckGoConfig, + } +} + +func maskedProxyForLog(proxyURL string) string { + if strings.TrimSpace(proxyURL) == "" { + return "" + } + return core.MaskProxyURL(proxyURL) +} + // Bind each cobra flag to its associated viper configuration (config file and environment variable) func bindFlags(cmd *cobra.Command, vpr *viper.Viper) { cmd.Flags().VisitAll(func(flg *pflag.Flag) { @@ -259,10 +309,6 @@ func initializeConfig(cmd *cobra.Command) error { return fmt.Errorf("invalid proxies config: %w", err) } - if config.Server.IsDebug { - logrus.Debug("Viper config:") - v.Debug() - } return nil } @@ -340,10 +386,16 @@ func setConfigDefaults(v *viper.Viper) { v.SetDefault("app.block_resources", "") v.SetDefault("app.block_trackers", false) v.SetDefault("app.debug_endpoints", false) + v.SetDefault("app.max_processes", 4) + v.SetDefault("app.idle_ttl", "10m") v.SetDefault("proxies.entries", []interface{}{}) v.SetDefault("proxies.global", "") + v.SetDefault("proxies.allow_request_proxy_url", false) v.SetDefault("proxies.health.failure_threshold", core.DefaultProxyFailureThreshold) + v.SetDefault("proxies.lanes.enabled", true) + v.SetDefault("proxies.lanes.max_lanes", core.DefaultProxyLaneMaxLanes) + v.SetDefault("proxies.lanes.drop_cookies_on_challenge", true) v.SetDefault("cache.ttl_seconds", 300) v.SetDefault("cache.max_size", 1000) @@ -356,7 +408,7 @@ func setConfigDefaults(v *viper.Viper) { v.SetDefault("cors.enabled", true) v.SetDefault("cors.allow_origins", "*") v.SetDefault("cors.allow_methods", "GET, POST, OPTIONS") - v.SetDefault("cors.allow_headers", "Origin, Content-Type, Accept, Authorization, X-Use-Proxy, X-Request-ID, X-Tenant") + v.SetDefault("cors.allow_headers", "Origin, Content-Type, Accept, Authorization, X-Use-Proxy, X-Proxy-URL, X-Proxy-Country, X-Proxy-Class, X-Proxy-Provider, X-Proxy-Session-ID, X-Request-ID, X-Tenant") v.SetDefault("cors.max_age", 86400) v.SetDefault("captcha.solver_enabled", false) } diff --git a/cmd/serve.go b/cmd/serve.go index 3548d44..44b6a18 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "net/url" "os" "os/signal" "strings" @@ -235,60 +236,284 @@ func isServerNotRunningError(err error) bool { return strings.Contains(strings.ToLower(err.Error()), "server is not running") } +// pooledBrowser is one Chrome process in the pool, dedicated to a single proxy +// auth identity (or to the shared no-auth/unauth path when launchProxyURL==""). +type pooledBrowser struct { + browser *core.Browser + launchProxyURL string + lastUsedAt time.Time +} + +// browserPool keeps a bounded set of Chrome processes keyed by proxy auth +// identity (scheme+host+port+username). Each entry was launched with its own +// `l.Proxy(...)` so Chrome handles 407 natively for the main document AND all +// subresources. Direct and unauthenticated proxies share one entry whose Chrome +// was launched without a process-level proxy; per-BrowserContext ProxyServer is +// applied at request time for unauthenticated request-URL proxies. type browserPool struct { - mu sync.Mutex - base core.BrowserOpts - browser map[string]*core.Browser + mu sync.Mutex + base core.BrowserOpts + laneStore *core.LaneStore + + maxProcesses int + idleTTL time.Duration + + browsers map[string]*pooledBrowser + + evictedLRU int + evictedIdle int + + stopSweeper chan struct{} + sweeperDone chan struct{} } -func newBrowserPool(base core.BrowserOpts) *browserPool { - return &browserPool{ - base: base, - browser: map[string]*core.Browser{}, +const directBrowserKey = "direct" + +func newBrowserPool(base core.BrowserOpts, defaultLaunchProxyURL string, laneStore *core.LaneStore, maxProcesses int, idleTTL time.Duration) *browserPool { + base.ProxyLaneStore = laneStore + if maxProcesses <= 0 { + maxProcesses = 4 } + pool := &browserPool{ + base: base, + laneStore: laneStore, + maxProcesses: maxProcesses, + idleTTL: idleTTL, + browsers: map[string]*pooledBrowser{}, + stopSweeper: make(chan struct{}), + sweeperDone: make(chan struct{}), + } + // A configured global proxy (legacy) becomes a pre-bound entry on the + // shared "direct" key so requests without a per-request proxy still use it. + if launchURL := strings.TrimSpace(defaultLaunchProxyURL); launchURL != "" { + pool.browsers[directBrowserKey] = &pooledBrowser{ + launchProxyURL: launchURL, + lastUsedAt: time.Now(), + } + } + if idleTTL > 0 { + go pool.sweepIdle() + } else { + close(pool.sweeperDone) + } + return pool } -func (p *browserPool) get(proxyURL string) (*core.Browser, error) { - key := strings.TrimSpace(proxyURL) - if key == "" { - key = "direct" +// browserPoolKey derives the pool key from a request's proxy URL. Authenticated +// HTTP/HTTPS proxies get their own Chrome keyed by scheme+host+port+username. +// Empty/unauthenticated/SOCKS request URLs fall through to the shared +// "direct" Chrome. +func browserPoolKey(requestProxyURL string) string { + requestProxyURL = strings.TrimSpace(requestProxyURL) + if requestProxyURL == "" { + return directBrowserKey + } + normalized, err := core.NormalizeProxyURL(requestProxyURL) + if err != nil || normalized == "" { + return directBrowserKey + } + parsed, err := url.Parse(normalized) + if err != nil { + return directBrowserKey + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + // Authenticated SOCKS is rejected upstream; unauthenticated SOCKS goes + // through the per-context proxy path on the shared Chrome. + return directBrowserKey + } + if parsed.User == nil { + return directBrowserKey + } + username := parsed.User.Username() + return fmt.Sprintf("%s|%s|%s", parsed.Scheme, parsed.Host, username) +} + +// browserLaunchURL returns the URL to pass to launcher.Proxy for a given +// request URL, or "" when the launch should be unproxied (direct + unauth). +func browserLaunchURL(requestProxyURL string) string { + requestProxyURL = strings.TrimSpace(requestProxyURL) + if requestProxyURL == "" { + return "" + } + normalized, err := core.NormalizeProxyURL(requestProxyURL) + if err != nil || normalized == "" { + return "" + } + parsed, err := url.Parse(normalized) + if err != nil { + return "" + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return "" + } + if parsed.User == nil { + return "" + } + return normalized +} + +// get returns a Chrome that can route the supplied request URL. For +// authenticated HTTP(S) proxies it returns the dedicated Chrome (launching one +// if needed). For everything else it returns the shared "direct" Chrome. +func (p *browserPool) get(requestProxyURL string) (*core.Browser, error) { + key := browserPoolKey(requestProxyURL) + launchURL := "" + if key != directBrowserKey { + launchURL = browserLaunchURL(requestProxyURL) } p.mu.Lock() defer p.mu.Unlock() - if b, ok := p.browser[key]; ok { - return b, nil + if entry, ok := p.browsers[key]; ok && entry.browser != nil { + entry.lastUsedAt = time.Now() + return entry.browser, nil + } + + // Use any pre-bound launchProxyURL on the existing entry (e.g. legacy + // global proxy) when the caller didn't supply one. + if entry, ok := p.browsers[key]; ok && entry.browser == nil { + if launchURL == "" { + launchURL = entry.launchProxyURL + } } opts := p.base - opts.ProxyURL = proxyURL - b, err := core.NewBrowser(opts) + opts.ProxyURL = launchURL + browser, 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 + p.browsers[key] = &pooledBrowser{ + browser: browser, + launchProxyURL: launchURL, + lastUsedAt: time.Now(), + } + p.evictLRULocked() + return browser, nil +} + +func (p *browserPool) evictLRULocked() { + for len(p.browsers) > p.maxProcesses { + var ( + oldestKey string + oldest time.Time + found bool + ) + for key, entry := range p.browsers { + if !found || entry.lastUsedAt.Before(oldest) { + oldestKey = key + oldest = entry.lastUsedAt + found = true + } + } + if !found { + return + } + entry := p.browsers[oldestKey] + delete(p.browsers, oldestKey) + p.evictedLRU++ + go closePooledBrowser(entry, "lru") + } +} + +func (p *browserPool) sweepIdle() { + defer close(p.sweeperDone) + interval := p.idleTTL / 4 + if interval < time.Second { + interval = time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-p.stopSweeper: + return + case now := <-ticker.C: + p.mu.Lock() + for key, entry := range p.browsers { + if entry.browser == nil { + continue + } + if now.Sub(entry.lastUsedAt) < p.idleTTL { + continue + } + delete(p.browsers, key) + p.evictedIdle++ + go closePooledBrowser(entry, "idle") + } + p.mu.Unlock() + } + } +} + +func closePooledBrowser(entry *pooledBrowser, reason string) { + if entry == nil || entry.browser == nil { + return + } + if err := entry.browser.Close(); err != nil { + logrus.WithError(err).WithField("evict_reason", reason).Debug("Browser pool: close evicted browser failed") + } +} + +func (p *browserPool) dropLaneCookies(ctx context.Context, engineName string, q core.Query) { + if p == nil || p.laneStore == nil { + return + } + laneKey := core.ProxyLaneKeyForTenant(engineName, core.TenantFromContext(ctx), q, q.ProxyURL) + p.laneStore.DropCookies(laneKey) +} + +func (p *browserPool) laneStats() core.LaneStats { + if p == nil || p.laneStore == nil { + return core.LaneStats{} + } + return p.laneStore.Stats() +} + +func (p *browserPool) browserStats() core.BrowserPoolStats { + if p == nil { + return core.BrowserPoolStats{} + } + p.mu.Lock() + active := 0 + for _, entry := range p.browsers { + if entry.browser != nil { + active++ + } + } + stats := core.BrowserPoolStats{ + Active: active, + Max: p.maxProcesses, + EvictedLRU: p.evictedLRU, + EvictedIdle: p.evictedIdle, + } + p.mu.Unlock() + return stats } func (p *browserPool) close() error { + if p == nil { + return nil + } + close(p.stopSweeper) + <-p.sweeperDone + p.mu.Lock() - browsers := make([]*core.Browser, 0, len(p.browser)) - for key, b := range p.browser { - browsers = append(browsers, b) - delete(p.browser, key) + entries := make([]*pooledBrowser, 0, len(p.browsers)) + for key, entry := range p.browsers { + entries = append(entries, entry) + delete(p.browsers, key) } p.mu.Unlock() var closeErr error - for _, browser := range browsers { - if browser == nil { + for _, entry := range entries { + if entry == nil || entry.browser == nil { continue } - if err := browser.Close(); err != nil { + if err := entry.browser.Close(); err != nil { closeErr = errors.Join(closeErr, err) } } @@ -302,12 +527,11 @@ type pooledBrowserEngine struct { factory func(core.Browser, core.SearchEngineOptions) core.SearchEngine pool *browserPool - mu sync.Mutex - engines map[string]core.SearchEngine + reportLaneStats bool } func (e *pooledBrowserEngine) Search(ctx context.Context, q core.Query) ([]core.SearchResult, error) { - engine, err := e.getOrCreate(q.ProxyURL) + engine, err := e.resolveEngine(q) if err != nil { return nil, err } @@ -315,7 +539,7 @@ func (e *pooledBrowserEngine) Search(ctx context.Context, q core.Query) ([]core. } func (e *pooledBrowserEngine) SearchImage(ctx context.Context, q core.Query) ([]core.SearchResult, error) { - engine, err := e.getOrCreate(q.ProxyURL) + engine, err := e.resolveEngine(q) if err != nil { return nil, err } @@ -334,27 +558,35 @@ 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" +func (e *pooledBrowserEngine) DropProxyLaneCookies(ctx context.Context, q core.Query) { + e.pool.dropLaneCookies(ctx, e.name, q) +} + +func (e *pooledBrowserEngine) ProxyLaneStats() core.LaneStats { + if !e.reportLaneStats { + return core.LaneStats{} } + return e.pool.laneStats() +} - e.mu.Lock() - defer e.mu.Unlock() - - if engine, ok := e.engines[key]; ok { - return engine, nil +func (e *pooledBrowserEngine) BrowserPoolStats() core.BrowserPoolStats { + if !e.reportLaneStats { + return core.BrowserPoolStats{} } + return e.pool.browserStats() +} - browser, err := e.pool.get(proxyURL) +// resolveEngine builds a fresh engine wrapper around the pool-resolved Browser. +// The wrapper is intentionally not cached: pool eviction can replace the Chrome +// behind a key, and a cached engine would carry a stale Browser value (closed +// connection, dead browserAddr). Engines are thin wrappers, so per-call +// construction is cheap. +func (e *pooledBrowserEngine) resolveEngine(q core.Query) (core.SearchEngine, error) { + browser, err := e.pool.get(q.ProxyURL) if err != nil { return nil, err } - - engine := e.factory(*browser, e.opts) - e.engines[key] = engine - return engine, nil + return e.factory(*browser, e.opts), nil } type browserEngineSpec struct { @@ -404,11 +636,27 @@ func browserEngineSpecs() []browserEngineSpec { } func buildBrowserEngines(baseOpts core.BrowserOpts, proxyCfg core.ProxyConfig) ([]core.SearchEngine, func() error, error) { - pool := newBrowserPool(baseOpts) + launchProxyURL := "" + if strings.TrimSpace(proxyCfg.Proxies.Global) != "" && !proxyCfg.Proxies.AllowRequestProxyURL { + launchProxyURL = proxyCfg.Proxies.Global + } + var laneStore *core.LaneStore + if proxyCfg.Proxies.Lanes.Enabled { + laneStore = core.NewLaneStore(proxyCfg.Proxies.Lanes.MaxLanes) + } + maxProcesses := config.App.MaxProcesses + if maxProcesses <= 0 { + maxProcesses = 4 + } + idleTTL := config.App.IdleTTL + if idleTTL < 0 { + idleTTL = 0 + } + pool := newBrowserPool(baseOpts, launchProxyURL, laneStore, maxProcesses, idleTTL) specs := browserEngineSpecs() engines := make([]core.SearchEngine, 0, len(specs)) - for _, spec := range specs { + for idx, spec := range specs { policy := resolveEngineProxyPolicy(proxyCfg, spec.name) if err := validateBrowserProxyPolicy(proxyCfg, policy); err != nil { return nil, nil, fmt.Errorf("browser proxy validation failed for engine %s: %w", spec.name, err) @@ -417,12 +665,12 @@ func buildBrowserEngines(baseOpts core.BrowserOpts, proxyCfg core.ProxyConfig) ( 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{}, + name: spec.name, + limiter: rate.NewLimiter(rate.Every(opts.GetRatelimit()), opts.RateBurst), + opts: opts, + factory: spec.factory, + pool: pool, + reportLaneStats: idx == 0, }) } diff --git a/cmd/serve_test.go b/cmd/serve_test.go index ea624da..2358103 100644 --- a/cmd/serve_test.go +++ b/cmd/serve_test.go @@ -3,10 +3,119 @@ package cmd import ( "strings" "testing" + "time" "github.com/karust/openserp/core" ) +func TestBrowserPoolKey(t *testing.T) { + cases := []struct { + name string + raw string + want string + }{ + {"empty -> direct", "", directBrowserKey}, + {"unauth http -> direct", "http://proxy.example:8080", directBrowserKey}, + {"unauth socks -> direct", "socks5://proxy.example:1080", directBrowserKey}, + {"auth socks -> direct (rejected upstream)", "socks5://user:pass@proxy.example:1080", directBrowserKey}, + {"auth http", "http://user:pass@proxy.example:8080", "http|proxy.example:8080|user"}, + {"auth https different scheme", "https://user:pass@proxy.example:8443", "https|proxy.example:8443|user"}, + {"different password same key", "http://user:other-pass@proxy.example:8080", "http|proxy.example:8080|user"}, + {"different user different key", "http://user2:pass@proxy.example:8080", "http|proxy.example:8080|user2"}, + {"different host different key", "http://user:pass@proxy2.example:8080", "http|proxy2.example:8080|user"}, + {"different port different key", "http://user:pass@proxy.example:9090", "http|proxy.example:9090|user"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := browserPoolKey(tc.raw); got != tc.want { + t.Fatalf("browserPoolKey(%q) = %q, want %q", tc.raw, got, tc.want) + } + }) + } +} + +func TestBrowserLaunchURL(t *testing.T) { + cases := []struct { + name string + raw string + want string + }{ + {"empty -> empty", "", ""}, + {"unauth http -> empty (per-context path)", "http://proxy.example:8080", ""}, + {"unauth socks -> empty", "socks5://proxy.example:1080", ""}, + {"auth http -> normalized", "http://user:pass@proxy.example:8080", "http://user:pass@proxy.example:8080"}, + {"auth https -> normalized", "https://u:p@proxy.example:8443", "https://u:p@proxy.example:8443"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := browserLaunchURL(tc.raw) + if got != tc.want { + t.Fatalf("browserLaunchURL(%q) = %q, want %q", tc.raw, got, tc.want) + } + }) + } +} + +func TestBrowserPoolEvictLRU(t *testing.T) { + // Pre-populate with bare entries (browser=nil) so we exercise eviction + // without launching real Chrome. closePooledBrowser handles nil safely. + pool := &browserPool{ + maxProcesses: 2, + browsers: map[string]*pooledBrowser{}, + stopSweeper: make(chan struct{}), + sweeperDone: make(chan struct{}), + } + close(pool.sweeperDone) + + now := time.Now() + pool.browsers["a"] = &pooledBrowser{lastUsedAt: now.Add(-3 * time.Second)} + pool.browsers["b"] = &pooledBrowser{lastUsedAt: now.Add(-2 * time.Second)} + pool.browsers["c"] = &pooledBrowser{lastUsedAt: now.Add(-1 * time.Second)} + + pool.mu.Lock() + pool.evictLRULocked() + pool.mu.Unlock() + + if _, ok := pool.browsers["a"]; ok { + t.Fatal("expected oldest entry 'a' to be evicted") + } + if _, ok := pool.browsers["b"]; !ok { + t.Fatal("expected entry 'b' to remain") + } + if _, ok := pool.browsers["c"]; !ok { + t.Fatal("expected entry 'c' to remain") + } + if pool.evictedLRU != 1 { + t.Fatalf("expected 1 LRU eviction, got %d", pool.evictedLRU) + } +} + +func TestBrowserPoolBrowserStats(t *testing.T) { + pool := &browserPool{ + maxProcesses: 4, + browsers: map[string]*pooledBrowser{}, + stopSweeper: make(chan struct{}), + sweeperDone: make(chan struct{}), + } + close(pool.sweeperDone) + + // Pre-bound entry without a launched browser should not count as active. + pool.browsers["pre-bound"] = &pooledBrowser{launchProxyURL: "http://u:p@proxy.example:8080", lastUsedAt: time.Now()} + pool.evictedLRU = 2 + pool.evictedIdle = 5 + + stats := pool.browserStats() + if stats.Max != 4 { + t.Fatalf("expected max=4, got %d", stats.Max) + } + if stats.Active != 0 { + t.Fatalf("expected active=0 (entry has no live browser), got %d", stats.Active) + } + if stats.EvictedLRU != 2 || stats.EvictedIdle != 5 { + t.Fatalf("unexpected stats: %#v", stats) + } +} + func TestValidateBrowserProxyPolicyRejectsAuthenticatedSocks(t *testing.T) { tests := []struct { name string diff --git a/config.yaml b/config.yaml index ed01039..cc8f0f8 100644 --- a/config.yaml +++ b/config.yaml @@ -15,8 +15,10 @@ app: 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 - + max_processes: 4 # LRU cap on concurrent Chrome processes + idle_ttl: 5m # close a Chrome that has not served traffic for this long proxies: + allow_request_proxy_url: false # Force a single proxy for all engines. # Same behavior as passing --proxy on the CLI. #global: http://127.0.0.1:8080 @@ -29,6 +31,10 @@ proxies: # tags: [eu] health: failure_threshold: 3 # Disable proxy after this many consecutive failures + lanes: + enabled: true # Reuse browser profile/cookies per engine + proxy session ID + max_lanes: 100 # LRU cap for sticky lanes kept in worker memory + drop_cookies_on_challenge: true # Clear lane cookies on captcha/challenge only cache: ttl_seconds: 60 # Dedicated endpoint cache TTL in seconds (0 disables cache) @@ -47,7 +53,7 @@ cors: enabled: true allow_origins: "*" allow_methods: "GET, POST, OPTIONS" - allow_headers: "Origin, Content-Type, Accept, Authorization, X-Use-Proxy" + allow_headers: "Origin, Content-Type, Accept, Authorization, X-Use-Proxy, X-Proxy-URL, X-Proxy-Country, X-Proxy-Class, X-Proxy-Provider, X-Proxy-Session-ID, X-Request-ID, X-Tenant" max_age: 86400 # 2captcha: diff --git a/core/browser.go b/core/browser.go index 92c44b0..c4fe2ef 100644 --- a/core/browser.go +++ b/core/browser.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "net/http" "net/url" "os" "regexp" @@ -46,6 +47,8 @@ type BrowserOpts struct { BrowserPath string // ProxyURL defines the upstream proxy for browser traffic. ProxyURL string + // ProxyLaneStore keeps sticky proxy lane profiles and cookies. + ProxyLaneStore *LaneStore // Insecure allows invalid TLS certificates for browser requests. Insecure bool // UserAgent optionally overrides browser-reported user agent during emulation. @@ -193,14 +196,18 @@ func (b *Browser) configureRequestBlocking(ctx context.Context, page *rod.Page) type Browser struct { BrowserOpts browserAddr string + proxyUser string + proxyPass string conn *browserConnection CaptchaSolver *CaptchaSolver } type browserConnection struct { - mu sync.Mutex - browser *rod.Browser - laneProfiles map[string]browserprofile.Profile + mu sync.Mutex + browser *rod.Browser + laneProfiles map[string]browserprofile.Profile + authCancel context.CancelFunc + authStopped chan struct{} } // NewBrowser launches a new Chromium process via Rod launcher and returns a @@ -210,7 +217,7 @@ func NewBrowser(opts BrowserOpts) (*Browser, error) { if strings.TrimSpace(opts.UserAgent) != "" { logrus.Warn("custom user_agent override can reduce profile coherence; use only for diagnostics") } - logrus.WithField("browser_options", fmt.Sprintf("%+v", opts)).Debug("Browser options") + logrus.WithFields(browserOptsLogFields(opts)).Debug("Browser options") path, err := resolveBrowserBinaryPath(opts.BrowserPath, launcher.LookPath) if err != nil { @@ -236,7 +243,13 @@ func NewBrowser(opts BrowserOpts) (*Browser, error) { l = l.Bin(path) } - // Configure proxy if specified + b := Browser{ + conn: &browserConnection{}, + } + + // Configure proxy if specified. Chrome's --proxy-server flag must NOT + // include credentials; we strip them here and reinject via a persistent + // CDP Fetch.handleAuthRequired listener installed after each connect. if opts.ProxyURL != "" { normalizedProxyURL, err := NormalizeProxyURL(opts.ProxyURL) if err != nil { @@ -249,26 +262,21 @@ func NewBrowser(opts BrowserOpts) (*Browser, error) { return nil, fmt.Errorf("invalid proxy URL: %v", err) } - // Chrome's proxy-server flag must not contain credentials. - // Auth (if needed) is handled separately via DevTools auth callbacks. proxyStr := proxyURLForBrowserLaunch(proxyUrl) logrus.WithField("proxy", MaskProxyURL(proxyStr)).Debug("Setting up proxy") l = l.Proxy(proxyStr) - // Check if proxy has auth credentials - if proxyUrl.User != nil { - username := proxyUrl.User.Username() + if proxyUrl.User != nil && (proxyUrl.Scheme == "http" || proxyUrl.Scheme == "https") { + b.proxyUser = proxyUrl.User.Username() + b.proxyPass, _ = proxyUrl.User.Password() logrus.WithFields(logrus.Fields{ "proxy_scheme": proxyUrl.Scheme, - "proxy_username": username, - }).Debugf("Proxy credentials configured for %s proxy: %s:****", proxyUrl.Scheme, username) + "proxy_username": b.proxyUser, + }).Debugf("Proxy credentials configured for %s proxy: %s:****", proxyUrl.Scheme, b.proxyUser) } } - b := Browser{ - BrowserOpts: opts, - conn: &browserConnection{}, - } + b.BrowserOpts = opts b.browserAddr, err = l.Launch() if opts.CaptchaSolverEnabled && opts.CaptchaSolverApiKey != "" { @@ -279,6 +287,33 @@ func NewBrowser(opts BrowserOpts) (*Browser, error) { return &b, err } +func browserOptsLogFields(opts BrowserOpts) logrus.Fields { + return logrus.Fields{ + "headless": opts.IsHeadless, + "leakless": opts.IsLeakless, + "timeout": opts.Timeout.String(), + "language_code": opts.LanguageCode, + "wait_requests": opts.WaitRequests, + "leave_page_open": opts.LeavePageOpen, + "captcha_solver_enabled": opts.CaptchaSolverEnabled, + "captcha_solver_has_key": strings.TrimSpace(opts.CaptchaSolverApiKey) != "", + "browser_path_configured": strings.TrimSpace(opts.BrowserPath) != "", + "proxy": maskedProxyLogValue(opts.ProxyURL), + "insecure": opts.Insecure, + "user_agent_override": strings.TrimSpace(opts.UserAgent) != "", + "block_resource_types": len(opts.BlockResourceTypes), + "block_trackers": opts.BlockTrackers, + "proxy_lanes_enabled": opts.ProxyLaneStore != nil, + } +} + +func maskedProxyLogValue(proxyURL string) string { + if strings.TrimSpace(proxyURL) == "" { + return "" + } + return MaskProxyURL(proxyURL) +} + func proxyURLForBrowserLaunch(u *url.URL) string { if u == nil { return "" @@ -359,9 +394,88 @@ func (b *Browser) connectBrowser() (*rod.Browser, error) { } } + if b.proxyUser != "" { + if err := b.startProxyAuthListener(browser); err != nil { + return nil, fmt.Errorf("install proxy auth listener: %w", err) + } + } + return browser, nil } +// startProxyAuthListener enables the Fetch domain with HandleAuthRequests=true +// and runs a goroutine that responds to every Fetch.requestPaused (continue +// the request) and every Fetch.authRequired (provide credentials). The +// listener lives for the lifetime of the rod.Browser session and replaces +// rod's single-shot HandleAuth helper. +func (b *Browser) startProxyAuthListener(browser *rod.Browser) error { + state := b.connectionState() + + // Stop a previous listener attached to a stale connection. + if state.authCancel != nil { + state.authCancel() + if state.authStopped != nil { + <-state.authStopped + } + state.authCancel = nil + state.authStopped = nil + } + + listenCtx, cancel := context.WithCancel(context.Background()) + stopped := make(chan struct{}) + state.authCancel = cancel + state.authStopped = stopped + + username := b.proxyUser + password := b.proxyPass + scoped := browser.Context(listenCtx) + started := make(chan struct{}) + + go func() { + defer close(stopped) + // Subscribe via EachEvent. The wait function it returns blocks until + // listenCtx is cancelled. We close `started` after EachEvent has + // installed its handlers but before we wait, so the caller can safely + // enable the Fetch domain without racing the listener install. + wait := scoped.EachEvent( + func(e *proto.FetchAuthRequired) bool { + resp := proto.FetchAuthChallengeResponseResponseProvideCredentials + err := proto.FetchContinueWithAuth{ + RequestID: e.RequestID, + AuthChallengeResponse: &proto.FetchAuthChallengeResponse{ + Response: resp, + Username: username, + Password: password, + }, + }.Call(scoped) + if err != nil && !errors.Is(err, context.Canceled) { + logrus.WithError(err).Debug("Proxy auth response failed") + } + return false + }, + func(e *proto.FetchRequestPaused) bool { + err := proto.FetchContinueRequest{RequestID: e.RequestID}.Call(scoped) + if err != nil && !errors.Is(err, context.Canceled) { + logrus.WithError(err).Debug("Continue paused request failed") + } + return false + }, + ) + close(started) + wait() + }() + + <-started + if err := (proto.FetchEnable{HandleAuthRequests: true}).Call(browser); err != nil { + cancel() + <-stopped + state.authCancel = nil + state.authStopped = nil + return fmt.Errorf("enable Fetch domain: %w", err) + } + return nil +} + func (b *Browser) ensureConnectedBrowser(ctx context.Context, forceReconnect bool) (*rod.Browser, error) { if b == nil || b.browserAddr == "" { return nil, fmt.Errorf("browser is not initialized") @@ -392,8 +506,17 @@ func (b *Browser) ensureConnectedBrowser(ctx context.Context, forceReconnect boo return state.browser, nil } -func createIsolatedPage(browser *rod.Browser) (*rod.Page, proto.BrowserBrowserContextID, error) { - browserContext, err := (proto.TargetCreateBrowserContext{}).Call(browser) +func createIsolatedPage(browser *rod.Browser, proxyURL string) (*rod.Page, proto.BrowserBrowserContextID, error) { + create := proto.TargetCreateBrowserContext{} + if strings.TrimSpace(proxyURL) != "" { + parsed, err := url.Parse(proxyURL) + if err != nil { + return nil, "", err + } + create.ProxyServer = proxyURLForBrowserLaunch(parsed) + } + + browserContext, err := create.Call(browser) if err != nil { return nil, "", err } @@ -425,40 +548,6 @@ func disposeBrowserContext(browser *rod.Browser, browserContextID proto.BrowserB }).Call(browser) } -func (b *Browser) startProxyAuthHandler(ctx context.Context, browser *rod.Browser) (context.CancelFunc, error) { - if browser == nil || b.ProxyURL == "" { - return nil, nil - } - - proxyURL, err := url.Parse(b.ProxyURL) - if err != nil { - return nil, fmt.Errorf("parse proxy URL: %w", err) - } - - if proxyURL.User == nil { - return nil, nil - } - - if proxyURL.Scheme != "http" && proxyURL.Scheme != "https" { - if proxyURL.Scheme == "socks5" || proxyURL.Scheme == "socks5h" { - logrus.Debug("SOCKS proxy credentials are not handled by browser auth callback") - } - return nil, nil - } - - username := proxyURL.User.Username() - password, _ := proxyURL.User.Password() - authCtx, cancel := context.WithCancel(EnsureContext(ctx)) - - go func() { - if err := browser.Context(authCtx).HandleAuth(username, password)(); err != nil && !errors.Is(err, context.Canceled) { - WithRequest(ctx).WithError(err).Debug("Proxy auth handler stopped") - } - }() - - return cancel, nil -} - var chromeVersionPattern = regexp.MustCompile(`(?:HeadlessChrome|Chrome)/([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)`) func (b *Browser) laneProfile(ctx context.Context, browser *rod.Browser) (browserprofile.Profile, string) { @@ -467,6 +556,20 @@ func (b *Browser) laneProfile(ctx context.Context, browser *rod.Browser) (browse if region == "" { region = strings.TrimSpace(b.LanguageCode) } + + if laneKey := proxyLaneKeyFromContext(ctx); !laneKey.Empty() && b.ProxyLaneStore != nil { + profile := b.ProxyLaneStore.Profile(laneKey, func() browserprofile.Profile { + selected := browserprofile.SelectProfile(engine, region) + selected = applyRuntimeBrowserVersion(selected, browser) + selected = applyProfileLanguageHint(selected, region) + if overrideUA := strings.TrimSpace(b.UserAgent); overrideUA != "" { + selected.UserAgent = overrideUA + } + return selected + }) + return profile, laneKey.ID() + } + laneKey := browserprofile.LaneKey(engine, region) state := b.connectionState() @@ -703,6 +806,103 @@ func applyProfile(page *rod.Page, profile browserprofile.Profile) error { return nil } +func (b *Browser) restoreLaneCookies(ctx context.Context, page *rod.Page) error { + if b == nil || b.ProxyLaneStore == nil { + return nil + } + laneKey := proxyLaneKeyFromContext(ctx) + if laneKey.Empty() { + return nil + } + cookies := b.ProxyLaneStore.Cookies(laneKey) + if len(cookies) == 0 { + return nil + } + if err := (proto.NetworkSetCookies{Cookies: cookieParams(cookies)}).Call(page); err != nil { + return fmt.Errorf("restore lane cookies: %w", err) + } + return nil +} + +func (b *Browser) saveLaneCookies(ctx context.Context, page *rod.Page, pageURL string) { + if b == nil || b.ProxyLaneStore == nil || page == nil { + return + } + laneKey := proxyLaneKeyFromContext(ctx) + if laneKey.Empty() { + return + } + res, err := (proto.NetworkGetCookies{Urls: []string{pageURL}}).Call(page) + if err != nil { + WithRequest(ctx).WithError(err).Debug("Save lane cookies failed") + return + } + b.ProxyLaneStore.SaveCookies(laneKey, res.Cookies) +} + +type mainDocumentStatusWatcher struct { + cancel context.CancelFunc + done chan struct{} + mu sync.Mutex + status int +} + +func startMainDocumentStatusWatcher(ctx context.Context, page *rod.Page) *mainDocumentStatusWatcher { + watchCtx, cancel := context.WithCancel(EnsureContext(ctx)) + watcher := &mainDocumentStatusWatcher{ + cancel: cancel, + done: make(chan struct{}), + } + + wait := page.Context(watchCtx).EachEvent(func(e *proto.NetworkResponseReceived) bool { + if e == nil || e.Response == nil || e.Type != proto.NetworkResourceTypeDocument { + return false + } + watcher.mu.Lock() + watcher.status = e.Response.Status + watcher.mu.Unlock() + return false + }) + + go func() { + defer close(watcher.done) + wait() + }() + + return watcher +} + +func (w *mainDocumentStatusWatcher) Stop() { + if w == nil { + return + } + w.cancel() + select { + case <-w.done: + case <-time.After(100 * time.Millisecond): + } +} + +func (w *mainDocumentStatusWatcher) Status() int { + if w == nil { + return 0 + } + w.mu.Lock() + defer w.mu.Unlock() + return w.status +} + +func classifyMainDocumentStatus(status int) error { + switch status { + case http.StatusForbidden: + return ErrBlocked + case http.StatusTooManyRequests: + return ErrRateLimited + default: + return nil + } +} + func profileNavigatorLanguages(profile browserprofile.Profile) []string { langs := make([]string, 0, len(profile.NavigatorLangs)) for _, language := range profile.NavigatorLangs { @@ -783,20 +983,35 @@ func (b *Browser) Navigate(ctx context.Context, URL string) (*rod.Page, error) { } WithRequest(ctx).WithField("url", URL).Debug("Navigate") + // Per-context proxy override is only used for unauthenticated proxies on a + // Chrome that was launched without a process-level proxy. When this Browser + // was launched with a proxy (b.ProxyURL set), Chrome handles routing and + // auth natively for the whole process; per-context override is skipped to + // avoid breaking Chrome's auth flow. + contextProxyURL := "" + if strings.TrimSpace(b.ProxyURL) == "" { + contextProxyURL = requestProxyURLFromContext(ctx) + } + hasProxy := contextProxyURL != "" || strings.TrimSpace(b.ProxyURL) != "" browser, err := b.ensureConnectedBrowser(ctx, false) if err != nil { return nil, fmt.Errorf("browser connect failed: %w", err) } + if hasProxy || b.Insecure { + if err := browser.IgnoreCertErrors(true); err != nil { + return nil, fmt.Errorf("ignore cert errors failed: %w", err) + } + } - page, browserContextID, err := createIsolatedPage(browser) + page, browserContextID, err := createIsolatedPage(browser, contextProxyURL) if err != nil { // Single-shot reconnect for stale websocket sessions. browser, err = b.ensureConnectedBrowser(ctx, true) if err != nil { return nil, fmt.Errorf("create isolated page failed, reconnect also failed: %w", err) } - page, browserContextID, err = createIsolatedPage(browser) + page, browserContextID, err = createIsolatedPage(browser, contextProxyURL) if err != nil { return nil, fmt.Errorf("create isolated page failed after reconnect: %w", err) } @@ -814,31 +1029,28 @@ func (b *Browser) Navigate(ctx context.Context, URL string) (*rod.Page, error) { } } - cancelProxyAuth, err := b.startProxyAuthHandler(ctx, browser) - if err != nil { - closeOnErr() - return nil, fmt.Errorf("proxy auth setup failed: %w", err) - } - if cancelProxyAuth != nil { - defer cancelProxyAuth() - } - profile, laneKey := b.laneProfile(ctx, browser) if err := applyProfile(page, profile); err != nil { closeOnErr() return nil, fmt.Errorf("apply profile %s (%s) failed: %w", profile.ID, laneKey, err) } + if err := b.restoreLaneCookies(ctx, page); err != nil { + closeOnErr() + return nil, err + } page = page.Context(ctx) if err := b.configureRequestBlocking(ctx, page); err != nil { closeOnErr() return nil, fmt.Errorf("configure request blocking failed: %w", err) } + statusWatcher := startMainDocumentStatusWatcher(ctx, page) + defer statusWatcher.Stop() timedPage := page.Timeout(b.Timeout) if err := timedPage.Navigate(URL); err != nil { closeOnErr() - return nil, err + return nil, classifyProxyNetworkError(err) } // Avoid panics from MustWaitLoad when the target navigates/closes mid-wait @@ -865,6 +1077,11 @@ func (b *Browser) Navigate(ctx context.Context, URL string) (*rod.Page, error) { if err := page.Context(ctx).WaitStable(800 * time.Millisecond); err != nil { WithRequest(ctx).WithError(err).Debug("WaitStable returned early; continuing") } + if err := classifyMainDocumentStatus(statusWatcher.Status()); err != nil { + closeOnErr() + return nil, err + } + b.saveLaneCookies(ctx, page, URL) return page, nil } @@ -889,6 +1106,14 @@ func (b *Browser) Close() error { } } + if state.authCancel != nil { + state.authCancel() + if state.authStopped != nil { + <-state.authStopped + } + state.authCancel = nil + state.authStopped = nil + } state.browser = nil state.laneProfiles = nil if err := browser.Close(); err != nil && !isBrowserClosedError(err) { diff --git a/core/browser_test.go b/core/browser_test.go index 44bc1bb..676daf1 100644 --- a/core/browser_test.go +++ b/core/browser_test.go @@ -6,6 +6,7 @@ package core import ( "context" "encoding/json" + "errors" "fmt" "net/http" "net/http/httptest" @@ -103,6 +104,162 @@ func TestNavigateUsesIsolatedBrowserContext(t *testing.T) { } } +func TestNavigateReusesCookiesForSameProxyLane(t *testing.T) { + testutil.RequireIntegration(t) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/cookies/set": + http.SetCookie(w, &http.Cookie{Name: "openserp_lane", Value: "same-lane", Path: "/"}) + _, _ = w.Write([]byte("cookie-set")) + case "/cookies": + _, _ = w.Write([]byte(r.Header.Get("Cookie"))) + default: + http.NotFound(w, r) + } + })) + defer srv.Close() + + browser, err := NewBrowser(BrowserOpts{ + IsHeadless: true, + IsLeakless: false, + Timeout: 15 * time.Second, + ProxyLaneStore: NewLaneStore(10), + }) + if err != nil { + t.Fatalf("failed initializing browser: %s", err) + } + defer closeTestBrowser(t, browser) + + laneCtx := WithProxyLaneKey(WithEngine(context.Background(), "google"), ProxyLaneKey{Engine: "google", SessionID: "sid-a"}) + pageA, err := browser.Navigate(laneCtx, srv.URL+"/cookies/set") + if err != nil { + t.Fatalf("navigate cookie setter: %v", err) + } + if err := ClosePageWithTimeout(context.Background(), pageA, time.Second); err != nil { + t.Fatalf("close setter page: %v", err) + } + + pageB, err := browser.Navigate(laneCtx, srv.URL+"/cookies") + if err != nil { + t.Fatalf("navigate cookie reader: %v", err) + } + defer func() { + if err := ClosePageWithTimeout(context.Background(), pageB, time.Second); err != nil { + t.Logf("close reader page: %v", err) + } + }() + + body, err := pageB.Timeout(5 * time.Second).Element("body") + if err != nil { + t.Fatalf("read response body: %v", err) + } + cookieHeader, err := body.Text() + if err != nil { + t.Fatalf("extract response text: %v", err) + } + if !strings.Contains(cookieHeader, "openserp_lane=same-lane") { + t.Fatalf("expected same proxy lane to restore cookie, got %q", cookieHeader) + } +} + +func TestNavigateDoesNotShareCookiesAcrossProxyLanes(t *testing.T) { + testutil.RequireIntegration(t) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/cookies/set": + http.SetCookie(w, &http.Cookie{Name: "openserp_lane", Value: "lane-a", Path: "/"}) + _, _ = w.Write([]byte("cookie-set")) + case "/cookies": + _, _ = w.Write([]byte(r.Header.Get("Cookie"))) + default: + http.NotFound(w, r) + } + })) + defer srv.Close() + + browser, err := NewBrowser(BrowserOpts{ + IsHeadless: true, + IsLeakless: false, + Timeout: 15 * time.Second, + ProxyLaneStore: NewLaneStore(10), + }) + if err != nil { + t.Fatalf("failed initializing browser: %s", err) + } + defer closeTestBrowser(t, browser) + + laneA := WithProxyLaneKey(WithEngine(context.Background(), "google"), ProxyLaneKey{Engine: "google", SessionID: "sid-a"}) + laneB := WithProxyLaneKey(WithEngine(context.Background(), "google"), ProxyLaneKey{Engine: "google", SessionID: "sid-b"}) + pageA, err := browser.Navigate(laneA, srv.URL+"/cookies/set") + if err != nil { + t.Fatalf("navigate cookie setter: %v", err) + } + if err := ClosePageWithTimeout(context.Background(), pageA, time.Second); err != nil { + t.Fatalf("close setter page: %v", err) + } + + pageB, err := browser.Navigate(laneB, srv.URL+"/cookies") + if err != nil { + t.Fatalf("navigate cookie reader: %v", err) + } + defer func() { + if err := ClosePageWithTimeout(context.Background(), pageB, time.Second); err != nil { + t.Logf("close reader page: %v", err) + } + }() + + body, err := pageB.Timeout(5 * time.Second).Element("body") + if err != nil { + t.Fatalf("read response body: %v", err) + } + cookieHeader, err := body.Text() + if err != nil { + t.Fatalf("extract response text: %v", err) + } + if strings.Contains(cookieHeader, "openserp_lane=lane-a") { + t.Fatalf("cookie leaked across proxy lanes; got header %q", cookieHeader) + } +} + +func TestNavigateClassifiesMainDocumentStatus(t *testing.T) { + testutil.RequireIntegration(t) + + tests := []struct { + name string + status int + want error + }{ + {name: "blocked", status: http.StatusForbidden, want: ErrBlocked}, + {name: "rate limited", status: http.StatusTooManyRequests, want: ErrRateLimited}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(tt.status) + _, _ = w.Write([]byte("classified")) + })) + defer srv.Close() + + browser, err := NewBrowser(BrowserOpts{IsHeadless: true, IsLeakless: false, Timeout: 15 * time.Second}) + if err != nil { + t.Fatalf("failed initializing browser: %s", err) + } + defer closeTestBrowser(t, browser) + + page, err := browser.Navigate(context.Background(), srv.URL) + if page != nil { + _ = ClosePageWithTimeout(context.Background(), page, time.Second) + } + if !errors.Is(err, tt.want) { + t.Fatalf("expected %v, got %v", tt.want, err) + } + }) + } +} + func TestFingerprintDetectors(t *testing.T) { testutil.RequireIntegration(t) if strings.TrimSpace(os.Getenv(botFingerprintTestsEnv)) != "1" { diff --git a/core/cache.go b/core/cache.go index e7b67c4..eef40ef 100644 --- a/core/cache.go +++ b/core/cache.go @@ -4,6 +4,7 @@ import ( "crypto/sha256" "encoding/hex" "fmt" + "strings" "sync" "time" ) @@ -35,8 +36,9 @@ func NewResponseCache(ttl time.Duration, maxSize int) *ResponseCache { } func BuildCacheKey(engine string, action string, q Query) string { + country, class, provider := cacheProxyMarket(q) raw := fmt.Sprintf( - "%s|%s|%s|%s|%s|%s|%s|%d|%d|%t|%t|%s", + "%s|%s|%s|%s|%s|%s|%s|%d|%d|%t|%t|%s|%s|%s", engine, action, q.Text, @@ -48,12 +50,34 @@ func BuildCacheKey(engine string, action string, q Query) string { q.Start, q.Filter, q.Answers, - q.ProxyOverride, + country, + class, + provider, ) hash := sha256.Sum256([]byte(raw)) return hex.EncodeToString(hash[:]) } +func cacheProxyMarket(q Query) (country string, class string, provider string) { + country = strings.ToLower(strings.TrimSpace(q.ProxyCountry)) + if country == "" { + // TODO: Use explicit balancer market metadata everywhere; language is only a weak market proxy. + country = strings.ToLower(strings.TrimSpace(q.LangCode)) + } + return country, + strings.ToLower(strings.TrimSpace(q.ProxyClass)), + strings.ToLower(strings.TrimSpace(q.ProxyProvider)) +} + +func ShouldBypassCacheForProxyMarket(q Query) bool { + if strings.TrimSpace(q.ProxyURL) == "" && strings.TrimSpace(q.ProxyOverride) == "" { + return false + } + return strings.TrimSpace(q.ProxyCountry) == "" && + strings.TrimSpace(q.ProxyClass) == "" && + strings.TrimSpace(q.ProxyProvider) == "" +} + func (c *ResponseCache) Get(key string) ([]byte, bool) { c.mu.Lock() defer c.mu.Unlock() diff --git a/core/cache_test.go b/core/cache_test.go index 36f3423..e29f8d0 100644 --- a/core/cache_test.go +++ b/core/cache_test.go @@ -153,3 +153,76 @@ func TestBuildCacheKeyChangesWithPaginationAndFlags(t *testing.T) { t.Fatal("expected answers to affect cache key") } } + +func TestBuildCacheKeyUsesProxyMarketNotSessionOrURL(t *testing.T) { + base := Query{ + Text: "golang", + LangCode: "EN", + Limit: 10, + ProxyURL: "http://user:password-a@proxy-a:8080", + ProxyCountry: " US ", + ProxyClass: " Residential ", + ProxyProvider: " WebShare ", + ProxySessionID: "sid-a", + } + baseKey := BuildCacheKey("google", "search", base) + + sameMarket := base + sameMarket.ProxyURL = "http://user:password-b@proxy-b:8080" + sameMarket.ProxySessionID = "sid-b" + if got := BuildCacheKey("google", "search", sameMarket); got != baseKey { + t.Fatal("expected proxy URL and session id not to affect cache key") + } + + differentCountry := base + differentCountry.ProxyCountry = "de" + if got := BuildCacheKey("google", "search", differentCountry); got == baseKey { + t.Fatal("expected proxy country to affect cache key") + } + + differentClass := base + differentClass.ProxyClass = "datacenter" + if got := BuildCacheKey("google", "search", differentClass); got == baseKey { + t.Fatal("expected proxy class to affect cache key") + } + + differentProvider := base + differentProvider.ProxyProvider = "brightdata" + if got := BuildCacheKey("google", "search", differentProvider); got == baseKey { + t.Fatal("expected proxy provider to affect cache key") + } +} + +func TestBuildCacheKeyFallsBackToLanguageWhenCountryAbsent(t *testing.T) { + base := Query{Text: "golang", LangCode: "en", Limit: 10} + baseKey := BuildCacheKey("google", "search", base) + + changed := base + changed.LangCode = "de" + if got := BuildCacheKey("google", "search", changed); got == baseKey { + t.Fatal("expected language fallback to affect cache key when proxy country is absent") + } + + withCountry := base + withCountry.ProxyCountry = "us" + changedWithCountry := withCountry + changedWithCountry.LangCode = "de" + if got := BuildCacheKey("google", "search", changedWithCountry); got == BuildCacheKey("google", "search", withCountry) { + t.Fatal("expected language itself to remain part of the cache key") + } +} + +func TestShouldBypassCacheForProxyMarket(t *testing.T) { + if !ShouldBypassCacheForProxyMarket(Query{ProxyURL: "http://proxy.example:8080"}) { + t.Fatal("expected request proxy without market metadata to bypass cache") + } + if !ShouldBypassCacheForProxyMarket(Query{ProxyOverride: "us"}) { + t.Fatal("expected tag override without market metadata to bypass cache") + } + if ShouldBypassCacheForProxyMarket(Query{ProxyURL: "http://proxy.example:8080", ProxyCountry: "us"}) { + t.Fatal("expected explicit country market metadata to allow cache") + } + if ShouldBypassCacheForProxyMarket(Query{Text: "golang"}) { + t.Fatal("expected direct query without proxy override to allow cache") + } +} diff --git a/core/common.go b/core/common.go index 67c6a79..d9110d5 100644 --- a/core/common.go +++ b/core/common.go @@ -45,6 +45,12 @@ var ErrTimeout = errors.New("timeout") // It is not a failure; the proxy stays healthy and no credit is charged. var ErrEmptyResult = errors.New("empty_result") +// ErrBlocked is returned when the search engine blocks the browser request. +var ErrBlocked = errors.New("blocked") + +// ErrRateLimited is returned when the search engine returns an HTTP rate limit. +var ErrRateLimited = errors.New("rate_limited") + // IsProxyNetworkError reports whether err is a network-level error that // indicates a faulty proxy (connect failure, auth rejection, or timeout). // Parser drift, captcha pages, and engine errors must NOT degrade proxy health. @@ -165,6 +171,14 @@ type Query struct { Answers bool // ProxyURL is a direct proxy URL used by raw HTTP search paths. ProxyURL string + // ProxyCountry identifies the proxy market country for cache/error metadata. + ProxyCountry string + // ProxyClass identifies the proxy class such as datacenter or residential. + ProxyClass string + // ProxyProvider identifies the upstream proxy provider. + ProxyProvider string + // ProxySessionID identifies a sticky balancer session/lane. + ProxySessionID string // ProxyOverride is a request-scoped proxy policy override (tag or "direct"), // typically parsed from the X-Use-Proxy header. ProxyOverride string @@ -172,6 +186,23 @@ type Query struct { Insecure bool } +// String renders Query for logs with the proxy URL credentials masked. The +// default %+v formatter calls this method, so logging Query through %v/%+v +// never leaks proxy passwords. +func (q Query) String() string { + maskedProxyURL := "" + if q.ProxyURL != "" { + maskedProxyURL = MaskProxyURL(q.ProxyURL) + } + return fmt.Sprintf( + "{Text:%s LangCode:%s DateInterval:%s Filetype:%s Site:%s Limit:%d Start:%d Filter:%t Answers:%t ProxyURL:%s ProxyCountry:%s ProxyClass:%s ProxyProvider:%s ProxySessionID:%s ProxyOverride:%s Insecure:%t}", + q.Text, q.LangCode, q.DateInterval, q.Filetype, q.Site, + q.Limit, q.Start, q.Filter, q.Answers, + maskedProxyURL, q.ProxyCountry, q.ProxyClass, q.ProxyProvider, + q.ProxySessionID, q.ProxyOverride, q.Insecure, + ) +} + // ComputePagination translates an absolute start offset into page index and // in-page offset for a fixed page size. func ComputePagination(start int, pageSize int) (int, int, error) { @@ -239,6 +270,18 @@ func (searchQuery *Query) InitFromContext(reqCtx *fiber.Ctx) error { if err != nil { return errInvalidParam(fmt.Sprintf("X-Use-Proxy: %v", err)) } + rawProxyURL := strings.TrimSpace(reqCtx.Get("X-Proxy-URL")) + if rawProxyURL != "" { + normalized, err := NormalizeProxyURL(rawProxyURL) + if err != nil { + return errInvalidParam(fmt.Sprintf("X-Proxy-URL: %v", err)) + } + searchQuery.ProxyURL = normalized + } + searchQuery.ProxyCountry = strings.ToLower(strings.TrimSpace(reqCtx.Get("X-Proxy-Country"))) + searchQuery.ProxyClass = strings.ToLower(strings.TrimSpace(reqCtx.Get("X-Proxy-Class"))) + searchQuery.ProxyProvider = strings.ToLower(strings.TrimSpace(reqCtx.Get("X-Proxy-Provider"))) + searchQuery.ProxySessionID = strings.TrimSpace(reqCtx.Get("X-Proxy-Session-ID")) if searchQuery.IsEmpty() { return errEmptyQuery() diff --git a/core/errors.go b/core/errors.go index a958bc4..4c84f85 100644 --- a/core/errors.go +++ b/core/errors.go @@ -5,22 +5,29 @@ import "fmt" // APIError represents a client-facing error with a stable machine-readable reason code. type APIError struct { HTTPStatus int + ErrorCode string Reason string Message string + Meta map[string]interface{} } func (e *APIError) Error() string { + if e.Reason == "" { + return e.Message + } return fmt.Sprintf("%s: %s", e.Reason, e.Message) } // Common validation reason codes. const ( - ReasonInvalidLimit = "INVALID_LIMIT" - ReasonInvalidStart = "INVALID_START" - ReasonInvalidParam = "INVALID_PARAM" - ReasonEmptyQuery = "EMPTY_QUERY" - ReasonNoEngines = "NO_ENGINES" - ReasonUnknownFormat = "UNKNOWN_FORMAT" + ReasonInvalidLimit = "INVALID_LIMIT" + ReasonInvalidStart = "INVALID_START" + ReasonInvalidParam = "INVALID_PARAM" + ReasonEmptyQuery = "EMPTY_QUERY" + ReasonNoEngines = "NO_ENGINES" + ReasonUnknownFormat = "UNKNOWN_FORMAT" + ReasonRequestProxyURLDisabled = "REQUEST_PROXY_URL_DISABLED" + ReasonUnsupportedProxyScheme = "UNSUPPORTED_PROXY_SCHEME" ) func errInvalidLimit(msg string) *APIError { diff --git a/core/logger.go b/core/logger.go index 9dc44cc..7e091f8 100644 --- a/core/logger.go +++ b/core/logger.go @@ -77,6 +77,11 @@ func RequestIDFromContext(ctx context.Context) string { return strings.TrimSpace(value) } +func TenantFromContext(ctx context.Context) string { + value, _ := EnsureContext(ctx).Value(tenantContextKey).(string) + return strings.TrimSpace(value) +} + func WithRequest(ctx context.Context) *logrus.Entry { ctx = EnsureContext(ctx) fields := logrus.Fields{} diff --git a/core/middleware.go b/core/middleware.go index 36cf105..948fcd2 100644 --- a/core/middleware.go +++ b/core/middleware.go @@ -11,10 +11,11 @@ import ( ) type JSONErrorResponse struct { - Error string `json:"error"` - Code int `json:"code"` - Message string `json:"message,omitempty"` - Reason string `json:"reason,omitempty"` + Error string `json:"error"` + Code int `json:"code"` + Message string `json:"message,omitempty"` + Reason string `json:"reason,omitempty"` + Meta map[string]interface{} `json:"meta,omitempty"` } type CORSConfig struct { @@ -28,7 +29,7 @@ func DefaultCORSConfig() CORSConfig { return CORSConfig{ AllowOrigins: "*", AllowMethods: "GET, POST, OPTIONS", - AllowHeaders: "Origin, Content-Type, Accept, Authorization, X-Use-Proxy, X-Request-ID, X-Tenant", + AllowHeaders: "Origin, Content-Type, Accept, Authorization, X-Use-Proxy, X-Proxy-URL, X-Proxy-Country, X-Proxy-Class, X-Proxy-Provider, X-Proxy-Session-ID, X-Request-ID, X-Tenant", MaxAge: 86400, } } @@ -118,6 +119,7 @@ func RequestLoggerMiddleware() fiber.Handler { if query := c.Query("text"); query != "" { logFields["query_hash"] = QueryHash(query) } + addProxyLogFields(c, logFields) entry := WithRequest(c.UserContext()).WithFields(logFields) if status >= 500 { @@ -132,24 +134,56 @@ func RequestLoggerMiddleware() fiber.Handler { } } +func addProxyLogFields(c *fiber.Ctx, fields logrus.Fields) { + if country := strings.ToLower(strings.TrimSpace(c.Get("X-Proxy-Country"))); country != "" { + fields["proxy_country"] = country + } + if class := strings.ToLower(strings.TrimSpace(c.Get("X-Proxy-Class"))); class != "" { + fields["proxy_class"] = class + } + if provider := strings.ToLower(strings.TrimSpace(c.Get("X-Proxy-Provider"))); provider != "" { + fields["proxy_provider"] = provider + } + + if sessionID := strings.TrimSpace(c.Get("X-Proxy-Session-ID")); sessionID != "" { + fields["proxy_session_id"] = sessionID + } + + if proxyURL := strings.TrimSpace(c.Get("X-Proxy-URL")); proxyURL != "" { + fields["proxy_used"] = MaskProxyURL(proxyURL) + } + + if laneKey := proxyLaneKeyFromContext(c.UserContext()); !laneKey.Empty() { + fields["lane_id"] = laneKey.ID() + } +} + func JSONErrorMiddleware() fiber.ErrorHandler { return func(c *fiber.Ctx, err error) error { code := fiber.StatusInternalServerError + errorCode := "" reason := "" + var meta map[string]interface{} if e, ok := err.(*fiber.Error); ok { code = e.Code } if apiErr, ok := err.(*APIError); ok { code = apiErr.HTTPStatus + errorCode = apiErr.ErrorCode reason = apiErr.Reason + meta = apiErr.Meta + } + if errorCode == "" { + errorCode = statusText(code) } resp := JSONErrorResponse{ - Error: statusText(code), + Error: errorCode, Code: code, Message: err.Error(), Reason: reason, + Meta: meta, } c.Set("Content-Type", "application/json") diff --git a/core/proxy.go b/core/proxy.go index 890dd26..3201bbe 100644 --- a/core/proxy.go +++ b/core/proxy.go @@ -18,6 +18,7 @@ const ( ProxyRuntimeRaw = "raw" ProxyModeOff = "off" ProxyModeTagPool = "tag_pool" + ProxyModeRequestURL = "request_url" DefaultProxyFailureThreshold = 3 ProxyOverrideDirect = "direct" // ProxyPoolQuarantineDuration is how long an exhausted tag pool stays quarantined @@ -49,9 +50,11 @@ type ProxiesHealthConfig struct { } type ProxiesConfig struct { - Global string `json:"global,omitempty" mapstructure:"global"` - Entries []ProxyEntryConfig `json:"entries" mapstructure:"entries"` - Health ProxiesHealthConfig `json:"health" mapstructure:"health"` + Global string `json:"global,omitempty" mapstructure:"global"` + Entries []ProxyEntryConfig `json:"entries" mapstructure:"entries"` + Health ProxiesHealthConfig `json:"health" mapstructure:"health"` + AllowRequestProxyURL bool `json:"allow_request_proxy_url" mapstructure:"allow_request_proxy_url"` + Lanes ProxyLanesConfig `json:"lanes" mapstructure:"lanes"` } type ProxyConfig struct { @@ -80,12 +83,15 @@ type ProxyEngineStats struct { } 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"` + ConfiguredCount int `json:"configured_count"` + HealthyCount int `json:"healthy_count"` + UnhealthyCount int `json:"unhealthy_count"` + RequestProxyURLEnabled bool `json:"request_proxy_url_enabled"` + Lanes LaneStats `json:"lanes"` + BrowserProcesses BrowserPoolStats `json:"browser_processes"` + Tags map[string]ProxyTagSummary `json:"tags"` + Entries []ProxyStatsEntry `json:"entries"` + Engines map[string]ProxyEngineStats `json:"engines,omitempty"` } type proxyState struct { @@ -110,6 +116,7 @@ func DefaultProxiesConfig() ProxiesConfig { Global: "", Entries: []ProxyEntryConfig{}, Health: ProxiesHealthConfig{FailureThreshold: DefaultProxyFailureThreshold}, + Lanes: DefaultProxyLanesConfig(), } } @@ -203,6 +210,7 @@ func NormalizeProxiesConfig(cfg ProxiesConfig) (ProxiesConfig, error) { cfg.Entries = normalizedEntries cfg.Health = ProxiesHealthConfig{FailureThreshold: failureThreshold} + cfg.Lanes = NormalizeProxyLanesConfig(cfg.Lanes) return cfg, nil } diff --git a/core/proxy_context.go b/core/proxy_context.go new file mode 100644 index 0000000..24b73a0 --- /dev/null +++ b/core/proxy_context.go @@ -0,0 +1,37 @@ +package core + +import ( + "context" + "strings" +) + +type proxyContextKey string + +const requestProxyURLContextKey proxyContextKey = "request_proxy_url" +const proxyLaneKeyContextKey proxyContextKey = "proxy_lane_key" + +func WithRequestProxyURL(ctx context.Context, proxyURL string) context.Context { + proxyURL = strings.TrimSpace(proxyURL) + if proxyURL == "" { + return EnsureContext(ctx) + } + return context.WithValue(EnsureContext(ctx), requestProxyURLContextKey, proxyURL) +} + +func requestProxyURLFromContext(ctx context.Context) string { + value, _ := EnsureContext(ctx).Value(requestProxyURLContextKey).(string) + return strings.TrimSpace(value) +} + +func WithProxyLaneKey(ctx context.Context, key ProxyLaneKey) context.Context { + key = NormalizeProxyLaneKey(key) + if key.Empty() { + return EnsureContext(ctx) + } + return context.WithValue(EnsureContext(ctx), proxyLaneKeyContextKey, key) +} + +func proxyLaneKeyFromContext(ctx context.Context) ProxyLaneKey { + value, _ := EnsureContext(ctx).Value(proxyLaneKeyContextKey).(ProxyLaneKey) + return NormalizeProxyLaneKey(value) +} diff --git a/core/proxy_lane.go b/core/proxy_lane.go new file mode 100644 index 0000000..f726378 --- /dev/null +++ b/core/proxy_lane.go @@ -0,0 +1,289 @@ +package core + +import ( + "crypto/sha256" + "encoding/hex" + "net/url" + "strings" + "sync" + "time" + + "github.com/go-rod/rod/lib/proto" + browserprofile "github.com/karust/openserp/core/browser" +) + +const DefaultProxyLaneMaxLanes = 100 + +type ProxyLanesConfig struct { + Enabled bool `json:"enabled" mapstructure:"enabled"` + MaxLanes int `json:"max_lanes" mapstructure:"max_lanes"` + DropCookiesOnChallenge bool `json:"drop_cookies_on_challenge" mapstructure:"drop_cookies_on_challenge"` +} + +type ProxyLaneKey struct { + Tenant string + Engine string + SessionID string +} + +type LaneStats struct { + Active int `json:"active"` + EvictedLRU int `json:"evicted_lru"` + CookiesDropped int `json:"cookies_dropped"` +} + +// BrowserPoolStats describes the live state of the per-process browser pool that +// keeps one Chrome per authenticated upstream proxy identity. Reported via +// /stats/proxy as `browser_processes`. +type BrowserPoolStats struct { + Active int `json:"active"` + Max int `json:"max"` + EvictedLRU int `json:"evicted_lru"` + EvictedIdle int `json:"evicted_idle"` +} + +type laneState struct { + Key ProxyLaneKey + Profile browserprofile.Profile + Cookies []*proto.NetworkCookie + LastUsedAt time.Time +} + +type LaneStore struct { + mu sync.Mutex + maxLanes int + lanes map[ProxyLaneKey]*laneState + evictedLRU int + cookiesDropped int +} + +func DefaultProxyLanesConfig() ProxyLanesConfig { + return ProxyLanesConfig{ + Enabled: true, + MaxLanes: DefaultProxyLaneMaxLanes, + DropCookiesOnChallenge: true, + } +} + +func NormalizeProxyLanesConfig(cfg ProxyLanesConfig) ProxyLanesConfig { + if cfg.MaxLanes <= 0 { + cfg.MaxLanes = DefaultProxyLaneMaxLanes + } + return cfg +} + +func NewLaneStore(maxLanes int) *LaneStore { + if maxLanes <= 0 { + maxLanes = DefaultProxyLaneMaxLanes + } + return &LaneStore{ + maxLanes: maxLanes, + lanes: map[ProxyLaneKey]*laneState{}, + } +} + +func (s *LaneStore) Profile(key ProxyLaneKey, create func() browserprofile.Profile) browserprofile.Profile { + if s == nil || key.Empty() { + if create == nil { + return browserprofile.Profile{} + } + return create() + } + + now := time.Now() + s.mu.Lock() + defer s.mu.Unlock() + + if state, ok := s.lanes[key]; ok { + state.LastUsedAt = now + return state.Profile + } + + profile := browserprofile.Profile{} + if create != nil { + profile = create() + } + s.lanes[key] = &laneState{Key: key, Profile: profile, LastUsedAt: now} + s.evictLRULocked() + return profile +} + +func (s *LaneStore) Cookies(key ProxyLaneKey) []*proto.NetworkCookie { + if s == nil || key.Empty() { + return nil + } + + s.mu.Lock() + defer s.mu.Unlock() + + state, ok := s.lanes[key] + if !ok { + return nil + } + state.LastUsedAt = time.Now() + return cloneCookies(state.Cookies) +} + +func (s *LaneStore) SaveCookies(key ProxyLaneKey, cookies []*proto.NetworkCookie) { + if s == nil || key.Empty() { + return + } + + s.mu.Lock() + defer s.mu.Unlock() + + state, ok := s.lanes[key] + if !ok { + state = &laneState{Key: key} + s.lanes[key] = state + } + state.Cookies = cloneCookies(cookies) + state.LastUsedAt = time.Now() + s.evictLRULocked() +} + +func (s *LaneStore) DropCookies(key ProxyLaneKey) { + if s == nil || key.Empty() { + return + } + + s.mu.Lock() + defer s.mu.Unlock() + + state, ok := s.lanes[key] + if !ok { + return + } + if len(state.Cookies) > 0 { + s.cookiesDropped++ + } + state.Cookies = nil + state.LastUsedAt = time.Now() +} + +func (s *LaneStore) Stats() LaneStats { + if s == nil { + return LaneStats{} + } + s.mu.Lock() + defer s.mu.Unlock() + return LaneStats{ + Active: len(s.lanes), + EvictedLRU: s.evictedLRU, + CookiesDropped: s.cookiesDropped, + } +} + +func (s *LaneStore) evictLRULocked() { + for len(s.lanes) > s.maxLanes { + var ( + oldestKey ProxyLaneKey + oldest time.Time + hasOldest bool + ) + for key, state := range s.lanes { + if !hasOldest || state.LastUsedAt.Before(oldest) { + oldestKey = key + oldest = state.LastUsedAt + hasOldest = true + } + } + if !hasOldest { + return + } + delete(s.lanes, oldestKey) + s.evictedLRU++ + } +} + +func (k ProxyLaneKey) Empty() bool { + return strings.TrimSpace(k.Engine) == "" || strings.TrimSpace(k.SessionID) == "" +} + +func (k ProxyLaneKey) ID() string { + k = NormalizeProxyLaneKey(k) + if k.Empty() { + return "" + } + if k.Tenant != "" { + return k.Tenant + ":" + k.Engine + ":" + k.SessionID + } + return k.Engine + ":" + k.SessionID +} + +func NormalizeProxyLaneKey(key ProxyLaneKey) ProxyLaneKey { + return ProxyLaneKey{ + Tenant: strings.TrimSpace(key.Tenant), + Engine: normalizeEngineName(key.Engine), + SessionID: strings.TrimSpace(key.SessionID), + } +} + +func ProxyLaneKeyForTenant(engine string, tenant string, q Query, proxyURL string) ProxyLaneKey { + sessionID := strings.TrimSpace(q.ProxySessionID) + if sessionID == "" { + sessionID = proxyLaneIDFromProxyURL(proxyURL) + } + return NormalizeProxyLaneKey(ProxyLaneKey{Tenant: tenant, Engine: engine, SessionID: sessionID}) +} + +func proxyLaneIDFromProxyURL(raw string) string { + normalized, err := NormalizeProxyURL(raw) + if err != nil || normalized == "" { + return "" + } + parsed, err := url.Parse(normalized) + if err != nil { + return "" + } + username := "" + if parsed.User != nil { + username = parsed.User.Username() + } + sum := sha256.Sum256([]byte(parsed.Host + "|" + username)) + return hex.EncodeToString(sum[:])[:16] +} + +func cloneCookies(cookies []*proto.NetworkCookie) []*proto.NetworkCookie { + if len(cookies) == 0 { + return nil + } + out := make([]*proto.NetworkCookie, 0, len(cookies)) + for _, cookie := range cookies { + if cookie == nil { + continue + } + cloned := *cookie + out = append(out, &cloned) + } + return out +} + +func cookieParams(cookies []*proto.NetworkCookie) []*proto.NetworkCookieParam { + if len(cookies) == 0 { + return nil + } + params := make([]*proto.NetworkCookieParam, 0, len(cookies)) + for _, cookie := range cookies { + if cookie == nil { + continue + } + sourcePort := cookie.SourcePort + params = append(params, &proto.NetworkCookieParam{ + Name: cookie.Name, + Value: cookie.Value, + Domain: cookie.Domain, + Path: cookie.Path, + Secure: cookie.Secure, + HTTPOnly: cookie.HTTPOnly, + SameSite: cookie.SameSite, + Expires: cookie.Expires, + Priority: cookie.Priority, + SameParty: cookie.SameParty, + SourceScheme: cookie.SourceScheme, + SourcePort: &sourcePort, + PartitionKey: cookie.PartitionKey, + }) + } + return params +} diff --git a/core/proxy_lane_test.go b/core/proxy_lane_test.go new file mode 100644 index 0000000..69e3b60 --- /dev/null +++ b/core/proxy_lane_test.go @@ -0,0 +1,101 @@ +package core + +import ( + "testing" + "time" + + "github.com/go-rod/rod/lib/proto" + browserprofile "github.com/karust/openserp/core/browser" +) + +func TestLaneStoreReusesCookiesBySession(t *testing.T) { + store := NewLaneStore(10) + key := ProxyLaneKey{Engine: "google", SessionID: "sid-a"} + cookies := []*proto.NetworkCookie{{Name: "sid", Value: "a", Domain: "example.com", Path: "/"}} + + store.SaveCookies(key, cookies) + got := store.Cookies(key) + if len(got) != 1 || got[0].Name != "sid" || got[0].Value != "a" { + t.Fatalf("expected saved cookie, got %#v", got) + } + + other := store.Cookies(ProxyLaneKey{Engine: "google", SessionID: "sid-b"}) + if len(other) != 0 { + t.Fatalf("expected different SID to be cookie-clean, got %#v", other) + } +} + +func TestLaneStoreDropCookiesPreservesProfile(t *testing.T) { + store := NewLaneStore(10) + key := ProxyLaneKey{Engine: "google", SessionID: "sid-a"} + profile := store.Profile(key, func() browserprofile.Profile { + return browserprofile.Profile{ID: "profile-a"} + }) + if profile.ID != "profile-a" { + t.Fatalf("expected initial profile, got %#v", profile) + } + + store.SaveCookies(key, []*proto.NetworkCookie{{Name: "sid", Value: "a", Domain: "example.com", Path: "/"}}) + store.DropCookies(key) + + if got := store.Cookies(key); len(got) != 0 { + t.Fatalf("expected cookies to be dropped, got %#v", got) + } + profile = store.Profile(key, func() browserprofile.Profile { + return browserprofile.Profile{ID: "profile-b"} + }) + if profile.ID != "profile-a" { + t.Fatalf("expected profile to be preserved after cookie drop, got %#v", profile) + } + if stats := store.Stats(); stats.CookiesDropped != 1 { + t.Fatalf("expected cookies_dropped=1, got %#v", stats) + } +} + +func TestLaneStoreEvictsLRU(t *testing.T) { + store := NewLaneStore(2) + keyA := ProxyLaneKey{Engine: "google", SessionID: "a"} + keyB := ProxyLaneKey{Engine: "google", SessionID: "b"} + keyC := ProxyLaneKey{Engine: "google", SessionID: "c"} + + store.SaveCookies(keyA, []*proto.NetworkCookie{{Name: "sid", Value: "a"}}) + time.Sleep(time.Millisecond) + store.SaveCookies(keyB, []*proto.NetworkCookie{{Name: "sid", Value: "b"}}) + time.Sleep(time.Millisecond) + _ = store.Cookies(keyB) + time.Sleep(time.Millisecond) + store.SaveCookies(keyC, []*proto.NetworkCookie{{Name: "sid", Value: "c"}}) + + if got := store.Cookies(keyA); len(got) != 0 { + t.Fatalf("expected oldest lane A to be evicted, got %#v", got) + } + if got := store.Cookies(keyB); len(got) != 1 { + t.Fatalf("expected lane B to remain, got %#v", got) + } + if stats := store.Stats(); stats.Active != 2 || stats.EvictedLRU != 1 { + t.Fatalf("unexpected lane stats: %#v", stats) + } +} + +func TestProxyLaneKeyForOmitsPassword(t *testing.T) { + a := ProxyLaneKeyForTenant("Google", "", Query{}, "http://user:pass-a@proxy.example:8080") + b := ProxyLaneKeyForTenant("google", "", Query{}, "http://user:pass-b@proxy.example:8080") + if a.Empty() || b.Empty() { + t.Fatalf("expected derived lane keys, got %#v %#v", a, b) + } + if a != b { + t.Fatalf("expected password changes not to affect lane key: %#v %#v", a, b) + } +} + +func TestProxyLaneKeyIncludesTenant(t *testing.T) { + q := Query{ProxySessionID: "sid-a"} + a := ProxyLaneKeyForTenant("google", "tenant-a", q, "http://proxy.example:8080") + b := ProxyLaneKeyForTenant("google", "tenant-b", q, "http://proxy.example:8080") + if a == b { + t.Fatalf("expected different tenants to produce different lane keys: %#v", a) + } + if got := a.ID(); got != "tenant-a:google:sid-a" { + t.Fatalf("unexpected tenant lane id: %q", got) + } +} diff --git a/core/proxy_per_context_spike_test.go b/core/proxy_per_context_spike_test.go new file mode 100644 index 0000000..5c635b5 --- /dev/null +++ b/core/proxy_per_context_spike_test.go @@ -0,0 +1,93 @@ +//go:build integration +// +build integration + +package core + +import ( + "context" + "encoding/base64" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/karust/openserp/testutil" +) + +func TestProxyPerContextAuthIsolationSpike(t *testing.T) { + testutil.RequireIntegration(t) + + type authHit struct { + proxy string + auth string + } + var ( + mu sync.Mutex + hits []authHit + ) + + newAuthProxy := func(name, username, password string) *httptest.Server { + want := "Basic " + base64.StdEncoding.EncodeToString([]byte(username+":"+password)) + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got := r.Header.Get("Proxy-Authorization") + mu.Lock() + hits = append(hits, authHit{proxy: name, auth: got}) + mu.Unlock() + if got != want { + w.Header().Set("Proxy-Authenticate", `Basic realm="openserp-spike"`) + w.WriteHeader(http.StatusProxyAuthRequired) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = fmt.Fprintf(w, "%s", name) + })) + } + + proxyA := newAuthProxy("proxy-a", "user-a", "pass-a") + defer proxyA.Close() + proxyB := newAuthProxy("proxy-b", "user-b", "pass-b") + defer proxyB.Close() + + browser, err := NewBrowser(BrowserOpts{IsHeadless: true, Timeout: 20 * time.Second}) + if err != nil { + t.Fatalf("create browser: %v", err) + } + defer closeTestBrowser(t, browser) + + run := func(proxyURL string) error { + ctx := WithRequestProxyURL(context.Background(), proxyURL) + page, err := browser.Navigate(ctx, "http://proxy-auth-spike.invalid/") + if err != nil { + return err + } + defer func() { + _ = ClosePageWithTimeout(context.Background(), page, time.Second) + }() + body, err := page.Timeout(5 * time.Second).Element("body") + if err != nil { + return err + } + _, err = body.Text() + return err + } + + errCh := make(chan error, 2) + go func() { errCh <- run(strings.Replace(proxyA.URL, "http://", "http://user-a:pass-a@", 1)) }() + go func() { errCh <- run(strings.Replace(proxyB.URL, "http://", "http://user-b:pass-b@", 1)) }() + + for i := 0; i < 2; i++ { + if err := <-errCh; err != nil { + t.Fatalf("proxied navigation failed: %v", err) + } + } + + mu.Lock() + defer mu.Unlock() + t.Logf("per-context auth spike report: %d proxy requests observed; Browser serializes authenticated proxy auth handlers to avoid cross-context credential leakage", len(hits)) + for _, hit := range hits { + t.Logf("proxy=%s auth_prefix=%t", hit.proxy, strings.HasPrefix(hit.auth, "Basic ")) + } +} diff --git a/core/proxy_test.go b/core/proxy_test.go index 0494b62..3b32f6c 100644 --- a/core/proxy_test.go +++ b/core/proxy_test.go @@ -10,6 +10,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "strings" "testing" "time" @@ -189,6 +190,24 @@ func TestMaskProxyURLRedactsCredentials(t *testing.T) { } } +func TestBrowserOptsLogFieldsRedactSecrets(t *testing.T) { + fields := browserOptsLogFields(BrowserOpts{ + ProxyURL: "http://user:sentinel-password@proxy.example:8080", + CaptchaSolverApiKey: "captcha-secret", + CaptchaSolverEnabled: true, + }) + rendered := fmt.Sprintf("%v", fields) + if strings.Contains(rendered, "sentinel-password") || strings.Contains(rendered, "captcha-secret") { + t.Fatalf("browser option log fields leaked secret: %s", rendered) + } + if fields["proxy"] != "http://proxy.example:8080" { + t.Fatalf("expected masked proxy field, got %#v", fields["proxy"]) + } + if fields["captcha_solver_has_key"] != true { + t.Fatalf("expected captcha key presence boolean") + } +} + func TestProxyURLForBrowserLaunchStripsCredentials(t *testing.T) { u, err := url.Parse("http://user:pass@127.0.0.1:18888") if err != nil { @@ -356,6 +375,18 @@ func TestClassifyProxyNetworkError(t *testing.T) { } } +func TestClassifyMainDocumentStatus(t *testing.T) { + if !errors.Is(classifyMainDocumentStatus(http.StatusForbidden), ErrBlocked) { + t.Fatal("expected 403 to classify as ErrBlocked") + } + if !errors.Is(classifyMainDocumentStatus(http.StatusTooManyRequests), ErrRateLimited) { + t.Fatal("expected 429 to classify as ErrRateLimited") + } + if classifyMainDocumentStatus(http.StatusOK) != nil { + t.Fatal("expected 200 to remain unclassified") + } +} + func TestNewRawHTTPClientClassifiesProxyAuthFailure(t *testing.T) { proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusProxyAuthRequired) diff --git a/core/resilient.go b/core/resilient.go index 58ed30e..88d8a7f 100644 --- a/core/resilient.go +++ b/core/resilient.go @@ -35,6 +35,18 @@ type ResilientConfig struct { Proxy ProxyConfig } +type proxyLaneCookieDropper interface { + DropProxyLaneCookies(context.Context, Query) +} + +type proxyLaneStatser interface { + ProxyLaneStats() LaneStats +} + +type browserPoolStatser interface { + BrowserPoolStats() BrowserPoolStats +} + func DefaultResilientConfig() ResilientConfig { return ResilientConfig{ Retry: DefaultRetryConfig(), @@ -174,6 +186,10 @@ func (rs *ResilientSearcher) searchWithProtection(ctx context.Context, engine Se case ProxyModeOff: attemptQuery.ProxyURL = "" attemptMeta.Used = "direct" + case ProxyModeRequestURL: + proxyURL = q.ProxyURL + attemptQuery.ProxyURL = proxyURL + attemptMeta.Used = MaskProxyURL(proxyURL) case ProxyModeTagPool: proxyURL = rs.selectProxyForQuery(policy, q, engineCtx) if proxyURL == "" { @@ -189,14 +205,24 @@ func (rs *ResilientSearcher) searchWithProtection(ctx context.Context, engine Se err error ) if isImage { - results, err = engine.SearchImage(callCtx, attemptQuery) + results, err = engine.SearchImage(proxyRequestContext(callCtx, engine.Name(), attemptQuery), attemptQuery) } else { - results, err = engine.Search(callCtx, attemptQuery) + results, err = engine.Search(proxyRequestContext(callCtx, engine.Name(), attemptQuery), attemptQuery) } if reportToRegistry { rs.reportProxyAttempt(engineCtx, proxyURL, err) } + if err != nil && errors.Is(err, ErrCaptcha) && rs.proxyCfg.Proxies.Lanes.DropCookiesOnChallenge { + // Recompute lane key only to gate the call: empty key means we have no + // session to drop cookies for. The dropper recomputes the key itself + // when it actually needs to mutate lane state. + if !ProxyLaneKeyForTenant(engine.Name(), TenantFromContext(callCtx), attemptQuery, attemptQuery.ProxyURL).Empty() { + if dropper, ok := engine.(proxyLaneCookieDropper); ok { + dropper.DropProxyLaneCookies(callCtx, attemptQuery) + } + } + } return results, err }) @@ -294,16 +320,21 @@ func (rs *ResilientSearcher) GetCircuitBreakerStats() []map[string]interface{} { func (rs *ResilientSearcher) GetProxyStats() ProxyStats { stats := ProxyStats{ - ConfiguredCount: 0, - HealthyCount: 0, - UnhealthyCount: 0, - Tags: map[string]ProxyTagSummary{}, - Entries: []ProxyStatsEntry{}, + ConfiguredCount: 0, + HealthyCount: 0, + UnhealthyCount: 0, + RequestProxyURLEnabled: rs.proxyCfg.Proxies.AllowRequestProxyURL, + Lanes: rs.proxyLaneStats(), + Tags: map[string]ProxyTagSummary{}, + Entries: []ProxyStatsEntry{}, } if rs.proxyRegistry != nil { stats = rs.proxyRegistry.BuildStats() } + stats.RequestProxyURLEnabled = rs.proxyCfg.Proxies.AllowRequestProxyURL + stats.Lanes = rs.proxyLaneStats() + stats.BrowserProcesses = rs.browserPoolStats() engines := map[string]ProxyEngineStats{} for _, engine := range rs.engines { @@ -332,6 +363,38 @@ func (rs *ResilientSearcher) GetProxyStats() ProxyStats { return stats } +func (rs *ResilientSearcher) proxyLaneStats() LaneStats { + var out LaneStats + for _, engine := range rs.engines { + statser, ok := engine.(proxyLaneStatser) + if !ok { + continue + } + stats := statser.ProxyLaneStats() + out.Active += stats.Active + out.EvictedLRU += stats.EvictedLRU + out.CookiesDropped += stats.CookiesDropped + } + return out +} + +// browserPoolStats reports the first non-zero browser pool stats found across +// engines. The pool is shared across engines, so reading from any engine that +// exposes it is sufficient; other engines' implementations return zero values. +func (rs *ResilientSearcher) browserPoolStats() BrowserPoolStats { + for _, engine := range rs.engines { + statser, ok := engine.(browserPoolStatser) + if !ok { + continue + } + stats := statser.BrowserPoolStats() + if stats.Max > 0 || stats.Active > 0 || stats.EvictedLRU > 0 || stats.EvictedIdle > 0 { + return stats + } + } + return BrowserPoolStats{} +} + func (rs *ResilientSearcher) ResolveMegaProxyMeta(q Query, engines []SearchEngine) ProxyExecutionMeta { if len(engines) == 0 { return ProxyExecutionMeta{Mode: ProxyModeOff, Used: "direct"} @@ -347,6 +410,9 @@ func (rs *ResilientSearcher) ResolveMegaProxyMeta(q Query, engines []SearchEngin hasOff = true continue } + if policy.Mode == ProxyModeRequestURL { + return ProxyExecutionMeta{Mode: ProxyModeRequestURL, Used: MaskProxyURL(q.ProxyURL)} + } allOff = false if policy.Tag != "" { @@ -406,10 +472,17 @@ func (rs *ResilientSearcher) effectivePolicyForEngine(engineName string) ProxyPo func (rs *ResilientSearcher) effectivePolicyForQuery(engineName string, q Query) ProxyPolicy { switch q.ProxyOverride { - case "": - return rs.effectivePolicyForEngine(engineName) case ProxyOverrideDirect: return ProxyPolicy{Mode: ProxyModeOff} + } + + if strings.TrimSpace(q.ProxyURL) != "" && rs.proxyCfg.Proxies.AllowRequestProxyURL { + return ProxyPolicy{Mode: ProxyModeRequestURL} + } + + switch q.ProxyOverride { + case "": + return rs.effectivePolicyForEngine(engineName) default: return ProxyPolicy{Mode: ProxyModeTagPool, Tag: q.ProxyOverride} } @@ -451,4 +524,18 @@ func (rs *ResilientSearcher) selectProxyForQuery(policy ProxyPolicy, q Query, ct return rs.selectProxyForTag(ctx, policy.Tag) } +func proxyRequestContext(ctx context.Context, engineName string, q Query) context.Context { + ctx = WithRequestProxyURL(ctx, q.ProxyURL) + if q.ProxyURL == "" { + return ctx + } + if laneKey := ProxyLaneKeyForTenant(engineName, TenantFromContext(ctx), q, q.ProxyURL); !laneKey.Empty() { + ctx = WithProxyLaneKey(ctx, laneKey) + if q.ProxyCountry != "" { + ctx = WithProfileRegion(ctx, q.ProxyCountry) + } + } + return ctx +} + var ErrAllEnginesFailed = fmt.Errorf("all search engines failed") diff --git a/core/retry.go b/core/retry.go index 0839338..2ae4543 100644 --- a/core/retry.go +++ b/core/retry.go @@ -36,7 +36,7 @@ type RetryResult struct { } // RetryableSearch executes searchFn with exponential backoff retries. -// CAPTCHA, parser, engine-internal, and proxy-unavailable errors are not retried. +// CAPTCHA, block, rate-limit, parser, engine-internal, and proxy-unavailable errors are not retried. func RetryableSearch(ctx context.Context, cfg RetryConfig, engineName string, searchFn func(context.Context) ([]SearchResult, error)) RetryResult { ctx = WithEngine(EnsureContext(ctx), engineName) logger := WithRequest(ctx) @@ -79,40 +79,8 @@ func RetryableSearch(ctx context.Context, cfg RetryConfig, engineName string, se } lastErr = err - if errors.Is(err, ErrCaptcha) { - logger.Warn("CAPTCHA detected, skipping retries") - return RetryResult{ - Err: err, - Attempts: attempt + 1, - Engine: engineName, - } - } - if errors.Is(err, ErrProxyUnavailable) { - logger.Warn("Proxy unavailable, skipping retries") - return RetryResult{ - Err: err, - Attempts: attempt + 1, - Engine: engineName, - } - } - if errors.Is(err, ErrParser) { - logger.Warn("Parser failure, skipping retries") - return RetryResult{ - Err: err, - Attempts: attempt + 1, - Engine: engineName, - } - } - if errors.Is(err, ErrEngineInternal) { - logger.Warn("Engine panic recovered, skipping retries") - return RetryResult{ - Err: err, - Attempts: attempt + 1, - Engine: engineName, - } - } - if IsContextDone(err) { - logger.Warn("Context canceled/deadline exceeded, skipping retries") + if reason, skip := nonRetryableReason(err); skip { + logger.Warnf("%s, skipping retries", reason) return RetryResult{ Err: err, Attempts: attempt + 1, @@ -130,6 +98,30 @@ func RetryableSearch(ctx context.Context, cfg RetryConfig, engineName string, se } } +var nonRetryableSentinels = []struct { + err error + reason string +}{ + {ErrCaptcha, "CAPTCHA detected"}, + {ErrBlocked, "Blocked response detected"}, + {ErrRateLimited, "Rate limited response detected"}, + {ErrProxyUnavailable, "Proxy unavailable"}, + {ErrParser, "Parser failure"}, + {ErrEngineInternal, "Engine panic recovered"}, +} + +func nonRetryableReason(err error) (string, bool) { + for _, s := range nonRetryableSentinels { + if errors.Is(err, s.err) { + return s.reason, true + } + } + if IsContextDone(err) { + return "Context canceled/deadline exceeded", true + } + return "", false +} + func calculateBackoff(cfg RetryConfig, attempt int) time.Duration { backoff := float64(cfg.InitialBackoff) * math.Pow(cfg.BackoffFactor, float64(attempt-1)) if backoff > float64(cfg.MaxBackoff) { diff --git a/core/server.go b/core/server.go index d77071a..539ece6 100644 --- a/core/server.go +++ b/core/server.go @@ -187,6 +187,10 @@ func (s *Server) handleDedicatedEndpoint(c *fiber.Ctx, engine SearchEngine, isIm WithRequest(c.UserContext()).WithError(err).Warn("Invalid query parameters") return err } + if err := s.validateRequestProxyURL(&q); err != nil { + WithRequest(c.UserContext()).WithError(err).Warn("Invalid request proxy URL") + return err + } format, err := resolveFormat(c) if err != nil { @@ -206,7 +210,7 @@ func (s *Server) handleDedicatedEndpoint(c *fiber.Ctx, engine SearchEngine, isIm WithField("action", action). Debugf("Starting %s request for query: %s", action, q.Text) - if format == "json" { + if format == "json" && !ShouldBypassCacheForProxyMarket(q) { if hit, err := s.tryServeCacheHit( c, startedAt, @@ -236,7 +240,7 @@ func (s *Server) handleDedicatedEndpoint(c *fiber.Ctx, engine SearchEngine, isIm s.applyProxyHeaders(c, proxyMeta) if searchErr != nil { WithRequest(requestCtx).WithFields(logrus.Fields{"action": action}).WithError(searchErr).Error("Search failed") - return fiber.NewError(fiber.StatusServiceUnavailable, classifySearchError(searchErr).Error()) + return searchAPIError(searchErr, usedEngine, q, proxyMeta) } env := NewImageEnvelope(q, requestID, startedAt, engineNames) @@ -277,7 +281,7 @@ func (s *Server) handleDedicatedEndpoint(c *fiber.Ctx, engine SearchEngine, isIm if searchErr != nil { WithRequest(requestCtx).WithFields(logrus.Fields{"action": action}).WithError(searchErr).Error("Search failed") - return fiber.NewError(fiber.StatusServiceUnavailable, classifySearchError(searchErr).Error()) + return searchAPIError(searchErr, usedEngine, q, proxyMeta) } env := NewEnvelope(q, requestID, startedAt, engineNames) @@ -308,21 +312,73 @@ func (s *Server) handleDedicatedEndpoint(c *fiber.Ctx, engine SearchEngine, isIm return sendEnvelope(c, format, env) } -// classifySearchError maps internal sentinel errors to user-facing messages. -func classifySearchError(err error) error { +type searchErrorSpec struct { + status int + code string + message string +} + +func mapSearchError(err error) searchErrorSpec { switch { case errors.Is(err, ErrCaptcha): - return fmt.Errorf("captcha found, please stop sending requests for a while: %w", err) + return searchErrorSpec{status: fiber.StatusTooManyRequests, code: "captcha_detected", message: "captcha detected"} + case errors.Is(err, ErrBlocked): + return searchErrorSpec{status: fiber.StatusForbidden, code: "blocked", message: "search engine blocked the request"} + case errors.Is(err, ErrRateLimited): + return searchErrorSpec{status: fiber.StatusTooManyRequests, code: "rate_limited", message: "search engine rate limited the request"} case errors.Is(err, ErrSearchTimeout): - return fmt.Errorf("%s", err) - case errors.Is(err, ErrParser): - return fmt.Errorf("%w", ErrParser) - case errors.Is(err, ErrEngineInternal): - return fmt.Errorf("%w", ErrEngineInternal) + return searchErrorSpec{status: fiber.StatusGatewayTimeout, code: "search_timeout", message: ErrSearchTimeout.Error()} + case errors.Is(err, ErrProxyAuth): + return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "proxy_auth", message: "proxy authentication failed"} + case errors.Is(err, ErrProxyConnect): + return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "proxy_connect", message: "proxy connection failed"} + case errors.Is(err, ErrTimeout): + return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "proxy_timeout", message: "proxy request timed out"} case errors.Is(err, ErrProxyUnavailable): - return fmt.Errorf("%s", err) + return searchErrorSpec{status: fiber.StatusServiceUnavailable, code: "proxy_unavailable", message: "proxy unavailable"} + case errors.Is(err, ErrParser): + return searchErrorSpec{status: fiber.StatusBadGateway, code: "parser_failure", message: "parser failure"} + case errors.Is(err, ErrEngineInternal): + return searchErrorSpec{status: fiber.StatusBadGateway, code: "engine_internal", message: "engine internal error"} } - return err + return searchErrorSpec{status: fiber.StatusBadGateway, code: "engine_internal", message: err.Error()} +} + +func searchAPIError(err error, engineName string, q Query, proxyMeta ProxyExecutionMeta) error { + spec := mapSearchError(err) + return &APIError{ + HTTPStatus: spec.status, + ErrorCode: spec.code, + Message: spec.message, + Meta: searchErrorMeta(engineName, q, proxyMeta), + } +} + +func searchErrorMeta(engineName string, q Query, proxyMeta ProxyExecutionMeta) map[string]interface{} { + meta := map[string]interface{}{} + if strings.TrimSpace(engineName) != "" { + meta["engine"] = engineName + } + proxyUsed := strings.TrimSpace(proxyMeta.Used) + if proxyUsed == "" && strings.TrimSpace(q.ProxyURL) != "" { + proxyUsed = MaskProxyURL(q.ProxyURL) + } + if proxyUsed != "" { + meta["proxy_used"] = proxyUsed + } + if q.ProxyCountry != "" { + meta["proxy_country"] = q.ProxyCountry + } + if q.ProxyClass != "" { + meta["proxy_class"] = q.ProxyClass + } + if q.ProxyProvider != "" { + meta["proxy_provider"] = q.ProxyProvider + } + if q.ProxySessionID != "" { + meta["proxy_session_id"] = q.ProxySessionID + } + return meta } // cacheEnvelopeIfEligible stores the envelope JSON and returns the cache status header value. @@ -330,6 +386,10 @@ func (s *Server) cacheEnvelopeIfEligible(engineName, usedEngine, action string, if s.cache == nil { return "" } + if ShouldBypassCacheForProxyMarket(q) { + s.cache.RecordBypass() + return "BYPASS" + } // Don't cache fallback responses so the primary engine can recover. if usedEngine != engineName { s.cache.RecordBypass() @@ -562,6 +622,30 @@ func (s *Server) parseFingerprintCheckRequest(c *fiber.Ctx) (fingerprintCheckReq browserOpts.ProxyURL = strings.TrimSpace(c.Query("proxy", browserOpts.ProxyURL)) browserOpts.LanguageCode = strings.TrimSpace(c.Query("language", browserOpts.LanguageCode)) + if headerProxyURL := strings.TrimSpace(c.Get("X-Proxy-URL")); headerProxyURL != "" { + if !s.opts.Resilience.Proxy.Proxies.AllowRequestProxyURL { + return fingerprintCheckRequest{}, &APIError{ + HTTPStatus: fiber.StatusBadRequest, + ErrorCode: "bad_request", + Reason: ReasonRequestProxyURLDisabled, + Message: "X-Proxy-URL is disabled by server configuration", + } + } + normalized, err := NormalizeProxyURL(headerProxyURL) + if err != nil { + return fingerprintCheckRequest{}, errInvalidParam(fmt.Sprintf("X-Proxy-URL: %v", err)) + } + if IsAuthenticatedSocksProxyURL(normalized) { + return fingerprintCheckRequest{}, &APIError{ + HTTPStatus: fiber.StatusBadRequest, + ErrorCode: "bad_request", + Reason: ReasonUnsupportedProxyScheme, + Message: "authenticated SOCKS proxies are not supported in browser mode", + } + } + browserOpts.ProxyURL = normalized + } + insecureDefault := browserOpts.Insecure if detectors.IsCustom(detectorName) { insecureDefault = true @@ -654,6 +738,10 @@ func (s *Server) handleMegaEndpoint(c *fiber.Ctx, action string, run func(contex WithRequest(c.UserContext()).WithError(err).Warn("Invalid query parameters") return err } + if err := s.validateRequestProxyURL(&q); err != nil { + WithRequest(c.UserContext()).WithError(err).Warn("Invalid request proxy URL") + return err + } format, err := resolveFormat(c) if err != nil { @@ -694,7 +782,7 @@ func (s *Server) handleMegaEndpoint(c *fiber.Ctx, action string, run func(contex logMessage: fmt.Sprintf("Cache hit for mega %s partial set: engines=%s query=%s", action, engineNamesJoined, q.Text), }) } - if format == "json" { + if format == "json" && !ShouldBypassCacheForProxyMarket(q) { if hit, err := s.tryServeCacheHit(c, startedAt, cacheHitCandidates...); hit || err != nil { return err } @@ -916,6 +1004,10 @@ func (s *Server) cacheMegaEnvelopeResults(action string, enginesToUse []SearchEn if s.cache == nil { return "" } + if ShouldBypassCacheForProxyMarket(q) { + s.cache.RecordBypass() + return "BYPASS" + } if len(env.Results) == 0 { s.cache.RecordBypass() return "BYPASS" @@ -935,6 +1027,10 @@ func (s *Server) cacheMegaImageResults(action string, enginesToUse []SearchEngin if s.cache == nil { return "" } + if ShouldBypassCacheForProxyMarket(q) { + s.cache.RecordBypass() + return "BYPASS" + } if len(env.Results) == 0 { s.cache.RecordBypass() return "BYPASS" @@ -1014,10 +1110,37 @@ func (s *Server) applyProxyHeaders(c *fiber.Ctx, meta ProxyExecutionMeta) { } c.Set("X-Proxy-Mode", mode) - c.Set("X-Proxy-Tag", tag) + if tag != "" { + c.Set("X-Proxy-Tag", tag) + } c.Set("X-Proxy-Used", used) } +func (s *Server) validateRequestProxyURL(q *Query) error { + if q == nil || strings.TrimSpace(q.ProxyURL) == "" || q.ProxyOverride == ProxyOverrideDirect { + return nil + } + + if !s.opts.Resilience.Proxy.Proxies.AllowRequestProxyURL { + return &APIError{ + HTTPStatus: fiber.StatusBadRequest, + ErrorCode: "bad_request", + Reason: ReasonRequestProxyURLDisabled, + Message: "X-Proxy-URL is disabled by server configuration", + } + } + + if s.opts.Resilience.Proxy.Runtime == ProxyRuntimeBrowser && IsAuthenticatedSocksProxyURL(q.ProxyURL) { + return &APIError{ + HTTPStatus: fiber.StatusBadRequest, + ErrorCode: "bad_request", + Reason: ReasonUnsupportedProxyScheme, + Message: "authenticated SOCKS proxies are not supported in browser mode", + } + } + return nil +} + func (s *Server) handleOpenAPISpec(c *fiber.Ctx) error { c.Set("Content-Type", "application/yaml; charset=utf-8") return c.Send(apidocs.OpenAPIYAML) diff --git a/core/server_test.go b/core/server_test.go index 537fe04..eb1266e 100644 --- a/core/server_test.go +++ b/core/server_test.go @@ -24,9 +24,10 @@ type engineMock struct { searchFn func(context.Context, Query) ([]SearchResult, error) imageFn func(context.Context, Query) ([]SearchResult, error) - mu sync.Mutex - searchCalls int - imageCalls int + mu sync.Mutex + searchCalls int + imageCalls int + droppedLaneQueries []Query } func (e *engineMock) Name() string { return e.name } @@ -55,6 +56,12 @@ func (e *engineMock) SearchImage(ctx context.Context, q Query) ([]SearchResult, return []SearchResult{{Rank: 1, URL: "https://img.example.com/" + e.name, Title: e.name}}, nil } +func (e *engineMock) DropProxyLaneCookies(_ context.Context, q Query) { + e.mu.Lock() + defer e.mu.Unlock() + e.droppedLaneQueries = append(e.droppedLaneQueries, q) +} + func request(t *testing.T, s *Server, path string) *http.Response { t.Helper() req := httptest.NewRequest(http.MethodGet, path, nil) @@ -547,8 +554,8 @@ func TestDedicatedEndpointNoFallbackByDefault(t *testing.T) { srv := NewServerWithOptions("127.0.0.1", 7072, opts, primary, fallback) resp := request(t, srv, "/google/search?text=golang") - if resp.StatusCode != http.StatusServiceUnavailable { - t.Fatalf("expected 503 when primary fails and fallback disabled, got %d", resp.StatusCode) + if resp.StatusCode != http.StatusBadGateway { + t.Fatalf("expected 502 when primary fails and fallback disabled, got %d", resp.StatusCode) } if got := resp.Header.Get("X-Fallback-Engine"); got != "" { t.Fatalf("unexpected fallback header: %s", got) @@ -618,6 +625,70 @@ func TestDedicatedEndpointCachesImageResults(t *testing.T) { } } +func TestProxiedRequestWithoutMarketMetadataBypassesCache(t *testing.T) { + engine := &engineMock{name: "google", initialized: true} + opts := DefaultServerOptions() + opts.Resilience.Retry.MaxRetries = 0 + opts.CacheTTL = time.Minute + opts.CacheMaxSize = 10 + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, + }, + } + srv := NewServerWithOptions("127.0.0.1", 7124, opts, engine) + + for i := 0; i < 2; i++ { + req := httptest.NewRequest(http.MethodGet, "/google/search?text=golang", nil) + req.Header.Set("X-Proxy-URL", "http://proxy.example:8080") + resp, err := srv.app.Test(req, -1) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected request to succeed, got %d", resp.StatusCode) + } + if got := resp.Header.Get("X-Cache"); got != "BYPASS" { + t.Fatalf("expected X-Cache=BYPASS, got %q", got) + } + } + if engine.searchCalls != 2 { + t.Fatalf("expected cache bypass to execute both searches, got %d calls", engine.searchCalls) + } +} + +func TestProxiedRequestWithMarketMetadataUsesCache(t *testing.T) { + engine := &engineMock{name: "google", initialized: true} + opts := DefaultServerOptions() + opts.Resilience.Retry.MaxRetries = 0 + opts.CacheTTL = time.Minute + opts.CacheMaxSize = 10 + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, + }, + } + srv := NewServerWithOptions("127.0.0.1", 7125, opts, engine) + + for i := 0; i < 2; i++ { + req := httptest.NewRequest(http.MethodGet, "/google/search?text=golang", nil) + req.Header.Set("X-Proxy-URL", "http://proxy.example:8080") + req.Header.Set("X-Proxy-Country", "us") + resp, err := srv.app.Test(req, -1) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected request to succeed, got %d", resp.StatusCode) + } + } + if engine.searchCalls != 1 { + t.Fatalf("expected second market-scoped request to hit cache, got %d calls", engine.searchCalls) + } +} + func TestDedicatedEndpointFallbackBypassesCache(t *testing.T) { primary := &engineMock{ name: "google", @@ -908,6 +979,7 @@ func TestStatsProxyV2Payload(t *testing.T) { opts.Resilience.Proxy = ProxyConfig{ Runtime: ProxyRuntimeRaw, Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, Entries: []ProxyEntryConfig{ {URL: "http://user:pass@proxy1:8080", Tags: []string{"default", "us"}}, {URL: "http://proxy2:8080", Tags: []string{"default"}}, @@ -937,6 +1009,18 @@ func TestStatsProxyV2Payload(t *testing.T) { if got := payload["unhealthy_count"].(float64); got != 0 { t.Fatalf("expected unhealthy_count=0, got %v", got) } + if got := payload["request_proxy_url_enabled"].(bool); !got { + t.Fatalf("expected request_proxy_url_enabled=true") + } + lanes, ok := payload["lanes"].(map[string]interface{}) + if !ok { + t.Fatalf("expected lanes object, got %T", payload["lanes"]) + } + for _, field := range []string{"active", "evicted_lru", "cookies_dropped"} { + if _, ok := lanes[field].(float64); !ok { + t.Fatalf("expected lanes.%s number, got %T", field, lanes[field]) + } + } if _, exists := payload["defaults"]; exists { t.Fatalf("defaults must not be exposed in proxy stats payload") @@ -1150,6 +1234,156 @@ func TestRequestProxyOverrideDirectBeatsGlobal(t *testing.T) { } } +func TestRequestProxyURLDisabledByDefault(t *testing.T) { + engine := &engineMock{name: "google", initialized: true} + srv := NewServerWithOptions("127.0.0.1", 7117, DefaultServerOptions(), engine) + + resp := requestWithHeader(t, srv, "/google/search?text=golang", "X-Proxy-URL", "http://user:pass@proxy.example:8080") + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("expected disabled request proxy URL to return 400, got %d", resp.StatusCode) + } + + var payload JSONErrorResponse + if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil { + t.Fatalf("decode error response: %v", err) + } + if payload.Error != "bad_request" { + t.Fatalf("expected bad_request error, got %q", payload.Error) + } + if payload.Reason != ReasonRequestProxyURLDisabled { + t.Fatalf("expected reason=%q, got %q", ReasonRequestProxyURLDisabled, payload.Reason) + } +} + +func TestRequestProxyURLHonoredWhenEnabled(t *testing.T) { + var got Query + engine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(_ context.Context, q Query) ([]SearchResult, error) { + got = q + return []SearchResult{{Rank: 1, URL: "https://example.com/google", Title: "google"}}, nil + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, + }, + } + srv := NewServerWithOptions("127.0.0.1", 7118, opts, engine) + + req := httptest.NewRequest(http.MethodGet, "/google/search?text=golang", nil) + req.Header.Set("X-Proxy-URL", "http://user:pass@proxy.example:8080") + req.Header.Set("X-Proxy-Country", " US ") + req.Header.Set("X-Proxy-Class", " Residential ") + req.Header.Set("X-Proxy-Provider", " WebShare ") + req.Header.Set("X-Proxy-Session-ID", "SID-1") + resp, err := srv.app.Test(req, -1) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected request proxy URL request to succeed, got %d", resp.StatusCode) + } + if got.ProxyURL != "http://user:pass@proxy.example:8080" { + t.Fatalf("expected raw proxy URL on query, got %q", got.ProxyURL) + } + if got.ProxyCountry != "us" || got.ProxyClass != "residential" || got.ProxyProvider != "webshare" || got.ProxySessionID != "SID-1" { + t.Fatalf("unexpected normalized proxy metadata: %#v", got) + } + if header := resp.Header.Get("X-Proxy-Mode"); header != ProxyModeRequestURL { + t.Fatalf("expected X-Proxy-Mode=%s, got %q", ProxyModeRequestURL, header) + } + if header := resp.Header.Get("X-Proxy-Tag"); header != "" { + t.Fatalf("expected empty X-Proxy-Tag, got %q", header) + } + if header := resp.Header.Get("X-Proxy-Used"); header != "http://proxy.example:8080" { + t.Fatalf("expected masked X-Proxy-Used, got %q", header) + } +} + +func TestRequestProxyURLBeatsTagOverride(t *testing.T) { + var googleProxy string + engine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(_ context.Context, q Query) ([]SearchResult, error) { + googleProxy = q.ProxyURL + return []SearchResult{{Rank: 1, URL: "https://example.com/google", Title: "google"}}, nil + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, + Entries: []ProxyEntryConfig{ + {URL: "http://tag-proxy:8080", Tags: []string{"us"}}, + }, + }, + } + srv := NewServerWithOptions("127.0.0.1", 7123, opts, engine) + + req := httptest.NewRequest(http.MethodGet, "/google/search?text=golang", nil) + req.Header.Set("X-Use-Proxy", "us") + req.Header.Set("X-Proxy-URL", "http://request-proxy:8080") + resp, err := srv.app.Test(req, -1) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected request to succeed, got %d", resp.StatusCode) + } + if googleProxy != "http://request-proxy:8080" { + t.Fatalf("expected request proxy URL to beat tag override, got %q", googleProxy) + } + if got := resp.Header.Get("X-Proxy-Mode"); got != ProxyModeRequestURL { + t.Fatalf("expected X-Proxy-Mode=%s, got %q", ProxyModeRequestURL, got) + } +} + +func TestRequestProxyOverrideDirectIgnoresRequestProxyURL(t *testing.T) { + var googleProxy string + engine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(_ context.Context, q Query) ([]SearchResult, error) { + googleProxy = q.ProxyURL + return []SearchResult{{Rank: 1, URL: "https://example.com/google", Title: "google"}}, nil + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, + }, + } + srv := NewServerWithOptions("127.0.0.1", 7119, opts, engine) + + req := httptest.NewRequest(http.MethodGet, "/google/search?text=golang", nil) + req.Header.Set("X-Use-Proxy", "direct") + req.Header.Set("X-Proxy-URL", "http://user:pass@proxy.example:8080") + resp, err := srv.app.Test(req, -1) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected direct override to succeed, got %d", resp.StatusCode) + } + if googleProxy != "" { + t.Fatalf("expected direct override to clear proxy, got %q", googleProxy) + } + if header := resp.Header.Get("X-Proxy-Mode"); header != ProxyModeOff { + t.Fatalf("expected X-Proxy-Mode=%s, got %q", ProxyModeOff, header) + } +} + func TestRequestProxyOverrideTagBeatsGlobal(t *testing.T) { var googleProxy string engine := &engineMock{ @@ -1188,6 +1422,164 @@ func TestRequestProxyOverrideTagBeatsGlobal(t *testing.T) { } } +func TestCaptchaDropsProxyLaneCookies(t *testing.T) { + engine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(_ context.Context, _ Query) ([]SearchResult, error) { + return nil, ErrCaptcha + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Retry.MaxRetries = 0 + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, + Lanes: DefaultProxyLanesConfig(), + }, + } + srv := NewServerWithOptions("127.0.0.1", 7120, opts, engine) + + req := httptest.NewRequest(http.MethodGet, "/google/search?text=golang", nil) + req.Header.Set("X-Proxy-URL", "http://user:pass@proxy.example:8080") + req.Header.Set("X-Proxy-Session-ID", "sid-a") + resp, err := srv.app.Test(req, -1) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if resp.StatusCode != http.StatusTooManyRequests { + t.Fatalf("expected captcha failure to return 429, got %d", resp.StatusCode) + } + + engine.mu.Lock() + defer engine.mu.Unlock() + if len(engine.droppedLaneQueries) != 1 { + t.Fatalf("expected one cookie-drop hook call, got %d", len(engine.droppedLaneQueries)) + } + if got := engine.droppedLaneQueries[0].ProxySessionID; got != "sid-a" { + t.Fatalf("expected drop hook to receive session id sid-a, got %q", got) + } +} + +func TestProxyErrorDoesNotDropProxyLaneCookies(t *testing.T) { + engine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(_ context.Context, _ Query) ([]SearchResult, error) { + return nil, ErrProxyConnect + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Retry.MaxRetries = 0 + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, + Lanes: DefaultProxyLanesConfig(), + }, + } + srv := NewServerWithOptions("127.0.0.1", 7121, opts, engine) + + req := httptest.NewRequest(http.MethodGet, "/google/search?text=golang", nil) + req.Header.Set("X-Proxy-URL", "http://user:pass@proxy.example:8080") + req.Header.Set("X-Proxy-Session-ID", "sid-a") + resp, err := srv.app.Test(req, -1) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if resp.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("expected proxy failure before P5 status mapping, got %d", resp.StatusCode) + } + + engine.mu.Lock() + defer engine.mu.Unlock() + if len(engine.droppedLaneQueries) != 0 { + t.Fatalf("expected no cookie-drop hook for proxy error, got %d", len(engine.droppedLaneQueries)) + } +} + +func TestStableSearchErrorJSONWithProxyMeta(t *testing.T) { + tests := []struct { + name string + err error + wantStatus int + wantError string + }{ + {name: "captcha", err: ErrCaptcha, wantStatus: http.StatusTooManyRequests, wantError: "captcha_detected"}, + {name: "blocked", err: ErrBlocked, wantStatus: http.StatusForbidden, wantError: "blocked"}, + {name: "rate limited", err: ErrRateLimited, wantStatus: http.StatusTooManyRequests, wantError: "rate_limited"}, + {name: "search timeout", err: ErrSearchTimeout, wantStatus: http.StatusGatewayTimeout, wantError: "search_timeout"}, + {name: "proxy connect", err: ErrProxyConnect, wantStatus: http.StatusServiceUnavailable, wantError: "proxy_connect"}, + {name: "proxy auth", err: ErrProxyAuth, wantStatus: http.StatusServiceUnavailable, wantError: "proxy_auth"}, + {name: "proxy timeout", err: ErrTimeout, wantStatus: http.StatusServiceUnavailable, wantError: "proxy_timeout"}, + {name: "proxy unavailable", err: ErrProxyUnavailable, wantStatus: http.StatusServiceUnavailable, wantError: "proxy_unavailable"}, + {name: "parser", err: ErrParser, wantStatus: http.StatusBadGateway, wantError: "parser_failure"}, + {name: "engine internal", err: ErrEngineInternal, wantStatus: http.StatusBadGateway, wantError: "engine_internal"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + engine := &engineMock{ + name: "google", + initialized: true, + searchFn: func(_ context.Context, _ Query) ([]SearchResult, error) { + return nil, tt.err + }, + } + + opts := DefaultServerOptions() + opts.Resilience.Retry.MaxRetries = 0 + opts.Resilience.Proxy = ProxyConfig{ + Runtime: ProxyRuntimeRaw, + Proxies: ProxiesConfig{ + AllowRequestProxyURL: true, + }, + } + srv := NewServerWithOptions("127.0.0.1", 7122, opts, engine) + + req := httptest.NewRequest(http.MethodGet, "/google/search?text=golang", nil) + req.Header.Set("X-Proxy-URL", "http://user:sentinel-password@proxy.example:8080") + req.Header.Set("X-Proxy-Country", "US") + req.Header.Set("X-Proxy-Class", "Residential") + req.Header.Set("X-Proxy-Provider", "WebShare") + req.Header.Set("X-Proxy-Session-ID", "sid-a") + resp, err := srv.app.Test(req, -1) + if err != nil { + t.Fatalf("request failed: %v", err) + } + if resp.StatusCode != tt.wantStatus { + t.Fatalf("expected status %d, got %d", tt.wantStatus, resp.StatusCode) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read response body: %v", err) + } + if strings.Contains(string(body), "sentinel-password") { + t.Fatalf("response leaked proxy password: %s", string(body)) + } + + var payload JSONErrorResponse + if err := json.Unmarshal(body, &payload); err != nil { + t.Fatalf("decode error response: %v", err) + } + if payload.Error != tt.wantError { + t.Fatalf("expected error=%q, got %q", tt.wantError, payload.Error) + } + if payload.Meta["proxy_used"] != "http://proxy.example:8080" { + t.Fatalf("expected masked proxy_used, got %#v", payload.Meta["proxy_used"]) + } + if payload.Meta["proxy_country"] != "us" || payload.Meta["proxy_class"] != "residential" || + payload.Meta["proxy_provider"] != "webshare" || payload.Meta["proxy_session_id"] != "sid-a" { + t.Fatalf("unexpected proxy meta: %#v", payload.Meta) + } + }) + } +} + func TestBrowserProxyPoolRotatesPerRequest(t *testing.T) { var attemptedProxies []string engine := &engineMock{ @@ -1374,7 +1766,7 @@ func TestRetryAppliesRateLimiterOnEachAttempt(t *testing.T) { start := time.Now() resp := request(t, srv, "/google/search?text=golang") elapsed := time.Since(start) - if resp.StatusCode != http.StatusServiceUnavailable { + if resp.StatusCode != http.StatusBadGateway { t.Fatalf("expected failure response, got %d", resp.StatusCode) } diff --git a/docs/openapi.yaml b/docs/openapi.yaml index b14a47a..fa01fc2 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -48,6 +48,12 @@ paths: - $ref: "#/components/parameters/AnswersQuery" - $ref: "#/components/parameters/FormatQuery" - $ref: "#/components/parameters/UseProxyHeader" + - $ref: "#/components/parameters/ProxyURLHeader" + - $ref: "#/components/parameters/ProxyCountryHeader" + - $ref: "#/components/parameters/ProxyClassHeader" + - $ref: "#/components/parameters/ProxyProviderHeader" + - $ref: "#/components/parameters/ProxySessionIDHeader" + - $ref: "#/components/parameters/TenantHeader" responses: "200": description: Search results envelope @@ -125,8 +131,16 @@ paths: type: string "400": $ref: "#/components/responses/BadRequestError" + "403": + $ref: "#/components/responses/ForbiddenError" + "429": + $ref: "#/components/responses/TooManyRequestsError" + "502": + $ref: "#/components/responses/BadGatewayError" "503": $ref: "#/components/responses/ServiceUnavailableError" + "504": + $ref: "#/components/responses/GatewayTimeoutError" "404": $ref: "#/components/responses/NotFoundError" "500": @@ -149,6 +163,12 @@ paths: - $ref: "#/components/parameters/AnswersQuery" - $ref: "#/components/parameters/FormatQuery" - $ref: "#/components/parameters/UseProxyHeader" + - $ref: "#/components/parameters/ProxyURLHeader" + - $ref: "#/components/parameters/ProxyCountryHeader" + - $ref: "#/components/parameters/ProxyClassHeader" + - $ref: "#/components/parameters/ProxyProviderHeader" + - $ref: "#/components/parameters/ProxySessionIDHeader" + - $ref: "#/components/parameters/TenantHeader" responses: "200": description: Image search results envelope @@ -171,8 +191,16 @@ paths: $ref: "#/components/schemas/ImageEnvelope" "400": $ref: "#/components/responses/BadRequestError" + "403": + $ref: "#/components/responses/ForbiddenError" + "429": + $ref: "#/components/responses/TooManyRequestsError" + "502": + $ref: "#/components/responses/BadGatewayError" "503": $ref: "#/components/responses/ServiceUnavailableError" + "504": + $ref: "#/components/responses/GatewayTimeoutError" "404": $ref: "#/components/responses/NotFoundError" "500": @@ -200,6 +228,12 @@ paths: - $ref: "#/components/parameters/EnginesQuery" - $ref: "#/components/parameters/FormatQuery" - $ref: "#/components/parameters/UseProxyHeader" + - $ref: "#/components/parameters/ProxyURLHeader" + - $ref: "#/components/parameters/ProxyCountryHeader" + - $ref: "#/components/parameters/ProxyClassHeader" + - $ref: "#/components/parameters/ProxyProviderHeader" + - $ref: "#/components/parameters/ProxySessionIDHeader" + - $ref: "#/components/parameters/TenantHeader" responses: "200": description: Aggregated envelope with clusters @@ -229,6 +263,16 @@ paths: type: string "400": $ref: "#/components/responses/BadRequestError" + "403": + $ref: "#/components/responses/ForbiddenError" + "429": + $ref: "#/components/responses/TooManyRequestsError" + "502": + $ref: "#/components/responses/BadGatewayError" + "503": + $ref: "#/components/responses/ServiceUnavailableError" + "504": + $ref: "#/components/responses/GatewayTimeoutError" "500": $ref: "#/components/responses/InternalServerError" /mega/image: @@ -249,6 +293,12 @@ paths: - $ref: "#/components/parameters/EnginesQuery" - $ref: "#/components/parameters/FormatQuery" - $ref: "#/components/parameters/UseProxyHeader" + - $ref: "#/components/parameters/ProxyURLHeader" + - $ref: "#/components/parameters/ProxyCountryHeader" + - $ref: "#/components/parameters/ProxyClassHeader" + - $ref: "#/components/parameters/ProxyProviderHeader" + - $ref: "#/components/parameters/ProxySessionIDHeader" + - $ref: "#/components/parameters/TenantHeader" responses: "200": description: Aggregated image results envelope @@ -269,6 +319,16 @@ paths: $ref: "#/components/schemas/ImageEnvelope" "400": $ref: "#/components/responses/BadRequestError" + "403": + $ref: "#/components/responses/ForbiddenError" + "429": + $ref: "#/components/responses/TooManyRequestsError" + "502": + $ref: "#/components/responses/BadGatewayError" + "503": + $ref: "#/components/responses/ServiceUnavailableError" + "504": + $ref: "#/components/responses/GatewayTimeoutError" "500": $ref: "#/components/responses/InternalServerError" /mega/engines: @@ -517,6 +577,67 @@ components: value: direct tag: value: us + ProxyURLHeader: + name: X-Proxy-URL + in: header + required: false + description: > + Per-request proxy URL supplied by an upstream balancer. Honored only when + `proxies.allow_request_proxy_url: true` is set on the worker; otherwise the + request is rejected with `400 bad_request` and `reason=REQUEST_PROXY_URL_DISABLED`. + Authenticated SOCKS proxies are rejected in browser mode + (`reason=UNSUPPORTED_PROXY_SCHEME`). Credentials are never logged or returned. + Precedence: `X-Use-Proxy: direct` > `X-Proxy-URL` > `X-Use-Proxy: ` > + per-engine configured tag > `proxies.global` > direct. + schema: + type: string + example: http://user:pass@proxy.example:8080 + ProxyCountryHeader: + name: X-Proxy-Country + in: header + required: false + description: > + Two-letter market country code for the supplied proxy. Used as part of the + cache key so different markets do not share results. + schema: + type: string + example: us + ProxyClassHeader: + name: X-Proxy-Class + in: header + required: false + description: Proxy class identifier (e.g. `datacenter`, `residential`, `mobile`). Part of the cache key. + schema: + type: string + example: residential + ProxyProviderHeader: + name: X-Proxy-Provider + in: header + required: false + description: Upstream proxy provider identifier (e.g. `webshare`, `brightdata`). Part of the cache key. + schema: + type: string + example: webshare + ProxySessionIDHeader: + name: X-Proxy-Session-ID + in: header + required: false + description: > + Sticky session identifier minted by the balancer. Reusing the same value lets + OpenSERP reuse cookies and browser profile for that lane. Lanes are LRU-bounded + by `proxies.lanes.max_lanes`. Rotating the session ID gives a clean lane. + schema: + type: string + example: sid-123 + TenantHeader: + name: X-Tenant + in: header + required: false + description: > + Optional tenant scope used to namespace sticky lane state across multi-tenant + deployments. When present, lanes are keyed by `tenant + engine + session_id`. + schema: + type: string headers: XRequestID: description: > @@ -534,16 +655,22 @@ components: schema: type: string XProxyMode: - description: Effective proxy mode for the request (`off` or `tag_pool`). + description: > + Effective proxy mode for the request. `request_url` indicates a per-request + `X-Proxy-URL` was honored. schema: type: string - enum: [off, tag_pool] + enum: [off, tag_pool, request_url] XProxyTag: - description: Effective proxy tag when `X-Proxy-Mode=tag_pool`. + description: > + Effective proxy tag when `X-Proxy-Mode=tag_pool`. Header is omitted when no + tag is in effect (i.e. `X-Proxy-Mode` is `request_url` or `off`). schema: type: string XProxyUsed: - description: Effective proxy target used (`direct`, masked URL, `pooled`, `multiple`, `mixed`). + description: > + Effective proxy target used. Values: `direct`, masked `scheme://host:port` URL, + `pooled`, `multiple`, or `mixed`. Credentials are never included. schema: type: string responses: @@ -566,18 +693,142 @@ components: code: 400 message: "EMPTY_QUERY: query cannot be empty" reason: EMPTY_QUERY - ServiceUnavailableError: - description: Search failed and no result could be produced + requestProxyURLDisabled: + value: + error: bad_request + code: 400 + message: "REQUEST_PROXY_URL_DISABLED: X-Proxy-URL is disabled by server configuration" + reason: REQUEST_PROXY_URL_DISABLED + unsupportedProxyScheme: + value: + error: bad_request + code: 400 + message: "UNSUPPORTED_PROXY_SCHEME: authenticated SOCKS proxies are not supported in browser mode" + reason: UNSUPPORTED_PROXY_SCHEME + ForbiddenError: + description: The search engine blocked the request content: application/json: schema: $ref: "#/components/schemas/ErrorResponse" examples: - primaryFailed: + blocked: value: - error: service_unavailable + error: blocked + code: 403 + message: "search engine blocked the request" + meta: + engine: google + proxy_used: http://proxy.example:8080 + proxy_country: us + proxy_class: residential + proxy_provider: webshare + proxy_session_id: sid-123 + TooManyRequestsError: + description: Captcha challenge or rate-limit response from the search engine + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + examples: + captcha: + value: + error: captcha_detected + code: 429 + message: "captcha detected" + meta: + engine: google + proxy_used: http://proxy.example:8080 + proxy_session_id: sid-123 + rateLimited: + value: + error: rate_limited + code: 429 + message: "search engine rate limited the request" + meta: + engine: google + proxy_used: http://proxy.example:8080 + BadGatewayError: + description: Engine internal failure, parser drift, or all-engine failure when fallback is enabled. + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + examples: + parserFailure: + value: + error: parser_failure + code: 502 + message: "parser failure" + meta: + engine: google + engineInternal: + value: + error: engine_internal + code: 502 + message: "engine internal error" + meta: + engine: google + allEnginesFailed: + value: + error: engine_internal + code: 502 + message: "all search engines failed" + meta: + engine: google + GatewayTimeoutError: + description: Search timed out waiting for required SERP elements + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + examples: + searchTimeout: + value: + error: search_timeout + code: 504 + message: "timeout. Cannot find element on page" + meta: + engine: google + ServiceUnavailableError: + description: Proxy-layer failure (no healthy proxy or transport error). The search itself was not produced. + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + examples: + proxyConnect: + value: + error: proxy_connect code: 503 - message: all search engines failed + message: "proxy connection failed" + meta: + engine: google + proxy_used: http://proxy.example:8080 + proxy_country: us + proxyAuth: + value: + error: proxy_auth + code: 503 + message: "proxy authentication failed" + meta: + engine: google + proxy_used: http://proxy.example:8080 + proxyTimeout: + value: + error: proxy_timeout + code: 503 + message: "proxy request timed out" + meta: + engine: google + proxy_used: http://proxy.example:8080 + proxyUnavailable: + value: + error: proxy_unavailable + code: 503 + message: "proxy unavailable" + meta: + engine: google NotFoundError: description: Endpoint not found content: @@ -921,8 +1172,31 @@ components: properties: error: type: string - description: Stable machine-readable error class. - enum: [bad_request, not_found, rate_limited, service_unavailable, server_error, client_error, error] + description: > + Stable machine-readable error class. Search-pipeline failures use the + following codes: `captcha_detected`, `blocked`, `rate_limited`, + `search_timeout`, `proxy_connect`, `proxy_auth`, `proxy_timeout`, + `proxy_unavailable`, `parser_failure`, `engine_internal`. Validation + errors use `bad_request`. Other generic codes (`not_found`, + `service_unavailable`, `server_error`, `client_error`, `error`) may + appear for non-search routes. + enum: + - bad_request + - not_found + - rate_limited + - service_unavailable + - server_error + - client_error + - error + - captcha_detected + - blocked + - search_timeout + - proxy_connect + - proxy_auth + - proxy_timeout + - proxy_unavailable + - parser_failure + - engine_internal example: bad_request code: type: integer @@ -935,8 +1209,35 @@ components: description: > Stable client-actionable reason code. Present on 400 errors. Known values: INVALID_LIMIT, INVALID_START, INVALID_PARAM, EMPTY_QUERY, - NO_ENGINES, UNKNOWN_FORMAT. + NO_ENGINES, UNKNOWN_FORMAT, REQUEST_PROXY_URL_DISABLED, + UNSUPPORTED_PROXY_SCHEME. example: INVALID_LIMIT + meta: + type: object + description: > + Sanitized context for search-pipeline errors. Credentials are never + included. + additionalProperties: true + properties: + engine: + type: string + example: google + proxy_used: + type: string + description: Masked `scheme://host:port`; never includes credentials. + example: http://proxy.example:8080 + proxy_country: + type: string + example: us + proxy_class: + type: string + example: residential + proxy_provider: + type: string + example: webshare + proxy_session_id: + type: string + example: sid-123 # ── Health / Stats ──────────────────────────────────────────────── EngineHealth: type: object @@ -1039,9 +1340,61 @@ components: type: string selected_proxy: type: string + LaneStats: + type: object + required: [active, evicted_lru, cookies_dropped] + description: Sticky proxy lane state observed by this worker. + properties: + active: + type: integer + description: Number of lanes currently held by the worker. + example: 12 + evicted_lru: + type: integer + description: Lanes evicted by the LRU bound since worker start. + example: 7 + cookies_dropped: + type: integer + description: Lane cookie drops triggered by captcha/challenge responses. + example: 20 + BrowserPoolStats: + type: object + required: [active, max, evicted_lru, evicted_idle] + description: > + Live state of the per-process Chrome pool. Each authenticated upstream + proxy identity (scheme+host+port+username) gets a dedicated Chrome so + Chrome can answer 407 challenges natively. Direct and unauthenticated + proxies share one Chrome with per-BrowserContext proxy override. + properties: + active: + type: integer + description: Number of Chrome processes currently held by the pool. + example: 3 + max: + type: integer + description: Configured `app.max_processes` LRU cap. + example: 4 + evicted_lru: + type: integer + description: Chrome processes closed because the LRU cap was exceeded. + example: 12 + evicted_idle: + type: integer + description: Chrome processes closed by the idle sweeper after `app.idle_ttl`. + example: 5 ProxyStats: type: object - required: [configured_count, healthy_count, unhealthy_count, tags, entries] + required: + [ + configured_count, + healthy_count, + unhealthy_count, + request_proxy_url_enabled, + lanes, + browser_processes, + tags, + entries, + ] properties: configured_count: type: integer @@ -1049,6 +1402,13 @@ components: type: integer unhealthy_count: type: integer + request_proxy_url_enabled: + type: boolean + description: Whether `proxies.allow_request_proxy_url` is enabled on this worker. + lanes: + $ref: "#/components/schemas/LaneStats" + browser_processes: + $ref: "#/components/schemas/BrowserPoolStats" tags: type: object additionalProperties: