zlei6/openserp
1
0
Fork 0
mirror of https://github.com/karust/openserp.git synced 2026-09-07 18:07:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(proxy): support X-Proxy-URL via per-process browser pool

Browse Source
This commit is contained in:
Rustem Kamalov
2026-04-28 03:46:08 +03:00
parent 7a0fb21daf
commit e8fb0c24fa
23 changed files with 2732 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

353
core/browser.go
View File

@@ -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) {

157
core/browser_test.go
View File

@@ -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" {

28
core/cache.go
View File

@@ -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()

73
core/cache_test.go
View File

@@ -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")
}
}

43
core/common.go
View File

@@ -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()

19
core/errors.go
View File

@@ -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 {

5
core/logger.go
View File

@@ -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{}

46
core/middleware.go
View File

@@ -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")

26
core/proxy.go
View File

@@ -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
}

37
core/proxy_context.go Normal file
View File

@@ -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)
}

289
core/proxy_lane.go Normal file
View File

@@ -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
}

101
core/proxy_lane_test.go Normal file
View File

@@ -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)
}
}

93
core/proxy_per_context_spike_test.go Normal file
View File

@@ -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, "<html><body>%s</body></html>", 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 "))
}
}

31
core/proxy_test.go
View File

@@ -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)

105
core/resilient.go
View File

@@ -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")

62
core/retry.go
View File

@@ -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) {

153
core/server.go
View File

@@ -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)

404
core/server_test.go
View File

@@ -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)
}