"""Browser-to-API discovery agent.
The Browserbase browser-to-api skill is a strong offline pipeline. This agent
turns the same idea into a hosted A2A workflow: drive a browser, capture useful
HTTP traffic, infer schemas, and emit a deployable OpenAPI bundle.
"""
from __future__ import annotations
import argparse
import asyncio
import hashlib
import html
import json
import os
import re
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.request
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
from urllib.parse import parse_qsl, urlparse
from a2a_pack import A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill
from pydantic import BaseModel, Field
DEFAULT_OUTPUT_DIR = Path(os.environ.get("BROWSER_TO_API_OUTPUT_DIR", "/tmp/browser-to-api"))
JSONISH_TYPES = (
"application/json",
"application/problem+json",
"application/vnd.api+json",
"text/json",
)
TEXT_TYPES = ("text/plain", "application/x-ndjson")
DEFAULT_REDACT_KEYS = {
"authorization",
"cookie",
"set-cookie",
"x-csrf-token",
"x-xsrf-token",
"x-api-key",
"proxy-authorization",
"password",
"token",
"secret",
"api_key",
"apikey",
"access_token",
"accesstoken",
"refresh_token",
"refreshtoken",
"creditcard",
"ssn",
}
NOISE_URL_PATTERNS = [
r"google-analytics\.com",
r"googletagmanager\.com",
r"doubleclick\.net",
r"facebook\.com/tr",
r"segment\.(io|com)",
r"mixpanel\.com",
r"amplitude\.com",
r"fullstory\.com",
r"hotjar\.com",
r"intercom\.io",
r"clarity\.ms",
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(\?|$)",
r"/manifest\.json(\?|$)",
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):
pass
@dataclass
class TrafficSample:
request_id: str
method: str
url: str
status: int | None = None
resource_type: str | None = None
request_headers: dict[str, Any] = field(default_factory=dict)
response_headers: dict[str, Any] = field(default_factory=dict)
request_body: Any = None
response_body: Any = None
response_body_truncated: bool = False
content_type: str | None = None
timestamp: float = field(default_factory=time.time)
error: str | None = None
@dataclass
class EndpointGroup:
key: str
origin: str
method: str
path: str
source_path: str
dispatch: str | None
samples: list[TrafficSample] = field(default_factory=list)
path_params: list[dict[str, Any]] = field(default_factory=list)
query_params: dict[str, list[str]] = field(default_factory=dict)
flags: list[str] = field(default_factory=list)
class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
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.9"
config_model = BrowserToApiConfig
auth_model = NoAuth
llm_provisioning = LLMProvisioning.PLATFORM
pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False)
tools_used = ("seleniumbase", "playwright", "browser-to-api", "openapi")
capabilities = {
"browser_to_api": {
"capture": "Launches a browser, records XHR/fetch/document traffic with response bodies, then infers OpenAPI.",
"offline": "Can replay JSONL request/response captures into the same OpenAPI emitter.",
"outputs": [
"openapi.json",
"openapi.yaml",
"report.md",
"index.html",
"client.mjs",
"confidence.json",
"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.",
"captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains.",
"llm_exploration": "Uses ctx.llm to choose bounded, non-destructive page actions after the page loads.",
}
}
@skill(
name="discover_api_from_browser",
description=(
"Open a website in a browser, capture observable HTTP traffic, infer "
"an OpenAPI 3.1 spec, and write a browsable API discovery report."
),
tags=("browser", "seleniumbase", "openapi", "api-discovery"),
timeout_seconds=600,
)
async def discover_api_from_browser(
self,
ctx: RunContext[NoAuth],
url: str,
title: str | None = None,
origins: list[str] | None = None,
include: list[str] | None = None,
exclude: list[str] | None = None,
wait_seconds: int = 8,
max_clicks: int = 4,
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,
ai_explore: bool = True,
ai_steps: int = 4,
ai_goal: 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")
samples, capture_log = await asyncio.to_thread(
capture_browser_traffic,
url=str(url).strip(),
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,
ai_explore=ai_explore,
ai_steps=max(0, min(int(ai_steps), 8)),
ai_goal=ai_goal,
llm_config=_llm_config_from_ctx(ctx),
)
await ctx.emit_progress(f"captured {len(samples)} network samples; inferring API surface")
bundle = build_openapi_bundle(
samples,
title=title or _title_from_url(url),
origins=origins,
include=include,
exclude=exclude,
min_samples=max(1, int(min_samples)),
redact=redact,
)
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,
browser_mode=browser_mode,
captcha_policy=captcha_policy,
allowed_captcha_domains=_effective_captcha_domains(allowed_captcha_domains),
ai_explore=ai_explore,
ai_steps=max(0, min(int(ai_steps), 8)),
)
except Exception as exc: # noqa: BLE001
return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"}
@skill(
name="discover_api_from_trace",
description=(
"Replay request/response JSONL traffic into the same OpenAPI 3.1 "
"discovery pipeline used by live browser capture."
),
tags=("trace", "openapi", "api-discovery"),
timeout_seconds=240,
)
async def discover_api_from_trace(
self,
ctx: RunContext[NoAuth],
requests_jsonl: str,
responses_jsonl: str = "",
title: str = "Discovered Website API",
origins: list[str] | None = None,
include: list[str] | None = None,
exclude: list[str] | None = None,
min_samples: int = 1,
redact: list[str] | None = None,
) -> dict[str, Any]:
try:
await ctx.emit_progress("pairing replayed request/response trace")
samples = parse_trace_jsonl(requests_jsonl=requests_jsonl, responses_jsonl=responses_jsonl)
bundle = build_openapi_bundle(
samples,
title=title,
origins=origins,
include=include,
exclude=exclude,
min_samples=max(1, int(min_samples)),
redact=redact,
)
output_dir = write_bundle(bundle, DEFAULT_OUTPUT_DIR / safe_run_id(title))
await ctx.emit_progress(f"wrote replayed OpenAPI bundle to {output_dir}")
return _result_payload(samples=samples, bundle=bundle, output_dir=output_dir)
except Exception as exc: # noqa: BLE001
return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"}
def capture_browser_traffic(
*,
url: str,
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,
ai_explore: bool = True,
ai_steps: int = 4,
ai_goal: str | None = None,
llm_config: dict[str, Any] | None = None,
) -> tuple[list[TrafficSample], list[str]]:
"""Capture useful browser traffic with Playwright.
SeleniumBase remains part of the packaged agent because it is valuable for
anti-bot/UC flows, but Playwright's response body APIs make it the pragmatic
capture engine for schema inference.
"""
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
from playwright.sync_api import sync_playwright
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)
context = browser.new_context(
viewport={"width": 1440, "height": 1000},
ignore_https_errors=True,
user_agent=(
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"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:
parsed_body = _parse_maybe_json(request.post_data)
samples_by_request[id(request)] = TrafficSample(
request_id=str(id(request)),
method=request.method.upper(),
url=request.url,
resource_type=request.resource_type,
request_headers=dict(request.headers or {}),
request_body=parsed_body,
timestamp=time.time(),
)
def on_response(response: Any) -> None:
request = response.request
sample = samples_by_request.get(id(request))
if not sample:
sample = TrafficSample(
request_id=str(id(request)),
method=request.method.upper(),
url=request.url,
resource_type=request.resource_type,
request_headers=dict(request.headers or {}),
request_body=_parse_maybe_json(request.post_data),
timestamp=time.time(),
)
samples_by_request[id(request)] = sample
sample.status = response.status
sample.response_headers = dict(response.headers or {})
sample.content_type = sample.response_headers.get("content-type")
try:
body = response.body()
sample.response_body_truncated = len(body) > 1_000_000
if body and _is_body_worth_capturing(sample.content_type):
body = body[:1_000_000]
text = body.decode("utf-8", errors="replace")
sample.response_body = _parse_maybe_json(text)
except Exception as exc: # noqa: BLE001
sample.error = f"{type(exc).__name__}: {exc}"
page.on("request", on_request)
page.on("response", on_response)
try:
page.goto(url, wait_until="domcontentloaded", timeout=45_000)
log.append(f"open: {url}")
except PlaywrightTimeoutError:
log.append("open warning: initial navigation timed out after domcontentloaded wait")
except Exception as exc: # noqa: BLE001
log.append(f"open warning: {type(exc).__name__}: {exc}")
page.wait_for_timeout(wait_seconds * 1000)
if ai_explore and ai_steps > 0 and llm_config:
log.append(
"ai: using ctx.llm "
f"model={llm_config.get('model', 'unknown')} "
f"source={llm_config.get('source', 'unknown')}"
)
_ai_explore_page(
page,
llm_config=llm_config,
steps=min(ai_steps, max_clicks if max_clicks > 0 else ai_steps),
goal=ai_goal or _default_ai_goal(url),
search_term=search_term,
log=log,
)
elif ai_explore and ai_steps > 0:
log.append("ai: ctx.llm unavailable; falling back to passive scroll only")
_auto_scroll(page, log)
else:
_fill_search_inputs(page, search_term, log)
_auto_scroll(page, log)
_click_interesting_elements(page, max_clicks=max_clicks, log=log)
page.wait_for_timeout(1500)
context.close()
browser.close()
return list(samples_by_request.values()), log
def _llm_config_from_ctx(ctx: RunContext[NoAuth]) -> dict[str, Any] | None:
creds = getattr(ctx, "llm", None)
if creds is None:
return None
api_key = str(getattr(creds, "api_key", "") or "")
base_url = str(getattr(creds, "base_url", "") or "").rstrip("/")
model = str(getattr(creds, "model", "") or "")
if not (api_key and base_url and model):
return None
return {
"api_key": api_key,
"base_url": base_url,
"model": model,
"source": str(getattr(creds, "source", "") or ""),
"extra_body": dict(getattr(creds, "extra_body", None) or {}),
}
def _default_ai_goal(url: str) -> str:
host = urlparse(url).netloc or "this site"
return (
f"Explore {host} to reveal useful same-site API, JSON, XHR, fetch, search, "
"navigation, post listing, filter, load-more, or content endpoints. Avoid "
"login, signup, checkout, destructive, payment, account, or external actions."
)
def _ai_explore_page(
page: Any,
*,
llm_config: dict[str, Any],
steps: int,
goal: str,
search_term: str | None,
log: list[str],
) -> None:
for step in range(1, max(steps, 0) + 1):
snapshot = _page_action_snapshot(page)
action = _llm_choose_action(
llm_config,
snapshot=snapshot,
goal=goal,
step=step,
search_term=search_term,
log=log,
)
if not action:
log.append(f"ai: step {step}: no valid action returned")
_auto_scroll(page, log)
continue
applied = _apply_ai_action(page, action, snapshot.get("candidates", []), search_term, log, step)
if not applied or action.get("action") == "stop":
break
page.wait_for_timeout(1800)
def _page_action_snapshot(page: Any) -> dict[str, Any]:
try:
candidates = page.evaluate(
"""
() => {
const nodes = Array.from(document.querySelectorAll(
'a[href], button, [role="button"], input, textarea, select'
));
const out = [];
let index = 0;
for (const node of nodes) {
const rect = node.getBoundingClientRect();
const style = window.getComputedStyle(node);
if (rect.width < 4 || rect.height < 4 || style.visibility === 'hidden' || style.display === 'none') {
continue;
}
const tag = node.tagName.toLowerCase();
const text = (node.innerText || node.value || node.getAttribute('aria-label') || node.getAttribute('title') || '').trim();
const href = node.getAttribute('href') || '';
const placeholder = node.getAttribute('placeholder') || '';
const role = node.getAttribute('role') || '';
const type = node.getAttribute('type') || '';
const id = `ai-discovery-${index}`;
node.setAttribute('data-ai-discovery-id', id);
out.push({ index, selector: `[data-ai-discovery-id="${id}"]`, tag, text: text.slice(0, 140), href, placeholder, role, type });
index += 1;
if (out.length >= 45) break;
}
return out;
}
"""
)
except Exception:
candidates = []
try:
visible_text = page.locator("body").inner_text(timeout=1500)[:4000]
except Exception:
visible_text = ""
return {
"url": page.url,
"title": _safe_page_title(page),
"visible_text": visible_text,
"candidates": candidates if isinstance(candidates, list) else [],
}
def _safe_page_title(page: Any) -> str:
try:
return str(page.title() or "")
except Exception:
return ""
def _llm_choose_action(
llm_config: dict[str, Any],
*,
snapshot: dict[str, Any],
goal: str,
step: int,
search_term: str | None,
log: list[str],
) -> dict[str, Any] | None:
candidates = snapshot.get("candidates", [])
compact_candidates = [
{key: item.get(key) for key in ("index", "tag", "text", "href", "placeholder", "role", "type")}
for item in candidates[:45]
if isinstance(item, dict)
]
system = (
"You choose one safe browser action for API discovery. Return only JSON. "
"Allowed actions: click, fill, scroll, wait, stop. Prefer actions likely "
"to trigger XHR/fetch/API traffic. Avoid login, signup, checkout, delete, "
"logout, payment, account settings, destructive operations, and external domains."
)
user = {
"goal": goal,
"step": step,
"current_url": snapshot.get("url"),
"title": snapshot.get("title"),
"visible_text": str(snapshot.get("visible_text") or "")[:2200],
"search_term": search_term,
"candidates": compact_candidates,
"response_schema": {
"action": "click|fill|scroll|wait|stop",
"index": "candidate index for click/fill, optional",
"text": "text to fill, optional",
"reason": "short reason",
},
}
try:
response = _chat_completion_json(llm_config, system=system, user=json.dumps(user, ensure_ascii=False))
except Exception as exc: # noqa: BLE001
log.append(f"ai warning: llm action selection failed: {type(exc).__name__}: {exc}")
return None
action = _parse_json_object(response)
if not isinstance(action, dict):
log.append(f"ai warning: non-json action: {_truncate_text(response, 240)}")
return None
return action
def _chat_completion_json(llm_config: dict[str, Any], *, system: str, user: str) -> str:
url = f"{llm_config['base_url'].rstrip('/')}/chat/completions"
base_payload = {
"model": llm_config["model"],
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
}
extra_body = dict(llm_config.get("extra_body") or {})
attempts = [
{
**base_payload,
"max_completion_tokens": 1200,
"reasoning_effort": "minimal",
**extra_body,
},
{
**base_payload,
"max_tokens": 800,
**extra_body,
},
{
**base_payload,
"max_completion_tokens": 1200,
"response_format": {"type": "json_object"},
**extra_body,
},
]
last_error: Exception | None = None
for payload in attempts:
try:
content = _post_chat_completion(url, llm_config["api_key"], payload)
except Exception as exc: # noqa: BLE001
last_error = exc
continue
if content.strip():
return content
if last_error is not None:
raise last_error
return ""
def _post_chat_completion(url: str, api_key: str, payload: dict[str, Any]) -> str:
req = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={
"authorization": f"Bearer {api_key}",
"content-type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=45) as response:
data = json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")[:1000]
raise RuntimeError(f"LLM HTTP {exc.code}: {detail}") from exc
choices = data.get("choices") if isinstance(data, dict) else None
if not choices:
return ""
message = choices[0].get("message") if isinstance(choices[0], dict) else {}
content = message.get("content")
if isinstance(content, list):
parts = [
str(item.get("text") or item.get("content") or "")
for item in content
if isinstance(item, dict)
]
return "\n".join(part for part in parts if part)
return str(content or "")
def _apply_ai_action(
page: Any,
action: dict[str, Any],
candidates: list[dict[str, Any]],
search_term: str | None,
log: list[str],
step: int,
) -> bool:
action_name = str(action.get("action") or "").lower().strip()
reason = str(action.get("reason") or "").strip()
if action_name == "stop":
log.append(f"ai: step {step}: stop ({reason})")
return False
if action_name == "wait":
log.append(f"ai: step {step}: wait ({reason})")
page.wait_for_timeout(2000)
return True
if action_name == "scroll":
log.append(f"ai: step {step}: scroll ({reason})")
_auto_scroll(page, log)
return True
index = _int_or_none(action.get("index"))
candidate = _candidate_by_index(candidates, index)
if not candidate:
log.append(f"ai warning: step {step}: missing candidate for action {action_name}")
return False
text_bits = " ".join(str(candidate.get(key) or "") for key in ("text", "href", "placeholder")).lower()
if _looks_destructive(text_bits):
log.append(f"ai: step {step}: skipped destructive-looking candidate {index}")
return True
selector = str(candidate.get("selector") or "")
if not selector:
return False
try:
locator = page.locator(selector).first
if action_name == "fill":
fill_text = str(action.get("text") or search_term or "api").strip()[:120]
locator.fill(fill_text, timeout=3000)
locator.press("Enter", timeout=3000)
log.append(f"ai: step {step}: fill candidate {index} ({reason})")
return True
if action_name == "click":
locator.click(timeout=4000)
log.append(f"ai: step {step}: click candidate {index} ({reason})")
return True
except Exception as exc: # noqa: BLE001
log.append(f"ai warning: step {step}: {action_name} failed: {type(exc).__name__}: {exc}")
return True
log.append(f"ai warning: step {step}: unsupported action {action_name}")
return False
def _candidate_by_index(candidates: list[dict[str, Any]], index: int | None) -> dict[str, Any] | None:
if index is None:
return None
for item in candidates:
if isinstance(item, dict) and item.get("index") == index:
return item
return None
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(**_seleniumbase_options()) 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:
log.extend(_attempt_allowed_captcha_solve(sb))
_seleniumbase_sleep(sb, 2)
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 _attempt_allowed_captcha_solve(sb: Any) -> list[str]:
log: list[str] = []
for method_name in ("solve_captcha", "uc_gui_click_captcha", "uc_gui_handle_captcha"):
method = getattr(sb, method_name, None)
if method is None:
continue
try:
method()
log.append(f"seleniumbase: {method_name} attempted for allowed domain")
return log
except Exception as exc: # noqa: BLE001
log.append(f"seleniumbase warning: {method_name} failed: {type(exc).__name__}: {exc}")
return log
log.append("seleniumbase warning: no supported CAPTCHA helper found")
return log
def _seleniumbase_options() -> dict[str, Any]:
options: dict[str, Any] = {
"uc": True,
"test": True,
"locale": "en-US",
"xvfb": True,
"chromium_arg": "--no-sandbox,--disable-dev-shm-usage,--disable-gpu",
}
system_chromium = _system_chromium_path()
if system_chromium:
options["binary_location"] = system_chromium
else:
options["use_chromium"] = True
return options
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,
"args": [
"--disable-dev-shm-usage",
"--disable-blink-features=AutomationControlled",
"--no-sandbox",
],
}
system_chromium = _system_chromium_path()
if system_chromium:
try:
log.append(f"browser: using system chromium at {system_chromium}")
return p.chromium.launch(executable_path=system_chromium, **launch_kwargs)
except Exception as exc: # noqa: BLE001
log.append(f"browser warning: system chromium failed: {type(exc).__name__}: {exc}")
try:
return p.chromium.launch(**launch_kwargs)
except Exception as exc: # noqa: BLE001
message = str(exc)
if "Executable doesn't exist" not in message and "playwright install" not in message:
raise
log.append("browser: chromium executable missing; installing with playwright")
proc = subprocess.run(
[sys.executable, "-m", "playwright", "install", "chromium"],
text=True,
capture_output=True,
timeout=300,
check=False,
)
if proc.returncode != 0:
detail = (proc.stderr or proc.stdout or "").strip()[-2000:]
raise RuntimeError(f"playwright install chromium failed: {detail}") from exc
log.append("browser: playwright chromium install complete")
return p.chromium.launch(**launch_kwargs)
def _system_chromium_path() -> str | None:
for candidate in ("chromium", "chromium-browser", "google-chrome", "google-chrome-stable"):
path = shutil.which(candidate)
if path:
return path
for path in ("/usr/bin/chromium", "/usr/bin/chromium-browser", "/usr/bin/google-chrome"):
if Path(path).exists():
return path
return None
def _fill_search_inputs(page: Any, search_term: str | None, log: list[str]) -> None:
if not search_term:
return
selectors = [
"input[type='search']",
"input[name*='search' i]",
"input[placeholder*='search' i]",
"input[type='text']",
]
for selector in selectors:
try:
locator = page.locator(selector).first
if locator.count() > 0 and locator.is_visible(timeout=1000):
locator.fill(search_term, timeout=2000)
locator.press("Enter", timeout=2000)
page.wait_for_timeout(2000)
log.append(f"interaction: filled {selector}")
return
except Exception:
continue
def _auto_scroll(page: Any, log: list[str]) -> None:
try:
for _ in range(3):
page.mouse.wheel(0, 900)
page.wait_for_timeout(700)
page.mouse.wheel(0, -400)
log.append("interaction: auto-scroll")
except Exception as exc: # noqa: BLE001
log.append(f"scroll warning: {type(exc).__name__}: {exc}")
def _click_interesting_elements(page: Any, *, max_clicks: int, log: list[str]) -> None:
if max_clicks <= 0:
return
selectors = [
"button:visible",
"[role='button']:visible",
"a[href]:visible",
]
clicked = 0
for selector in selectors:
if clicked >= max_clicks:
break
try:
count = min(page.locator(selector).count(), max_clicks * 3)
except Exception:
continue
for index in range(count):
if clicked >= max_clicks:
break
try:
item = page.locator(selector).nth(index)
text = (item.inner_text(timeout=800) or "").strip().lower()
href = item.get_attribute("href", timeout=800) if selector.startswith("a") else None
if href and href.startswith(("mailto:", "tel:", "#")):
continue
if href and urlparse(href).netloc and urlparse(href).netloc != urlparse(page.url).netloc:
continue
if _looks_destructive(text):
continue
item.click(timeout=2000)
page.wait_for_timeout(1600)
clicked += 1
log.append(f"interaction: clicked {selector} #{index + 1}")
except Exception:
continue
def _looks_destructive(text: str) -> bool:
return any(word in text for word in ("delete", "remove", "logout", "sign out", "purchase", "checkout", "pay"))
def parse_trace_jsonl(*, requests_jsonl: str, responses_jsonl: str = "") -> list[TrafficSample]:
request_rows = [_json_loads(line) for line in requests_jsonl.splitlines() if line.strip()]
response_rows = [_json_loads(line) for line in responses_jsonl.splitlines() if line.strip()]
response_by_id = {
str(row.get("requestId") or row.get("id") or row.get("request_id")): row
for row in response_rows
if isinstance(row, dict)
}
samples: list[TrafficSample] = []
for index, row in enumerate(row for row in request_rows if isinstance(row, dict)):
params = row.get("params") if isinstance(row.get("params"), dict) else row
req = params.get("request") if isinstance(params.get("request"), dict) else params
request_id = str(params.get("requestId") or row.get("requestId") or row.get("id") or index)
response = response_by_id.get(request_id, {})
response_params = response.get("params") if isinstance(response.get("params"), dict) else response
response_obj = response_params.get("response") if isinstance(response_params.get("response"), dict) else response_params
samples.append(
TrafficSample(
request_id=request_id,
method=str(req.get("method") or "GET").upper(),
url=str(req.get("url") or params.get("url") or ""),
status=_int_or_none(response_obj.get("status")),
resource_type=params.get("type") or params.get("resourceType"),
request_headers=_dict_or_empty(req.get("headers")),
response_headers=_dict_or_empty(response_obj.get("headers")),
request_body=_parse_maybe_json(req.get("postData") or req.get("body")),
response_body=_parse_maybe_json(response_obj.get("body") or response_params.get("body")),
content_type=_content_type(response_obj.get("headers")),
timestamp=float(params.get("timestamp") or time.time()),
)
)
return [sample for sample in samples if sample.url]
def build_openapi_bundle(
samples: list[TrafficSample],
*,
title: str,
origins: list[str] | None = None,
include: list[str] | None = None,
exclude: list[str] | None = None,
min_samples: int = 1,
redact: list[str] | None = None,
) -> dict[str, Any]:
redactions = set(DEFAULT_REDACT_KEYS)
redactions.update(_normalize_redact_key(key) for key in (redact or []))
filtered = [
_redact_sample(sample, redactions)
for sample in samples
if _include_sample(sample, origins=origins, include=include, exclude=exclude)
]
groups = group_samples(filtered)
openapi = emit_openapi(title=title, groups=groups, min_samples=min_samples)
confidence = emit_confidence(groups, min_samples=min_samples)
report = emit_markdown_report(title=title, groups=groups, confidence=confidence, min_samples=min_samples)
client = emit_client(openapi)
html_report = emit_html_report(title=title, markdown=report, confidence=confidence)
summary = summarize(groups, filtered, min_samples=min_samples)
return {
"openapi": openapi,
"confidence": confidence,
"report": report,
"html": html_report,
"client": client,
"samples": emit_samples(groups, min_samples=min_samples),
"summary": summary,
}
def _include_sample(
sample: TrafficSample,
*,
origins: list[str] | None,
include: list[str] | None,
exclude: list[str] | None,
) -> bool:
if not sample.url.startswith(("http://", "https://")):
return False
parsed = urlparse(sample.url)
origin = _origin(parsed)
if origins and not any(_origin_matches(origin, candidate) for candidate in origins):
return False
resource_type = (sample.resource_type or "").lower()
if resource_type and resource_type not in {"xhr", "fetch", "document"} and not _looks_like_api_url(sample.url):
return False
all_excludes = NOISE_URL_PATTERNS + list(exclude or [])
if any(_regex_search(pattern, sample.url) for pattern in all_excludes):
if not include or not any(_regex_search(pattern, sample.url) for pattern in include):
return False
if include and not any(_regex_search(pattern, sample.url) for pattern in include):
return False
content_type = sample.content_type or _content_type(sample.response_headers)
if sample.method == "GET" and content_type and "text/html" in content_type and not _looks_like_api_url(sample.url):
return False
return True
def group_samples(samples: list[TrafficSample]) -> list[EndpointGroup]:
groups: dict[str, EndpointGroup] = {}
for sample in samples:
parsed = urlparse(sample.url)
origin = _origin(parsed)
path, path_params = template_path(parsed.path or "/")
dispatch = _dispatch_name(sample, parsed)
display_path = path if not dispatch else f"{path}:{safe_operation_token(dispatch)}"
key = f"{sample.method} {origin}{display_path}"
group = groups.get(key)
if not group:
group = EndpointGroup(
key=key,
origin=origin,
method=sample.method.upper(),
path=display_path,
source_path=path,
dispatch=dispatch,
path_params=path_params,
)
groups[key] = group
group.samples.append(sample)
query = dict(parse_qsl(parsed.query, keep_blank_values=True))
for name, value in query.items():
group.query_params.setdefault(name, []).append(value)
for group in groups.values():
if len(group.samples) == 1:
group.flags.append("single-sample")
statuses = {sample.status for sample in group.samples if sample.status}
if len(statuses) <= 1:
group.flags.append("single-status")
content_types = {_content_type(sample.response_headers) for sample in group.samples if _content_type(sample.response_headers)}
if len(content_types) > 1:
group.flags.append("mixed-content-types")
if group.dispatch:
group.flags.append("dispatch-split")
return sorted(groups.values(), key=lambda group: (-len(group.samples), group.key))
def emit_openapi(*, title: str, groups: list[EndpointGroup], min_samples: int) -> dict[str, Any]:
servers = sorted({group.origin for group in groups if len(group.samples) >= min_samples})
spec: dict[str, Any] = {
"openapi": "3.1.0",
"info": {
"title": title,
"version": "0.1.0-discovered",
"description": "Generated from observed browser traffic. Coverage is bounded by captured flows.",
},
"servers": [{"url": server} for server in servers],
"paths": {},
"components": {"schemas": {}},
}
schema_components: dict[str, Any] = {}
seen_schema_hashes: dict[str, str] = {}
for group in groups:
if len(group.samples) < min_samples:
continue
method = group.method.lower()
path_item = spec["paths"].setdefault(group.path, {})
request_bodies = [sample.request_body for sample in group.samples if sample.request_body is not None]
response_bodies = [sample.response_body for sample in group.samples if sample.response_body is not None]
operation_id = operation_id_for(group)
parameters = list(group.path_params)
parameters.extend(query_parameters_for(group))
responses: dict[str, Any] = {}
for status in sorted({sample.status or 0 for sample in group.samples}):
status_key = str(status or "default")
matching = [sample.response_body for sample in group.samples if (sample.status or 0) == status and sample.response_body is not None]
response_schema = _schema_ref(infer_schema(matching or response_bodies), schema_components, seen_schema_hashes, operation_id + "Response")
responses[status_key] = {
"description": f"Observed {status_key} response",
"content": {"application/json": {"schema": response_schema}} if matching or response_bodies else {},
}
if not responses:
responses["default"] = {"description": "Observed response", "content": {}}
operation: dict[str, Any] = {
"operationId": operation_id,
"summary": f"{group.method} {group.source_path}",
"parameters": parameters,
"responses": responses,
"x-origin": group.origin,
"x-source-path": group.source_path,
"x-sample-count": len(group.samples),
"x-confidence": confidence_for(group),
}
auth_headers = observed_auth_headers(group)
if auth_headers:
operation["x-observed-auth"] = auth_headers
if group.dispatch:
operation["x-dispatch-key"] = group.dispatch
if request_bodies:
request_schema = _schema_ref(infer_schema(request_bodies), schema_components, seen_schema_hashes, operation_id + "Request")
operation["requestBody"] = {
"required": True,
"content": {"application/json": {"schema": request_schema}},
}
path_item[method] = operation
spec["components"]["schemas"] = schema_components
return spec
def query_parameters_for(group: EndpointGroup) -> list[dict[str, Any]]:
total = max(len(group.samples), 1)
params = []
for name, values in sorted(group.query_params.items()):
params.append(
{
"name": name,
"in": "query",
"required": len(values) == total,
"schema": infer_schema(values),
}
)
return params
def emit_confidence(groups: list[EndpointGroup], *, min_samples: int) -> dict[str, Any]:
endpoints = []
for group in groups:
response_body_known = any(sample.response_body is not None for sample in group.samples)
request_body_known = any(sample.request_body is not None for sample in group.samples)
statuses = sorted({sample.status for sample in group.samples if sample.status})
endpoints.append(
{
"key": group.key,
"method": group.method,
"origin": group.origin,
"path": group.path,
"sourcePath": group.source_path,
"dispatch": group.dispatch,
"samples": len(group.samples),
"included": len(group.samples) >= min_samples,
"statusCodes": statuses,
"requestBodyKnown": request_body_known,
"responseBodyKnown": response_body_known,
"normalizationFlags": group.flags,
"confidence": confidence_for(group)["level"],
}
)
return {"endpoints": endpoints}
def emit_markdown_report(
*,
title: str,
groups: list[EndpointGroup],
confidence: dict[str, Any],
min_samples: int,
) -> str:
lines = [
f"# {title}",
"",
"Generated from observed browser traffic. This is an inductive API map, not a vendor contract.",
"",
f"- Observed endpoints: {len(groups)}",
f"- Included endpoints: {sum(1 for group in groups if len(group.samples) >= min_samples)}",
f"- Minimum samples: {min_samples}",
"",
"## Endpoints",
"",
]
confidence_by_key = {entry["key"]: entry for entry in confidence["endpoints"]}
for group in groups:
entry = confidence_by_key[group.key]
included = "included" if entry["included"] else "below min_samples"
lines.extend(
[
f"### `{group.method} {group.path}`",
"",
f"- Origin: `{group.origin}`",
f"- Samples: {len(group.samples)} ({included})",
f"- Confidence: `{entry['confidence']}`",
f"- Statuses: `{', '.join(map(str, entry['statusCodes'])) or 'unknown'}`",
f"- Flags: `{', '.join(group.flags) or 'none'}`",
"",
"```bash",
curl_example(group),
"```",
"",
]
)
lines.extend(
[
"## Coverage Notes",
"",
"- Exercise more logged-in flows to reveal endpoints behind auth-gated UI.",
"- Run with `origins` restricted to the API host when marketing/vendor noise dominates.",
"- Raise `min_samples` to keep only endpoints seen repeatedly.",
"- Response schemas are strongest when the browser can read JSON response bodies.",
"",
]
)
return "\n".join(lines)
def emit_html_report(*, title: str, markdown: str, confidence: dict[str, Any]) -> str:
rows = []
for endpoint in confidence["endpoints"]:
rows.append(
""
f"{html.escape(endpoint['method'])} | "
f"{html.escape(endpoint['path'])} | "
f"{html.escape(endpoint['confidence'])} | "
f"{endpoint['samples']} | "
f"{'yes' if endpoint['responseBodyKnown'] else 'no'} | "
f"{html.escape(', '.join(endpoint['normalizationFlags']) or 'none')} | "
"
"
)
return f"""
{html.escape(title)} API discovery
{html.escape(title)} API discovery
OpenAPI, client, samples, and confidence files were generated next to this report.
| Method | Path | Confidence | Samples | Response body | Flags |
|---|
{''.join(rows)}
Markdown Report
{html.escape(markdown)}
"""
def emit_client(openapi: dict[str, Any]) -> str:
functions = []
for path, item in sorted(openapi.get("paths", {}).items()):
for method, operation in sorted(item.items()):
name = operation.get("operationId") or safe_operation_token(f"{method}_{path}")
functions.append(
f"""export async function {name}(baseUrl, options = {{}}) {{
const params = options.params || {{}};
let path = {json.dumps(path)};
for (const [key, value] of Object.entries(params.path || {{}})) {{
path = path.replace(`{{${{key}}}}`, encodeURIComponent(String(value)));
}}
const query = new URLSearchParams(params.query || {{}}).toString();
const response = await fetch(`${{baseUrl}}${{path}}${{query ? `?${{query}}` : ""}}`, {{
method: {json.dumps(method.upper())},
headers: {{ "content-type": "application/json", ...(options.headers || {{}}) }},
body: options.body === undefined ? undefined : JSON.stringify(options.body),
}});
const text = await response.text();
return text ? JSON.parse(text) : null;
}}"""
)
return "\n\n".join(functions) + "\n"
def emit_samples(groups: list[EndpointGroup], *, min_samples: int) -> dict[str, Any]:
output = {}
for group in groups:
if len(group.samples) < min_samples:
continue
sample = next((item for item in reversed(group.samples) if item.status and 200 <= item.status < 300), group.samples[-1])
output[operation_id_for(group)] = {
"method": sample.method,
"url": sample.url,
"status": sample.status,
"requestHeaders": sample.request_headers,
"requestBody": sample.request_body,
"responseHeaders": sample.response_headers,
"responseBody": sample.response_body,
}
return output
def summarize(groups: list[EndpointGroup], samples: list[TrafficSample], *, min_samples: int) -> dict[str, Any]:
included = [group for group in groups if len(group.samples) >= min_samples]
return {
"included_samples": len(samples),
"endpoints": len(included),
"origins": sorted({group.origin for group in included}),
"top_endpoints": [
{
"method": group.method,
"path": group.path,
"origin": group.origin,
"samples": len(group.samples),
"confidence": confidence_for(group)["level"],
}
for group in included[:10]
],
}
def write_bundle(bundle: dict[str, Any], output_dir: Path) -> Path:
output_dir.mkdir(parents=True, exist_ok=True)
(output_dir / "samples").mkdir(exist_ok=True)
(output_dir / "openapi.json").write_text(json.dumps(bundle["openapi"], indent=2, sort_keys=True), encoding="utf-8")
(output_dir / "openapi.yaml").write_text(to_yaml(bundle["openapi"]), encoding="utf-8")
(output_dir / "confidence.json").write_text(json.dumps(bundle["confidence"], indent=2, sort_keys=True), encoding="utf-8")
(output_dir / "report.md").write_text(bundle["report"], encoding="utf-8")
(output_dir / "index.html").write_text(bundle["html"], encoding="utf-8")
(output_dir / "client.mjs").write_text(bundle["client"], encoding="utf-8")
for name, sample in bundle["samples"].items():
(output_dir / "samples" / f"{name}.json").write_text(json.dumps(sample, indent=2, sort_keys=True), encoding="utf-8")
return output_dir
def _result_payload(
*,
samples: list[TrafficSample],
bundle: dict[str, Any],
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,
ai_explore: bool | None = None,
ai_steps: int | None = None,
) -> dict[str, Any]:
bot_signals = _bot_defense_signals(samples)
payload: dict[str, Any] = {
"captured_samples": len(samples),
"included_samples": bundle["summary"]["included_samples"],
"endpoints": bundle["summary"]["endpoints"],
"origins": bundle["summary"]["origins"],
"output_dir": str(output_dir),
"openapi_json": str(output_dir / "openapi.json"),
"openapi_yaml": str(output_dir / "openapi.yaml"),
"html_report": str(output_dir / "index.html"),
"markdown_report": str(output_dir / "report.md"),
"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 ai_explore is not None:
payload["ai_explore"] = ai_explore
if ai_steps is not None:
payload["ai_steps"] = ai_steps
if capture_log is not None:
payload["capture_log"] = capture_log
return payload
def _bot_defense_signals(samples: list[TrafficSample]) -> 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)
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:
return {}
if all(isinstance(value, dict) for value in values):
keys = sorted({key for value in values for key in value})
required = [key for key in keys if all(key in value for value in values)]
properties = {key: infer_schema([value.get(key) for value in values if key in value]) for key in keys}
schema: dict[str, Any] = {"type": "object", "properties": properties}
if required:
schema["required"] = required
return schema
if all(isinstance(value, list) for value in values):
items = [item for value in values for item in value]
return {"type": "array", "items": infer_schema(items)}
type_names = sorted({_json_type(value) for value in values})
schema = {"type": type_names[0] if len(type_names) == 1 else type_names}
if type_names == ["string"]:
formats = {_format_hint(str(value)) for value in values}
formats.discard(None)
if len(formats) == 1:
schema["format"] = formats.pop()
distinct = sorted({str(value) for value in values})
if 1 < len(distinct) <= 8 and len(values) >= 5:
schema["enum"] = distinct
return schema
def _schema_ref(schema: dict[str, Any], components: dict[str, Any], seen: dict[str, str], name_hint: str) -> dict[str, Any]:
if not schema:
return {}
body = json.dumps(schema, sort_keys=True)
digest = hashlib.sha1(body.encode("utf-8")).hexdigest()[:12]
if digest in seen:
return {"$ref": f"#/components/schemas/{seen[digest]}"}
name = safe_operation_token(name_hint)
if not name[:1].isalpha():
name = f"Schema{name}"
candidate = name
suffix = 2
while candidate in components:
candidate = f"{name}{suffix}"
suffix += 1
components[candidate] = schema
seen[digest] = candidate
return {"$ref": f"#/components/schemas/{candidate}"}
def template_path(path: str) -> tuple[str, list[dict[str, Any]]]:
params: list[dict[str, Any]] = []
counters: dict[str, int] = {}
output = []
for segment in path.split("/"):
if not segment:
continue
replacement, schema = _segment_template(segment)
if replacement:
counters[replacement] = counters.get(replacement, 0) + 1
name = replacement if counters[replacement] == 1 else f"{replacement}{counters[replacement]}"
output.append(f"{{{name}}}")
params.append({"name": name, "in": "path", "required": True, "schema": schema})
else:
output.append(segment)
return "/" + "/".join(output), params
def _segment_template(segment: str) -> tuple[str | None, dict[str, Any]]:
if re.fullmatch(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}", segment):
return "id", {"type": "string", "format": "uuid"}
if re.fullmatch(r"\d+", segment):
return "id", {"type": "integer"}
if re.fullmatch(r"[A-Za-z0-9_-]{12,}", segment):
return "id", {"type": "string"}
return None, {}
def _dispatch_name(sample: TrafficSample, parsed: Any) -> str | None:
body = sample.request_body
if isinstance(body, dict):
for key in ("operationName", "method", "action", "op", "opname"):
value = body.get(key)
if isinstance(value, str) and value.strip():
return value.strip()
query = dict(parse_qsl(parsed.query, keep_blank_values=True))
for key in ("operationName", "method", "action", "op", "opname"):
value = query.get(key)
if value:
return value
return None
def operation_id_for(group: EndpointGroup) -> str:
raw = f"{group.method}_{group.path}".replace("{", "by_").replace("}", "")
return safe_operation_token(raw)
def safe_operation_token(value: str) -> str:
token = re.sub(r"[^A-Za-z0-9_]+", "_", value).strip("_")
if not token:
token = "operation"
if token[0].isdigit():
token = f"op_{token}"
return token[:96]
def confidence_for(group: EndpointGroup) -> dict[str, Any]:
samples = len(group.samples)
statuses = sorted({sample.status for sample in group.samples if sample.status})
if samples >= 10 and len(statuses) > 1 and not group.flags:
level = "high"
elif samples >= 3 and not any(flag in group.flags for flag in ("mixed-content-types",)):
level = "medium"
else:
level = "low"
return {"level": level, "samples": samples, "statusCodes": statuses, "normalizationFlags": group.flags}
def observed_auth_headers(group: EndpointGroup) -> list[str]:
found = set()
for sample in group.samples:
for name in sample.request_headers:
clean = name.lower()
if clean in DEFAULT_REDACT_KEYS or "token" in clean or "secret" in clean or "signature" in clean:
found.add(clean)
return sorted(found)
def curl_example(group: EndpointGroup) -> str:
sample = next((item for item in reversed(group.samples) if item.status and 200 <= item.status < 300), group.samples[-1])
parts = ["curl", "-X", sample.method, json.dumps(sample.url)]
if sample.request_body is not None:
parts.extend(["-H", json.dumps("content-type: application/json"), "--data", json.dumps(json.dumps(sample.request_body))])
return " ".join(parts)
def to_yaml(value: Any, indent: int = 0) -> str:
spaces = " " * indent
if isinstance(value, dict):
lines = []
for key, item in value.items():
if isinstance(item, (dict, list)):
lines.append(f"{spaces}{key}:")
lines.append(to_yaml(item, indent + 2).rstrip())
else:
lines.append(f"{spaces}{key}: {yaml_scalar(item)}")
return "\n".join(lines) + "\n"
if isinstance(value, list):
if not value:
return f"{spaces}[]\n"
lines = []
for item in value:
if isinstance(item, (dict, list)):
lines.append(f"{spaces}-")
lines.append(to_yaml(item, indent + 2).rstrip())
else:
lines.append(f"{spaces}- {yaml_scalar(item)}")
return "\n".join(lines) + "\n"
return f"{spaces}{yaml_scalar(value)}\n"
def yaml_scalar(value: Any) -> str:
if value is None:
return "null"
if value is True:
return "true"
if value is False:
return "false"
if isinstance(value, (int, float)):
return str(value)
text = str(value)
if not text or any(ch in text for ch in ":{}[]#,&*?|<>=!%@`\\n") or text.strip() != text:
return json.dumps(text)
return text
def _redact_sample(sample: TrafficSample, redactions: set[str]) -> TrafficSample:
data = asdict(sample)
data["request_headers"] = _redact_value(data["request_headers"], redactions)
data["response_headers"] = _redact_value(data["response_headers"], redactions)
data["request_body"] = _redact_value(data["request_body"], redactions)
data["response_body"] = _redact_value(data["response_body"], redactions)
return TrafficSample(**data)
def _redact_value(value: Any, redactions: set[str]) -> Any:
if isinstance(value, dict):
output = {}
for key, item in value.items():
clean = _normalize_redact_key(str(key))
output[key] = "" if clean in redactions or "token" in clean or "secret" in clean or "signature" in clean else _redact_value(item, redactions)
return output
if isinstance(value, list):
return [_redact_value(item, redactions) for item in value]
if isinstance(value, str):
if re.fullmatch(r"eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+", value):
return ""
if re.search(r"[^@\s]+@[^@\s]+\.[^@\s]+", value):
return ""
return value
def _parse_maybe_json(value: Any) -> Any:
if value is None or isinstance(value, (dict, list, int, float, bool)):
return value
text = str(value)
if not text:
return None
try:
return json.loads(text)
except Exception:
return text[:100_000]
def _parse_json_object(value: Any) -> dict[str, Any] | None:
if isinstance(value, dict):
return value
if not isinstance(value, str):
return None
text = value.strip()
if not text:
return None
parsed = _try_json_object(text)
if parsed is not None:
return parsed
fenced = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, flags=re.DOTALL | re.IGNORECASE)
if fenced:
parsed = _try_json_object(fenced.group(1))
if parsed is not None:
return parsed
start = text.find("{")
end = text.rfind("}")
if start >= 0 and end > start:
return _try_json_object(text[start : end + 1])
return None
def _try_json_object(text: str) -> dict[str, Any] | None:
try:
parsed = json.loads(text)
except Exception:
return None
return parsed if isinstance(parsed, dict) else None
def _truncate_text(value: Any, limit: int = 240) -> str:
text = str(value or "").replace("\n", " ").strip()
if len(text) <= limit:
return text
return f"{text[: max(0, limit - 3)]}..."
def _json_loads(line: str) -> Any:
try:
return json.loads(line)
except Exception:
return None
def _json_type(value: Any) -> str:
if isinstance(value, bool):
return "boolean"
if isinstance(value, int) and not isinstance(value, bool):
return "integer"
if isinstance(value, float):
return "number"
if isinstance(value, str):
return "string"
if value is None:
return "null"
if isinstance(value, list):
return "array"
if isinstance(value, dict):
return "object"
return "string"
def _format_hint(value: str) -> str | None:
if re.fullmatch(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}", value):
return "uuid"
if re.fullmatch(r"\d{4}-\d{2}-\d{2}T.+", value):
return "date-time"
if value.startswith(("http://", "https://")):
return "uri"
if re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", value):
return "email"
return None
def _is_body_worth_capturing(content_type: str | None) -> bool:
if not content_type:
return False
clean = content_type.lower()
return any(kind in clean for kind in JSONISH_TYPES + TEXT_TYPES)
def _content_type(headers: Any) -> str | None:
if not isinstance(headers, dict):
return None
for key, value in headers.items():
if str(key).lower() == "content-type":
return str(value).split(";")[0].strip().lower()
return None
def _dict_or_empty(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _int_or_none(value: Any) -> int | None:
try:
return int(value)
except Exception:
return None
def _origin(parsed: Any) -> str:
return f"{parsed.scheme}://{parsed.netloc}"
def _origin_matches(origin: str, candidate: str) -> bool:
if candidate.startswith(("http://", "https://")):
return origin == candidate.rstrip("/")
return urlparse(origin).netloc == candidate.strip().strip("/")
def _looks_like_api_url(url: str) -> bool:
parsed = urlparse(url)
path = parsed.path.lower()
return any(marker in path for marker in ("/api/", "/graphql", "/gql", "/rpc", "/v1/", "/v2/", "/v3/")) or "json" in path
def _regex_search(pattern: str, value: str) -> bool:
try:
return re.search(pattern, value, re.IGNORECASE) is not None
except re.error:
return False
def _normalize_redact_key(value: str) -> str:
return re.sub(r"[^a-z0-9]+", "_", value.lower()).strip("_")
def _title_from_url(url: str) -> str:
host = urlparse(url).netloc or "Website"
return f"{host} Discovered API"
def safe_run_id(value: str) -> str:
stamp = time.strftime("%Y%m%d-%H%M%S")
token = re.sub(r"[^A-Za-z0-9]+", "-", value.lower()).strip("-")[:48] or "run"
return f"{stamp}-{token}"
def main() -> None:
parser = argparse.ArgumentParser(description="Discover an API from browser traffic.")
parser.add_argument("url")
parser.add_argument("--title")
parser.add_argument("--origin", action="append", dest="origins")
parser.add_argument("--wait-seconds", type=int, default=8)
parser.add_argument("--max-clicks", type=int, default=4)
args = parser.parse_args()
samples, log = capture_browser_traffic(url=args.url, wait_seconds=args.wait_seconds, max_clicks=args.max_clicks)
bundle = build_openapi_bundle(samples, title=args.title or _title_from_url(args.url), origins=args.origins)
output_dir = write_bundle(bundle, DEFAULT_OUTPUT_DIR / safe_run_id(args.url))
print(json.dumps({"output_dir": str(output_dir), "summary": bundle["summary"], "capture_log": log}, indent=2))
if __name__ == "__main__":
main()