diff --git a/.a2a/agent.dsl.json b/.a2a/agent.dsl.json index da3965f..011dea1 100644 --- a/.a2a/agent.dsl.json +++ b/.a2a/agent.dsl.json @@ -3,7 +3,7 @@ "language": "python", "name": "seleniumbase-website-scraper", "description": "SeleniumBase-style browser agent that turns observed website traffic into OpenAPI specs, coverage reports, samples, and a replay client.", - "version": "0.2.1", + "version": "0.2.2", "entrypoint": { "module": "agent", "class_name": "BrowserToApiAgent", @@ -125,6 +125,25 @@ "type": "null" } ] + }, + "browser_mode": { + "type": "string" + }, + "captcha_policy": { + "type": "string" + }, + "allowed_captcha_domains": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ] } }, "required": [ @@ -257,7 +276,8 @@ "samples/*.json" ], "noise_filtering": "Drops analytics, static assets, service workers, and obvious bot-defense plumbing by default.", - "schema_inference": "Infers request/response JSON schemas, query params, path params, GraphQL dispatches, and auth-shaped headers." + "schema_inference": "Infers request/response JSON schemas, query params, path params, GraphQL dispatches, and auth-shaped headers.", + "captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains." } }, "input_modes": [ @@ -344,7 +364,7 @@ "source": "python-a2a-pack", "project_manifest": { "name": "seleniumbase-website-scraper", - "version": "0.2.1", + "version": "0.2.2", "entrypoint": "agent:BrowserToApiAgent" } } diff --git a/a2a.yaml b/a2a.yaml index 87e8eb1..463380f 100644 --- a/a2a.yaml +++ b/a2a.yaml @@ -1,5 +1,5 @@ name: seleniumbase-website-scraper -version: 0.2.1 +version: 0.2.2 entrypoint: agent:BrowserToApiAgent expose: public: true diff --git a/agent.py b/agent.py index b55ae38..bbe435c 100644 --- a/agent.py +++ b/agent.py @@ -69,6 +69,11 @@ NOISE_URL_PATTERNS = [ r"cloudflareinsights\.com", r"sentry\.io", r"datadog(hq)?\.com", + r"perimeterx|px-captcha|_px", + r"/px/.*/captcha", + r"/akam/", + r"/si/[^/]+/obs", + r"captcha(\.js|/|\?)", r"/(pixel|beacon|track|pageview|impression)(/|$|\?)", r"\.(png|jpe?g|gif|svg|webp|ico|woff2?|ttf|eot|otf|css|map|mp4|webm|mp3)(\?|$)", r"/(sw|service-worker)\.js(\?|$)", @@ -76,6 +81,9 @@ NOISE_URL_PATTERNS = [ r"/robots\.txt(\?|$)", r"/favicon\.ico(\?|$)", ] +DEFAULT_CAPTCHA_SOLVE_DOMAINS = ("a2acloud.io", ".a2acloud.io") +CAPTCHA_POLICIES = {"detect_only", "solve_if_allowed", "never"} +BROWSER_MODES = {"seleniumbase_stealth", "playwright"} class BrowserToApiConfig(BaseModel): @@ -119,7 +127,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): "SeleniumBase-style browser agent that turns observed website traffic " "into OpenAPI specs, coverage reports, samples, and a replay client." ) - version = "0.2.1" + version = "0.2.2" config_model = BrowserToApiConfig auth_model = NoAuth @@ -141,6 +149,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): ], "noise_filtering": "Drops analytics, static assets, service workers, and obvious bot-defense plumbing by default.", "schema_inference": "Infers request/response JSON schemas, query params, path params, GraphQL dispatches, and auth-shaped headers.", + "captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains.", } } @@ -166,9 +175,16 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): search_term: str | None = None, min_samples: int = 1, redact: list[str] | None = None, + browser_mode: str = "seleniumbase_stealth", + captcha_policy: str = "solve_if_allowed", + allowed_captcha_domains: list[str] | None = None, ) -> dict[str, Any]: if not url or not str(url).strip().startswith(("http://", "https://")): return {"error": "url must start with http:// or https://"} + if browser_mode not in BROWSER_MODES: + return {"error": f"browser_mode must be one of: {', '.join(sorted(BROWSER_MODES))}"} + if captcha_policy not in CAPTCHA_POLICIES: + return {"error": f"captcha_policy must be one of: {', '.join(sorted(CAPTCHA_POLICIES))}"} try: await ctx.emit_progress("launching browser and capturing network traffic") @@ -178,6 +194,9 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): wait_seconds=max(1, min(int(wait_seconds), 60)), max_clicks=max(0, min(int(max_clicks), 20)), search_term=search_term, + browser_mode=browser_mode, + captcha_policy=captcha_policy, + allowed_captcha_domains=allowed_captcha_domains, ) await ctx.emit_progress(f"captured {len(samples)} network samples; inferring API surface") @@ -193,7 +212,16 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]): output_dir = write_bundle(bundle, DEFAULT_OUTPUT_DIR / safe_run_id(url)) await ctx.emit_progress(f"wrote OpenAPI bundle to {output_dir}") - return _result_payload(url=url, samples=samples, bundle=bundle, output_dir=output_dir, capture_log=capture_log) + return _result_payload( + url=url, + samples=samples, + bundle=bundle, + output_dir=output_dir, + capture_log=capture_log, + browser_mode=browser_mode, + captcha_policy=captcha_policy, + allowed_captcha_domains=_effective_captcha_domains(allowed_captcha_domains), + ) except Exception as exc: # noqa: BLE001 return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"} @@ -243,6 +271,9 @@ def capture_browser_traffic( wait_seconds: int = 8, max_clicks: int = 4, search_term: str | None = None, + browser_mode: str = "seleniumbase_stealth", + captcha_policy: str = "solve_if_allowed", + allowed_captcha_domains: list[str] | None = None, ) -> tuple[list[TrafficSample], list[str]]: """Capture useful browser traffic with Playwright. @@ -255,6 +286,15 @@ def capture_browser_traffic( log: list[str] = [] samples_by_request: dict[int, TrafficSample] = {} + stealth_cookies: list[dict[str, Any]] = [] + if browser_mode == "seleniumbase_stealth": + stealth_cookies, stealth_log = seleniumbase_stealth_prep( + url=url, + captcha_policy=captcha_policy, + allowed_captcha_domains=allowed_captcha_domains, + wait_seconds=min(wait_seconds, 12), + ) + log.extend(stealth_log) with sync_playwright() as p: browser = _launch_chromium(p, log) @@ -267,6 +307,12 @@ def capture_browser_traffic( "Chrome/126.0 Safari/537.36" ), ) + if stealth_cookies: + try: + context.add_cookies(stealth_cookies) + log.append(f"browser: transferred {len(stealth_cookies)} SeleniumBase cookies") + except Exception as exc: # noqa: BLE001 + log.append(f"browser warning: cookie transfer failed: {type(exc).__name__}: {exc}") page = context.new_page() def on_request(request: Any) -> None: @@ -330,6 +376,123 @@ def capture_browser_traffic( return list(samples_by_request.values()), log +def seleniumbase_stealth_prep( + *, + url: str, + captcha_policy: str, + allowed_captcha_domains: list[str] | None, + wait_seconds: int, +) -> tuple[list[dict[str, Any]], list[str]]: + log: list[str] = [] + cookies: list[dict[str, Any]] = [] + allowed_domains = _effective_captcha_domains(allowed_captcha_domains) + captcha_allowed = _host_allowed_for_captcha(url, allowed_domains) + + try: + from seleniumbase import SB + except Exception as exc: # noqa: BLE001 + return [], [f"seleniumbase warning: import failed: {type(exc).__name__}: {exc}"] + + try: + with SB(uc=True, test=True, locale="en-US", use_chromium=True) as sb: + log.append("seleniumbase: started UC/CDP stealth session") + try: + sb.uc_open_with_reconnect(url, reconnect_time=3) + log.append("seleniumbase: uc_open_with_reconnect") + except Exception as exc: # noqa: BLE001 + log.append(f"seleniumbase warning: uc reconnect failed: {type(exc).__name__}: {exc}") + try: + sb.open(url) + log.append("seleniumbase: open") + except Exception as open_exc: # noqa: BLE001 + log.append(f"seleniumbase warning: open failed: {type(open_exc).__name__}: {open_exc}") + _seleniumbase_sleep(sb, min(max(wait_seconds, 2), 8)) + + if captcha_policy == "solve_if_allowed": + if captcha_allowed: + try: + sb.solve_captcha() + log.append("seleniumbase: solve_captcha attempted for allowed domain") + _seleniumbase_sleep(sb, 2) + except Exception as exc: # noqa: BLE001 + log.append(f"seleniumbase warning: solve_captcha failed: {type(exc).__name__}: {exc}") + else: + host = urlparse(url).netloc + log.append(f"seleniumbase: captcha solve skipped; host not allowlisted: {host}") + elif captcha_policy == "detect_only": + log.append("seleniumbase: captcha policy detect_only") + else: + log.append("seleniumbase: captcha policy never") + + try: + raw_cookies = sb.driver.get_cookies() + cookies = _playwright_cookies(raw_cookies, url) + log.append(f"seleniumbase: collected {len(cookies)} cookies") + except Exception as exc: # noqa: BLE001 + log.append(f"seleniumbase warning: cookie collection failed: {type(exc).__name__}: {exc}") + except Exception as exc: # noqa: BLE001 + log.append(f"seleniumbase warning: stealth prep failed: {type(exc).__name__}: {exc}") + return cookies, log + + +def _seleniumbase_sleep(sb: Any, seconds: int) -> None: + try: + sb.sleep(seconds) + except Exception: + time.sleep(seconds) + + +def _effective_captcha_domains(allowed_captcha_domains: list[str] | None) -> list[str]: + domains = [str(item).strip().lower() for item in (allowed_captcha_domains or []) if str(item).strip()] + for domain in DEFAULT_CAPTCHA_SOLVE_DOMAINS: + if domain not in domains: + domains.append(domain) + return domains + + +def _host_allowed_for_captcha(url: str, allowed_domains: list[str]) -> bool: + host = urlparse(url).hostname or "" + host = host.lower().rstrip(".") + for raw_domain in allowed_domains: + domain = raw_domain.lower().strip().lstrip("*").rstrip(".") + if not domain: + continue + if domain.startswith("."): + suffix = domain[1:] + if host == suffix or host.endswith(domain): + return True + elif host == domain or host.endswith(f".{domain}"): + return True + return False + + +def _playwright_cookies(raw_cookies: list[dict[str, Any]], url: str) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + parsed = urlparse(url) + fallback_domain = parsed.hostname or "" + for item in raw_cookies: + name = item.get("name") + value = item.get("value") + if not name or value is None: + continue + cookie: dict[str, Any] = { + "name": str(name), + "value": str(value), + "domain": str(item.get("domain") or fallback_domain), + "path": str(item.get("path") or "/"), + "secure": bool(item.get("secure", parsed.scheme == "https")), + "httpOnly": bool(item.get("httpOnly", False)), + } + expiry = item.get("expiry") or item.get("expires") + if isinstance(expiry, (int, float)) and expiry > 0: + cookie["expires"] = float(expiry) + same_site = item.get("sameSite") + if same_site in {"Strict", "Lax", "None"}: + cookie["sameSite"] = same_site + out.append(cookie) + return out + + def _launch_chromium(p: Any, log: list[str]) -> Any: launch_kwargs = { "headless": True, @@ -872,7 +1035,11 @@ def _result_payload( output_dir: Path, url: str | None = None, capture_log: list[str] | None = None, + browser_mode: str | None = None, + captcha_policy: str | None = None, + allowed_captcha_domains: list[str] | None = None, ) -> dict[str, Any]: + bot_signals = _bot_defense_signals(samples, capture_log or []) payload: dict[str, Any] = { "captured_samples": len(samples), "included_samples": bundle["summary"]["included_samples"], @@ -886,14 +1053,42 @@ def _result_payload( "client": str(output_dir / "client.mjs"), "confidence": str(output_dir / "confidence.json"), "top_endpoints": bundle["summary"]["top_endpoints"], + "bot_defense_detected": bool(bot_signals), + "bot_defense_signals": bot_signals, } if url is not None: payload["url"] = url + if browser_mode is not None: + payload["browser_mode"] = browser_mode + if captcha_policy is not None: + payload["captcha_policy"] = captcha_policy + if allowed_captcha_domains is not None: + payload["allowed_captcha_domains"] = allowed_captcha_domains if capture_log is not None: payload["capture_log"] = capture_log return payload +def _bot_defense_signals(samples: list[TrafficSample], capture_log: list[str]) -> list[str]: + checks = { + "captcha": (r"captcha",), + "perimeterx": (r"perimeterx", r"/px/", r"_px", r"px-captcha"), + "akamai": (r"/akam/", r"akamai", r"bm_sz", r"_abck"), + "datadome": (r"datadome",), + "cloudflare": (r"cf-chl", r"cloudflare"), + } + haystacks = [sample.url for sample in samples] + for sample in samples: + haystacks.extend(str(key) for key in sample.request_headers) + haystacks.extend(str(key) for key in sample.response_headers) + haystacks.extend(capture_log) + found: list[str] = [] + for name, patterns in checks.items(): + if any(_regex_search(pattern, haystack) for pattern in patterns for haystack in haystacks): + found.append(name) + return found + + def infer_schema(values: list[Any]) -> dict[str, Any]: values = [value for value in values if value is not None] if not values: