1176 lines
45 KiB
Python
1176 lines
45 KiB
Python
"""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 re
|
|
import time
|
|
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("outputs/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"/(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(\?|$)",
|
|
]
|
|
|
|
|
|
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.0"
|
|
|
|
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.",
|
|
}
|
|
}
|
|
|
|
@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,
|
|
) -> dict[str, Any]:
|
|
if not url or not str(url).strip().startswith(("http://", "https://")):
|
|
return {"error": "url must start with http:// or https://"}
|
|
|
|
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,
|
|
)
|
|
|
|
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 {
|
|
"url": url,
|
|
"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"),
|
|
"capture_log": capture_log,
|
|
"top_endpoints": bundle["summary"]["top_endpoints"],
|
|
}
|
|
|
|
@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]:
|
|
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 {
|
|
"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"],
|
|
}
|
|
|
|
|
|
def capture_browser_traffic(
|
|
*,
|
|
url: str,
|
|
wait_seconds: int = 8,
|
|
max_clicks: int = 4,
|
|
search_term: str | 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] = {}
|
|
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch(
|
|
headless=True,
|
|
args=[
|
|
"--disable-dev-shm-usage",
|
|
"--disable-blink-features=AutomationControlled",
|
|
"--no-sandbox",
|
|
],
|
|
)
|
|
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"
|
|
),
|
|
)
|
|
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)
|
|
_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 _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(
|
|
"<tr>"
|
|
f"<td><code>{html.escape(endpoint['method'])}</code></td>"
|
|
f"<td><code>{html.escape(endpoint['path'])}</code></td>"
|
|
f"<td>{html.escape(endpoint['confidence'])}</td>"
|
|
f"<td>{endpoint['samples']}</td>"
|
|
f"<td>{'yes' if endpoint['responseBodyKnown'] else 'no'}</td>"
|
|
f"<td>{html.escape(', '.join(endpoint['normalizationFlags']) or 'none')}</td>"
|
|
"</tr>"
|
|
)
|
|
return f"""<!doctype html>
|
|
<html lang=\"en\">
|
|
<head>
|
|
<meta charset=\"utf-8\" />
|
|
<meta name=\"viewport\" content=\"width=device-width,initial-scale=1\" />
|
|
<title>{html.escape(title)} API discovery</title>
|
|
<style>
|
|
body {{ font-family: ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; margin: 32px; color: #0b1f1a; background: #f7faf8; }}
|
|
h1 {{ font-size: 28px; margin-bottom: 8px; }}
|
|
table {{ border-collapse: collapse; width: 100%; background: white; }}
|
|
th, td {{ border: 1px solid #d7e3dc; padding: 10px; text-align: left; vertical-align: top; }}
|
|
th {{ background: #eef6f1; }}
|
|
pre {{ background: #0b1f1a; color: #d7ffe5; padding: 16px; overflow: auto; }}
|
|
code {{ font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>{html.escape(title)} API discovery</h1>
|
|
<p>OpenAPI, client, samples, and confidence files were generated next to this report.</p>
|
|
<table>
|
|
<thead><tr><th>Method</th><th>Path</th><th>Confidence</th><th>Samples</th><th>Response body</th><th>Flags</th></tr></thead>
|
|
<tbody>{''.join(rows)}</tbody>
|
|
</table>
|
|
<h2>Markdown Report</h2>
|
|
<pre>{html.escape(markdown)}</pre>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
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 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 _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()
|