diff --git a/.env.example b/.env.example index f1c3489..df60fee 100644 --- a/.env.example +++ b/.env.example @@ -22,10 +22,15 @@ CIBERWEBSCAN_HTTP_TIMEOUT_READ=30.0 # -> http.timeout.read CIBERWEBSCAN_HTTP_PROXY_ROTATE=false # -> http.proxy.rotate CIBERWEBSCAN_HTTP_RETRY_MAX_ATTEMPTS=3 # -> http.retry.max_attempts CIBERWEBSCAN_HTTP_RATE_LIMIT_ADAPTIVE=true # -> http.rate_limit.adaptive -# Use standard proxy env vars for proxy URLs -HTTP_PROXY=http://proxy.example.com:3128 -HTTPS_PROXY=http://proxy.example.com:3128 -NO_PROXY=localhost,127.0.0.1 +# Permit loopback/RFC1918 proxy endpoints (e.g. local Burp). Independent +# from the *_ALLOW_LOCAL target opt-ins; IMDS/link-local stays blocked. +CIBERWEBSCAN_HTTP_PROXY_ALLOW_LOCAL=false # -> http.proxy.allow_local +# Standard proxy env vars are IGNORED unless use_env is enabled; when +# enabled, every env proxy is SSRF-validated before use (fail-closed). +CIBERWEBSCAN_HTTP_PROXY_USE_ENV=false # -> http.proxy.use_env +# HTTP_PROXY=http://proxy.example.com:3128 +# HTTPS_PROXY=http://proxy.example.com:3128 +# NO_PROXY=localhost,127.0.0.1 # Playwright / Scraping CIBERWEBSCAN_SCRAPING_DYNAMIC_ENABLED=false # -> scraping.dynamic.enabled diff --git a/docs/API.md b/docs/API.md index dd233bc..f6ba318 100644 --- a/docs/API.md +++ b/docs/API.md @@ -819,6 +819,33 @@ Update a single configuration value. } ``` +**Protected (operator-only) keys — HTTP 403:** + +Security-sensitive configuration cannot be modified through this +endpoint, regardless of the API key used (keys carry no administrative +scope distinction, so no key is trusted with security exceptions): + +- `http.proxy.*` — every proxy option (`http`, `https`, `socks5`, + `rotate`, `rotation_interval`, `proxy_list`, `allow_local`, `use_env`) +- Any `*.allow_local` security exception (`analysis.allow_local`, + `scraping.allow_local`, `attack.allow_local`) + +The check is case-insensitive and also covers section/batch updates that +embed protected keys inside the `value` object (e.g. `path: "http"` with +`{"proxy": {"allow_local": true}}`). Rejected requests return +`403 Forbidden` and change nothing. The local operator can still change +these settings through the configuration file, environment variables or +the CLI (`ciberwebscan config set ...`), which do not pass through this +endpoint. `POST /api/config/reset` remains allowed because it restores +the safe defaults. + +Additionally, every `proxy` field accepted by the scan endpoints +(`/api/analyze`, `/api/attack`, `/api/scrape`, `/api/quick`) is +validated against the SSRF proxy policy before any connection: local, +private, link-local (IMDS) and other forbidden proxy endpoints are +rejected with a validation-class error (`VALIDATION_ERROR`), never +silently ignored. + #### POST /api/config/export Export the current configuration to a file inside the configuration base diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 9317d73..8423ace 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -71,6 +71,8 @@ Examples that work out of the box: - `CIBERWEBSCAN_ATTACK_ALLOW_LOCAL` → `attack.allow_local` - `CIBERWEBSCAN_ANALYSIS_ALLOW_LOCAL` → `analysis.allow_local` - `CIBERWEBSCAN_SCRAPING_ALLOW_LOCAL` → `scraping.allow_local` +- `CIBERWEBSCAN_HTTP_PROXY_ALLOW_LOCAL` → `http.proxy.allow_local` +- `CIBERWEBSCAN_HTTP_PROXY_USE_ENV` → `http.proxy.use_env` - `CIBERWEBSCAN_ATTACK_USER_CONSENT` → `attack.user_consent` - `CIBERWEBSCAN_USER_AGENT_MODE` / `CIBERWEBSCAN_USER_AGENT_AGENTS` → `user_agent.*` - `CIBERWEBSCAN_EXPORT_INCLUDE_SCREENSHOTS` → `export.include_screenshots` @@ -87,9 +89,9 @@ Examples that work out of the box: Not supported (and why): -- **Whole sections** cannot be set from a single variable — only leaf fields. For example `CIBERWEBSCAN_HTTP_PROXY` targets the nested `http.proxy` model and is ignored with a debug log; set `CIBERWEBSCAN_HTTP_PROXY_HTTP` (→ `http.proxy.http`) or use the standard `HTTP_PROXY`/`HTTPS_PROXY` variables instead. +- **Whole sections** cannot be set from a single variable — only leaf fields. For example `CIBERWEBSCAN_HTTP_PROXY` targets the nested `http.proxy` model and is ignored with a debug log; set `CIBERWEBSCAN_HTTP_PROXY_HTTP` (→ `http.proxy.http`). The standard `HTTP_PROXY`/`HTTPS_PROXY` variables are ignored unless `http.proxy.use_env` is enabled (and are SSRF-validated when honoured). - Unknown variables are ignored with a debug log; they never break configuration loading. -- Invalid values on a valid key log the validation error and the loader falls back to defaults (see `Invalid configuration` error in the logs). +- Invalid values on a valid key log the validation error and the loader falls back to defaults (see `Invalid configuration` error in the logs). This includes proxies that violate the SSRF policy — the effective configuration is the safe default (no proxy), never an unvalidated one. **Note**: Command-line options are specific to individual commands and do not override global configuration. They are used to customize behavior for that particular command execution. @@ -186,7 +188,9 @@ Configure HTTP request behavior. "socks5": null, "rotate": false, "rotation_interval": 10, - "proxy_list": null + "proxy_list": null, + "allow_local": false, + "use_env": false }, "http2": true, "follow_redirects": true, @@ -216,6 +220,8 @@ Configure HTTP request behavior. | `http.rate_limit.latency_window` | `10` | Sliding window size for latency measurements | | `http.proxy.rotate` | `false` | Proxy rotation disabled by default | | `http.proxy.rotation_interval` | `10` | Requests per proxy when rotating | +| `http.proxy.allow_local` | `false` | Permit loopback/RFC1918 proxy endpoints | +| `http.proxy.use_env` | `false` | Honour `HTTP_PROXY`/`HTTPS_PROXY`/`ALL_PROXY` | | `http.http2` | `true` | Enable HTTP/2 by default | | `http.follow_redirects` | `true` | Follow redirects | | `http.max_redirects` | `10` | Max redirects to follow | @@ -260,6 +266,63 @@ Example with rotation enabled: } ``` +#### Proxy SSRF Policy + +Every proxy — regardless of its source (CLI/API parameter, config file, +environment, rotation list or helper client) — is classified against the +central SSRF policy (`ciberwebscan.core.client.ssrf.check_proxy_url`) +**before any connection is opened**: + +- Accepted schemes: `http`, `https` and `socks5` (SOCKS5 is really + supported via `httpx[socks]`/`socksio`, not just accepted syntactically). +- Loopback, RFC1918/ULA, link-local (incl. cloud IMDS `169.254.169.254`), + CGNAT, multicast, reserved and unspecified addresses are blocked **by + default**, including non-canonical IPv4 notations (`2130706433`, + `0x7f000001`, `127.1`, …) and IPv4-mapped IPv6. +- Hostnames are resolved and every resulting address is classified; + resolution failures are treated as blocks (fail-closed). +- Credentials (`http://user:pass@host:port`) are accepted but never + echoed in logs or error messages. + +**`http.proxy.allow_local`** (default `false`) permits loopback and +RFC1918/ULA proxy endpoints, e.g. Burp Suite or mitmproxy on the +operator workstation. It is fully independent from the per-module +`allow_local` target opt-ins (`analysis`/`scraping`/`attack`): neither +authorizes the other. Link-local/IMDS, CGNAT, multicast, reserved and +unspecified proxy addresses **stay blocked even when this flag is +enabled**. + +```bash +# Local Burp proxy through the CLI config (operator-only setting) +ciberwebscan config set http.proxy.allow_local true +ciberwebscan config set http.proxy.http http://127.0.0.1:8080 +``` + +**`http.proxy.use_env`** (default `false`): when `false`, the standard +`HTTP_PROXY`/`HTTPS_PROXY`/`ALL_PROXY` environment variables (and their +lowercase variants) are **ignored** — HTTP clients run with +`trust_env=False`. When `true`, every environment proxy is validated +against the same policy before the client is built; a single forbidden +value fails the client construction (fail-closed) instead of silently +bypassing the policy. + +**Operator-only settings**: `http.proxy.*` (including `allow_local` and +`use_env`) and every `*.allow_local` security exception cannot be +modified through `PUT /api/config` (HTTP 403). Change them through the +configuration file, environment variables or the local CLI. + +**Known residual limits** (not fully closable without custom transports): + +- _DNS rebinding_: the validated hostname is resolved again by the HTTP + library when the connection is opened, so a hostile authoritative DNS + could answer differently between validation and connection. The same + limitation applies to request targets since BUG-004. +- _Playwright redirects_: Chromium follows redirect hops on its own — + Playwright routing treats a request plus its redirects as a single + unit — so intermediate hops are not re-validated; only the final + `page.url` is re-checked. WebSocket handshakes are routed and validated + separately (`route_web_socket`). + ### User Agent Configure user agent rotation. diff --git a/pyproject.toml b/pyproject.toml index 11011b4..4828c7b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,7 +34,7 @@ requires-python = ">=3.10" dependencies = [ "beautifulsoup4==4.15.0", "click==8.5.0", - "httpx[http2]==0.28.1", + "httpx[http2,socks]==0.28.1", "playwright==1.61.0", "lxml==6.1.1", "pyOpenSSL==26.4.0", diff --git a/src/ciberwebscan/api/routes/config.py b/src/ciberwebscan/api/routes/config.py index 9a9bacc..2aed519 100644 --- a/src/ciberwebscan/api/routes/config.py +++ b/src/ciberwebscan/api/routes/config.py @@ -34,6 +34,51 @@ router = APIRouter() +# Configuration subtrees that only the local operator may change (config +# file, environment variables or the local CLI). Remote updates via +# PUT /config are rejected: API keys carry no administrative scope +# distinction today, so no key can be trusted with security exceptions. +_PROTECTED_CONFIG_PREFIXES: tuple[str, ...] = ("http.proxy",) +_PROTECTED_CONFIG_SUFFIXES: tuple[str, ...] = (".allow_local",) + + +def _is_protected_config_key(key: str) -> bool: + """Return True when *key* (dot-notation) is operator-only.""" + normalized = key.strip().strip(".").lower() + if not normalized: + return False + if normalized == "allow_local": + return True + if any( + normalized == prefix or normalized.startswith(f"{prefix}.") + for prefix in _PROTECTED_CONFIG_PREFIXES + ): + return True + return any(normalized.endswith(suffix) for suffix in _PROTECTED_CONFIG_SUFFIXES) + + +def config_update_touches_protected(path: str, value: Any) -> bool: + """Return True when a config update would modify a protected key. + + Covers direct updates (``http.proxy.http``), section updates and batch + updates that smuggle protected keys inside a value mapping + (``http`` with ``{"proxy": {...}}``, or root with + ``{"analysis": {"allow_local": true}}``), case-insensitively. + """ + base = (path or "").strip().strip(".").lower() + if _is_protected_config_key(base): + return True + + def _walk(prefix: str, node: Any) -> bool: + if isinstance(node, dict): + for key, subvalue in node.items(): + full = f"{prefix}.{key}" if prefix else str(key) + if _is_protected_config_key(full) or _walk(full, subvalue): + return True + return False + + return _walk(base, value) + @router.get( "/config", @@ -185,12 +230,31 @@ async def update_config( """ Update a configuration value. + Security-sensitive keys (``http.proxy.*`` and any ``*.allow_local`` + security exception) are operator-only: they cannot be modified + remotely through this endpoint. + Args: request: Contains key path, new value, and optional save flag. Returns: Updated configuration value with metadata. """ + if config_update_touches_protected(request.path, request.value): + logger.warning( + "Rejected remote modification of protected config key %r", + request.path, + ) + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=( + f"Configuration key '{request.path}' is protected: proxy " + "options and security exceptions (allow_local) can only be " + "changed by the local operator via the configuration file, " + "environment variables or the CLI." + ), + ) + try: service = ConfigService() result = service.set(request.path, request.value) diff --git a/src/ciberwebscan/config/models.py b/src/ciberwebscan/config/models.py index a22427a..23e0025 100644 --- a/src/ciberwebscan/config/models.py +++ b/src/ciberwebscan/config/models.py @@ -12,7 +12,7 @@ import logging from typing import Annotated, Literal -from pydantic import BaseModel, Field, HttpUrl, field_validator +from pydantic import BaseModel, Field, HttpUrl, field_validator, model_validator logger = logging.getLogger(__name__) @@ -70,22 +70,91 @@ class ProxyConfig(BaseModel): "Accepts a list of strings or a single comma/newline-separated string." ), ) + allow_local: bool = Field( + default=False, + description=( + "Permit loopback/RFC1918 proxy endpoints (e.g. Burp Suite or " + "mitmproxy on the operator workstation). Independent from the " + "per-module allow_local target opt-ins: neither authorizes the " + "other. Link-local/IMDS, CGNAT, multicast and reserved proxy " + "addresses stay blocked even when true." + ), + ) + use_env: bool = Field( + default=False, + description=( + "Honour HTTP_PROXY/HTTPS_PROXY/ALL_PROXY environment " + "variables. Every environment proxy is validated against the " + "SSRF policy before use (fail-closed)." + ), + ) @field_validator("proxy_list", mode="before") @classmethod def _normalize_proxy_list(cls, value: str | list[str] | None) -> list[str] | None: - """Normalize proxy_list: accept a string or list, always return list or None.""" + """Normalize proxy_list: accept a string or list, always return list or None. + + Purely syntactic (split on commas/newlines, default the scheme): + the SSRF policy is applied afterwards in the model validator, + where ``allow_local`` is already available, so string and list + forms are policed identically. + """ if value is None: return None if isinstance(value, str): - from ciberwebscan.core.client.proxy import parse_proxy_list - - parsed = parse_proxy_list(value) - return parsed if parsed else None + entries = [ + chunk.strip() + for chunk in value.replace("\n", ",").split(",") + if chunk.strip() + ] + normalized = [ + entry + if entry.startswith(("http://", "https://", "socks5://")) + else f"http://{entry}" + for entry in entries + ] + return normalized or None if isinstance(value, list): return [v for v in value if v] or None return value + @model_validator(mode="after") + def _validate_proxies_against_policy(self) -> ProxyConfig: + """Enforce the SSRF policy on every configured proxy at ingestion. + + Strings and lists in ``proxy_list`` go through the same + classification as the individual fields, so a proxy that is + rejected as a string is equally rejected as a list entry. The + effective usage boundary in ``HTTPClient`` re-validates. + """ + from ciberwebscan.core.client.ssrf import ( + SSRFBlockedError, + check_proxy_url, + ) + + candidates: list[tuple[str, str]] = [] + for field_name in ("http", "https"): + value = getattr(self, field_name) + if value is not None: + candidates.append((field_name, str(value))) + if self.socks5: + candidates.append(("socks5", self.socks5)) + for entry in self.proxy_list or []: + candidates.append(("proxy_list", entry)) + + for source, url in candidates: + try: + check_proxy_url(url, allow_local=self.allow_local) + except SSRFBlockedError as exc: + hint = ( + " Set http.proxy.allow_local=true to permit local " + "proxy endpoints (operator decision)." + if not self.allow_local + else "" + ) + raise ValueError(f"http.proxy.{source}: {exc}.{hint}") from exc + return self + class HTTPConfig(BaseModel): """Complete HTTP client configuration.""" diff --git a/src/ciberwebscan/core/client/http_client.py b/src/ciberwebscan/core/client/http_client.py index 9edaa76..90f695d 100644 --- a/src/ciberwebscan/core/client/http_client.py +++ b/src/ciberwebscan/core/client/http_client.py @@ -15,6 +15,7 @@ from __future__ import annotations import logging +import os import random import threading import time @@ -270,6 +271,8 @@ def __init__( max_redirects: int | None = None, ssrf_guard: bool = False, allow_local: bool = False, + allow_local_proxy: bool | None = None, + use_env_proxies: bool | None = None, ): """ Initialize the HTTP client. @@ -298,6 +301,18 @@ def __init__( event hook). Raises SSRFBlockedError for forbidden targets. allow_local: Policy opt-in for ssrf_guard: only validate the URL format and skip private/loopback classification. + allow_local_proxy: Permit loopback/RFC1918 proxy endpoints + (e.g. Burp Suite locally). Independent from ``allow_local`` + (targets): neither authorizes the other. When None, falls + back to ``http.proxy.allow_local`` from config (default + False). Link-local/IMDS, CGNAT, multicast and reserved + proxy addresses stay blocked even when True. + use_env_proxies: Honour ``HTTP_PROXY``/``HTTPS_PROXY``/ + ``ALL_PROXY`` environment variables (httpx ``trust_env``). + Defaults to False: environment proxies are ignored unless + explicitly enabled via this flag or ``http.proxy.use_env``. + When enabled, every environment proxy is validated against + the SSRF policy before the client is built (fail-closed). """ config = get_config().http @@ -346,6 +361,36 @@ def __init__( else config.proxy.socks5 ) + proxy_cfg = config.proxy + resolved_allow_local_proxy = ( + allow_local_proxy + if allow_local_proxy is not None + else bool(proxy_cfg.allow_local if proxy_cfg else False) + ) + resolved_use_env = ( + use_env_proxies + if use_env_proxies is not None + else bool(proxy_cfg.use_env if proxy_cfg else False) + ) + + # SSRF policy on the proxy endpoint: the TCP connection is opened + # towards the proxy, so it is classified here — before any + # transport exists — for every source (CLI/API parameter, config + # fallback, rotation, helper clients). Environment proxies are + # only consulted when httpx would actually use them (no explicit + # or configured proxy) and each one is validated as well, so + # trust_env can never smuggle an unvalidated proxy into a + # transport. Validation failures raise SSRFBlockedError. + from ciberwebscan.core.client.ssrf import ( + check_proxy_url as _check_proxy_url, + ) + + if resolved_proxy is not None: + _check_proxy_url(resolved_proxy, allow_local=resolved_allow_local_proxy) + elif resolved_use_env: + for env_proxy in _environment_proxies().values(): + _check_proxy_url(env_proxy, allow_local=resolved_allow_local_proxy) + self._proxy = resolved_proxy self._max_attempts = resolved_max_attempts self._backoff_factor = resolved_backoff_factor @@ -387,7 +432,9 @@ def _validate_request_target(request: httpx.Request) -> None: event_hooks["request"] = [_validate_request_target] - # Build httpx client + # Build httpx client. trust_env is explicit: environment proxies + # (HTTP_PROXY & co.) are ignored unless use_env_proxies enabled + # them, and in that case every env proxy was validated above. self._client = httpx.Client( timeout=resolved_timeout, http2=resolved_http2, @@ -398,6 +445,7 @@ def _validate_request_target(request: httpx.Request) -> None: cookies=cookies, proxy=resolved_proxy, event_hooks=event_hooks, + trust_env=resolved_use_env, ) def request( @@ -649,6 +697,32 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: _default_client: HTTPClient | None = None _default_client_lock = threading.Lock() +# Environment variables httpx may read proxies from under trust_env. +_ENV_PROXY_VARS: tuple[str, ...] = ( + "HTTP_PROXY", + "http_proxy", + "HTTPS_PROXY", + "https_proxy", + "ALL_PROXY", + "all_proxy", +) + + +def _environment_proxies() -> dict[str, str]: + """Return the proxy URLs declared in the process environment. + + Mirrors the variables httpx honours under ``trust_env`` (both the + uppercase and lowercase spellings, plus ``ALL_PROXY``). ``NO_PROXY`` + only narrows which destinations use a proxy and never introduces a + new endpoint, so it is intentionally not consulted for validation. + """ + proxies: dict[str, str] = {} + for name in _ENV_PROXY_VARS: + value = os.environ.get(name) + if value and value.strip(): + proxies[name] = value.strip() + return proxies + def get_client() -> HTTPClient: """Get or create the default HTTP client (thread-safe singleton).""" diff --git a/src/ciberwebscan/core/client/proxy.py b/src/ciberwebscan/core/client/proxy.py index 6a0f66e..ebb1ab6 100644 --- a/src/ciberwebscan/core/client/proxy.py +++ b/src/ciberwebscan/core/client/proxy.py @@ -10,7 +10,6 @@ import re from dataclasses import dataclass, field from typing import TYPE_CHECKING -from urllib.parse import urlparse if TYPE_CHECKING: from ciberwebscan.core.client.http_client import HTTPClient @@ -107,19 +106,15 @@ def parse_proxy( "Expected: protocol://[user:pass@]host[:port]" ) - parsed = urlparse(proxy_str) - host = parsed.hostname or "" + # Network policy: delegate to the central SSRF classification so + # parsing and the request boundary agree (covers DNS names, + # non-canonical IPv4 notations, link-local/IMDS, IPv6, etc.). + from ciberwebscan.core.client.ssrf import SSRFBlockedError, check_proxy_url - # Check for local/private addresses - if not allow_local and host in {"localhost", "127.0.0.1", "0.0.0.0", "::1"}: - raise ProxyValidationError( - "Local proxies are not allowed. Set allow_local=True to override." - ) - - if not allow_private and is_private_ip(host): - raise ProxyValidationError( - "Private network proxies are not allowed. Set allow_private=True to override." - ) + try: + check_proxy_url(proxy_str, allow_local=allow_local or allow_private) + except SSRFBlockedError as exc: + raise ProxyValidationError(str(exc)) from exc return proxy_str diff --git a/src/ciberwebscan/core/client/ssrf.py b/src/ciberwebscan/core/client/ssrf.py index e1054a3..c26bfd0 100644 --- a/src/ciberwebscan/core/client/ssrf.py +++ b/src/ciberwebscan/core/client/ssrf.py @@ -1,16 +1,28 @@ """ SSRF protection policy. -Single source of truth for deciding whether a target URL may be requested. -Validates scheme/format, resolves hostnames, folds non-canonical IPv4 -notations (``127.1``, ``2130706433``, ``0x7f000001``, ``0177.0.0.1``) and -classifies every resulting address against the local/private/link-local/ -multicast/reserved ranges. +Single source of truth for deciding whether a target URL may be requested +(:func:`check_url`) or a proxy endpoint may be connected to +(:func:`check_proxy_url`). Both validate scheme/format, resolve hostnames, +fold non-canonical IPv4 notations (``127.1``, ``2130706433``, +``0x7f000001``, ``0177.0.0.1``) and classify every resulting address +against the local/private/link-local/multicast/reserved ranges. Used by :func:`ciberwebscan.core.scraping.helpers.is_safe_url`, by -:class:`~ciberwebscan.core.client.http_client.HTTPClient` through a -synchronous request event hook (so every redirect hop is re-validated -before any bytes are sent) and by the service layer. +:class:`~ciberwebscan.core.client.http_client.HTTPClient` (proxy check at +construction time plus a synchronous request event hook so every redirect +hop is re-validated before any bytes are sent), by the Playwright dynamic +scraper (browser context proxy and WebSocket routes) and by the service +layer. + +The two authorizations are deliberately independent: + +- ``allow_local`` on :func:`check_url` permits local/private *targets*. +- ``allow_local`` on :func:`check_proxy_url` permits local/private + *proxies* (e.g. Burp Suite or mitmproxy on the operator workstation). + +Neither grants the other, and proxy ``allow_local`` never permits +link-local/IMDS, multicast, reserved or CGNAT addresses. """ from __future__ import annotations @@ -20,7 +32,7 @@ import socket import threading import time -from urllib.parse import urlparse +from urllib.parse import ParseResult, SplitResult, urlparse, urlsplit logger = logging.getLogger(__name__) @@ -31,6 +43,27 @@ _dns_cache: dict[str, tuple[float, list[str]]] = {} _dns_lock = threading.Lock() +# Schemes accepted for proxy endpoints. Only what httpx (via httpcore/ +# socksio) and the Playwright browser context actually support. +PROXY_ALLOWED_SCHEMES = frozenset({"http", "https", "socks5"}) + +# Ranges that stay blocked even when local proxies are authorized +# (``check_proxy_url(allow_local=True)``): link-local — which includes the +# cloud instance metadata service —, CGNAT shared address space, +# multicast, reserved (incl. IPv4 broadcast) and unspecified addresses. +_ALWAYS_FORBIDDEN_NETWORKS: tuple[ + ipaddress.IPv4Network | ipaddress.IPv6Network, ... +] = ( + ipaddress.ip_network("0.0.0.0/8"), # "this network" / unspecified + ipaddress.ip_network("100.64.0.0/10"), # CGNAT shared address space + ipaddress.ip_network("169.254.0.0/16"), # link-local + cloud IMDS + ipaddress.ip_network("224.0.0.0/4"), # multicast + ipaddress.ip_network("240.0.0.0/4"), # reserved + broadcast + ipaddress.ip_network("::/128"), # unspecified + ipaddress.ip_network("fe80::/10"), # IPv6 link-local + ipaddress.ip_network("ff00::/8"), # IPv6 multicast +) + class SSRFBlockedError(Exception): """A URL was rejected by the SSRF policy. @@ -106,11 +139,32 @@ def _parse_ip_literal( return None -def _address_is_forbidden(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: - """Return True when *ip* must not be contacted under the policy.""" +def _address_is_forbidden( + ip: ipaddress.IPv4Address | ipaddress.IPv6Address, + *, + allow_local: bool = False, +) -> bool: + """Return True when *ip* must not be contacted under the policy. + + Args: + ip: Address to classify (IPv4-mapped IPv6 is unwrapped first). + allow_local: Permit loopback, RFC1918/ULA and globally routable + addresses. Infrastructure-sensitive ranges (link-local/cloud + IMDS, CGNAT, multicast, reserved, unspecified) stay blocked + even when this is True. Mirrors ``ipaddress`` semantics + explicitly so the result does not drift across Python + versions (``is_private`` widened in 3.13). + """ mapped = getattr(ip, "ipv4_mapped", None) if mapped is not None: ip = mapped + + if allow_local: + return any( + ip.version == network.version and ip in network + for network in _ALWAYS_FORBIDDEN_NETWORKS + ) + return bool( ip.is_loopback or ip.is_link_local @@ -121,10 +175,19 @@ def _address_is_forbidden(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> ) -def _check_host(host: str) -> None: - """Raise :class:`SSRFBlockedError` if *host* resolves to a forbidden target.""" +def _check_host(host: str, *, allow_local: bool = False) -> None: + """Raise :class:`SSRFBlockedError` if *host* resolves to a forbidden target. + + Args: + host: Hostname or IP literal to classify. + allow_local: Permit loopback/private addresses (see + :func:`_address_is_forbidden`). ``localhost`` names are then + treated as loopback without a DNS lookup (RFC 6761). + """ if host == "localhost" or host.endswith(".localhost"): # RFC 6761 reserves these names for loopback; no DNS needed. + if allow_local: + return raise SSRFBlockedError( f"SSRF protection: host '{host}' is a local loopback name " "and is blocked by default." @@ -132,7 +195,7 @@ def _check_host(host: str) -> None: literal = _parse_ip_literal(host) if literal is not None: - if _address_is_forbidden(literal): + if _address_is_forbidden(literal, allow_local=allow_local): raise SSRFBlockedError( f"SSRF protection: target '{host}' points to a local or " f"private network address ({literal}) and is blocked by default." @@ -155,7 +218,7 @@ def _check_host(host: str) -> None: f"SSRF protection: host '{host}' resolved to an invalid " f"address ('{address}') and is blocked by default." ) from None - if _address_is_forbidden(resolved): + if _address_is_forbidden(resolved, allow_local=allow_local): raise SSRFBlockedError( f"SSRF protection: host '{host}' resolves to a local or " f"private network address ({address}) and is blocked by default." @@ -199,3 +262,73 @@ def check_url(url: str, *, allow_local: bool = False) -> None: return _check_host(parsed.hostname.lower()) + + +def _display_proxy(parsed: SplitResult | ParseResult) -> str: + """Return *parsed* proxy as ``scheme://host[:port]`` for user-facing + messages, never including credentials.""" + host = parsed.hostname or "" + if ":" in host: # IPv6 literal + host = f"[{host}]" + port = "" + try: + if parsed.port is not None: + port = f":{parsed.port}" + except ValueError: + port = ":" + return f"{parsed.scheme}://{host}{port}" + + +def check_proxy_url(proxy: str, *, allow_local: bool = False) -> None: + """Validate that *proxy* may be connected to under the SSRF policy. + + The proxy endpoint is where the TCP connection is actually opened, so + it must be classified independently of the request target. Callers + (``HTTPClient``, the Playwright dynamic scraper, config ingestion) + invoke this before any transport is created. + + Args: + proxy: Proxy URL (``http://``, ``https://`` or ``socks5://``), + optionally with ``user:pass@`` credentials. Credentials are + accepted but never echoed back in error messages. + allow_local: Permit loopback and RFC1918/ULA proxy endpoints + (e.g. Burp Suite on the operator workstation). Independent + from the ``allow_local`` target opt-in: neither authorizes the + other. Link-local/IMDS, CGNAT, multicast, reserved and + unspecified addresses stay blocked even when True. + + Raises: + SSRFBlockedError: If the URL is malformed, uses an unsupported + scheme, has no host or an invalid port, or its endpoint is + forbidden. Hostname resolution failures are treated as blocks + (fail-closed). Messages mention ``SSRF`` and ``blocked`` so + callers and tests can match on them. + """ + if not proxy or any(char.isspace() for char in proxy): + raise SSRFBlockedError( + "SSRF protection: proxy URL is malformed and is blocked by default." + ) + + try: + parsed = urlsplit(proxy) + host = parsed.hostname + _ = parsed.port # validates the port syntax + except ValueError: + raise SSRFBlockedError( + "SSRF protection: proxy URL is malformed and is blocked by default." + ) from None + + if parsed.scheme.lower() not in PROXY_ALLOWED_SCHEMES: + allowed = ", ".join(sorted(PROXY_ALLOWED_SCHEMES)) + raise SSRFBlockedError( + f"SSRF protection: proxy '{_display_proxy(parsed)}' uses an " + f"unsupported scheme (allowed: {allowed}) and is blocked by default." + ) + + if not host: + raise SSRFBlockedError( + f"SSRF protection: proxy '{_display_proxy(parsed)}' has no host " + "and is blocked by default." + ) + + _check_host(host.lower(), allow_local=allow_local) diff --git a/src/ciberwebscan/core/scraping/dynamic.py b/src/ciberwebscan/core/scraping/dynamic.py index cd75ebd..9af4851 100644 --- a/src/ciberwebscan/core/scraping/dynamic.py +++ b/src/ciberwebscan/core/scraping/dynamic.py @@ -18,7 +18,11 @@ from bs4 import BeautifulSoup -from ciberwebscan.core.client.ssrf import SSRFBlockedError, check_url +from ciberwebscan.core.client.ssrf import ( + SSRFBlockedError, + check_proxy_url, + check_url, +) from ciberwebscan.core.scraping.extractor import DataExtractor from ciberwebscan.core.scraping.helpers import ( check_robots_txt, @@ -107,6 +111,13 @@ class DynamicScrapeConfig: allow_local: bool = False """Whether to allow scraping local/private IPs.""" + allow_local_proxy: bool = False + """Whether to allow local/private proxy endpoints for Chromium. + + Independent from ``allow_local`` (targets): neither authorizes the + other. Link-local/IMDS, CGNAT, multicast and reserved proxy + addresses stay blocked even when True.""" + attributes: list[str] | None = None """Specific attributes to extract from elements.""" @@ -513,9 +524,12 @@ async def _create_context( if user_agent: context_options["user_agent"] = user_agent - # Set proxy + # Set proxy — validated against the SSRF policy before it is + # handed to Chromium: the browser opens its own connection, + # outside HTTPClient, so this is a second enforcement boundary. proxy = config.proxy or self._get_proxy() if proxy: + check_proxy_url(proxy, allow_local=config.allow_local_proxy) context_options["proxy"] = {"server": proxy} assert self._browser is not None # Ensured by caller @@ -569,9 +583,14 @@ async def _setup_page( """Set up page request interception. A guard route is always registered: every http(s) subresource is - validated against the SSRF policy before it is sent. Redirect hops - are not intercepted by Playwright routing, so the final ``page.url`` - is re-validated after navigation (see :meth:`scrape`). + validated against the SSRF policy before it is sent. Per the + Playwright docs, a request and its redirects are treated as a + single unit — the route handler runs once for the original + request and the browser follows redirects on its own — so + intermediate redirect hops are NOT re-validated here; only the + final ``page.url`` is re-checked after navigation (see + :meth:`scrape`). WebSocket handshakes are separately routed and + validated via :meth:`_handle_websocket_route`. """ block_list = config.block_resources or [] # Capture for closure allow_local = config.allow_local @@ -581,6 +600,43 @@ async def _route_handler(route) -> None: await page.route("**/*", _route_handler) + async def _websocket_handler(ws) -> None: + await self._handle_websocket_route(ws, allow_local=allow_local) + + await page.route_web_socket("**/*", _websocket_handler) + + async def _handle_websocket_route( + self, + ws, + *, + allow_local: bool = False, + ) -> None: + """Validate a routed WebSocket endpoint, then relay or block it. + + ``page.route`` does not intercept WebSocket handshakes; + ``route_web_socket`` does, and a routed socket is not connected + until :meth:`~playwright.async_api.WebSocketRoute.connect_to_server` + is called, so a blocked endpoint never reaches the network. + + Args: + ws: Playwright WebSocketRoute for the pending handshake. + allow_local: Whether localhost/private *destinations* are + permitted for this scrape (target opt-in, not proxy). + """ + url = ws.url + target = url + if url.startswith("ws://"): + target = f"http://{url[len('ws://') :]}" + elif url.startswith("wss://"): + target = f"https://{url[len('wss://') :]}" + try: + check_url(target, allow_local=allow_local) + except SSRFBlockedError: + logger.warning("SSRF policy blocked WebSocket to %s", url) + await ws.close(code=1008, reason="Blocked by SSRF policy") + return + await ws.connect_to_server() + async def _handle_route( self, route, diff --git a/src/ciberwebscan/services/analyze_service.py b/src/ciberwebscan/services/analyze_service.py index e94b843..adb1381 100644 --- a/src/ciberwebscan/services/analyze_service.py +++ b/src/ciberwebscan/services/analyze_service.py @@ -23,6 +23,7 @@ CVEAggregator, CVESource, ) +from ciberwebscan.core.client.ssrf import SSRFBlockedError from ciberwebscan.export.models import ( AnalysisReport, ConfidenceLevel, @@ -324,6 +325,11 @@ def analyze(self, options: AnalyzeOptions) -> ServiceResult[AnalysisReport]: result.error = str(e) result.error_code = e.code self.logger.error(f"Validation error: {e}") + except SSRFBlockedError as e: + # Proxy/transport blocked by the SSRF policy + result.error = str(e) + result.error_code = "VALIDATION_ERROR" + self.logger.error(f"SSRF policy blocked the analysis: {e}") except ExecutionError as e: result.error = str(e) result.error_code = e.code @@ -565,6 +571,10 @@ def _analyze_headers( score=overall_score, ) + except SSRFBlockedError: + # Forbidden proxy/transport: propagate so the caller reports + # a security decision instead of a partial success. + raise except Exception as e: self.logger.warning(f"Headers analysis failed: {e}") return None @@ -642,6 +652,10 @@ def _fingerprint( waf=fp_result.get("waf"), ) + except SSRFBlockedError: + # Forbidden proxy/transport: propagate so the caller reports + # a security decision instead of a partial success. + raise except Exception as e: self.logger.warning(f"Fingerprinting failed: {e}") return None diff --git a/src/ciberwebscan/services/attack_service.py b/src/ciberwebscan/services/attack_service.py index fe26f7c..2b525f9 100644 --- a/src/ciberwebscan/services/attack_service.py +++ b/src/ciberwebscan/services/attack_service.py @@ -30,6 +30,7 @@ ) from ciberwebscan.core.attacks.base import AttackIntensity from ciberwebscan.core.client import HTTPClient +from ciberwebscan.core.client.ssrf import SSRFBlockedError from ciberwebscan.export.models import ( AttackResult, ExportMeta, @@ -495,6 +496,9 @@ def attack(self, options: AttackOptions) -> ServiceResult[AttackResult]: raise except ExecutionError: raise + except SSRFBlockedError as e: + # Proxy/transport blocked by the SSRF policy + raise ValidationError(str(e), details={"url": options.url}) from e except Exception as e: self.logger.exception("Unexpected error during attack simulation") raise ExecutionError( diff --git a/src/ciberwebscan/services/quick_service.py b/src/ciberwebscan/services/quick_service.py index a3cf5c6..27648bd 100644 --- a/src/ciberwebscan/services/quick_service.py +++ b/src/ciberwebscan/services/quick_service.py @@ -12,6 +12,7 @@ from typing import Any from ciberwebscan.config.loader import get_config +from ciberwebscan.core.client.ssrf import SSRFBlockedError, check_proxy_url from ciberwebscan.export.models import AnalysisReport, ExportMeta from ciberwebscan.services.analyze_service import AnalyzeOptions, AnalyzeService from ciberwebscan.services.attack_service import AttackOptions, AttackService @@ -166,6 +167,16 @@ def quick_scan(self, options: QuickOptions) -> ServiceResult[AnalysisReport]: ) self._check_target_url(url, allow_local=allow_local) + # SSRF protection on the request proxy: rejected early so a + # forbidden proxy cannot silently degrade the scan into a + # partial result. The phase boundaries re-validate too. + if options.proxy: + proxy_cfg = cfg.http.proxy + check_proxy_url( + options.proxy, + allow_local=proxy_cfg.allow_local if proxy_cfg else False, + ) + # Validate preset preset = options.preset.lower() if preset not in PRESETS: @@ -220,6 +231,9 @@ def quick_scan(self, options: QuickOptions) -> ServiceResult[AnalysisReport]: except ValidationError: raise + except SSRFBlockedError as e: + # Proxy/transport blocked by the SSRF policy + raise ValidationError(str(e), details={"url": options.url}) from e except Exception as e: result.error = str(e) result.error_code = "QUICK_SCAN_ERROR" diff --git a/src/ciberwebscan/services/scrape_service.py b/src/ciberwebscan/services/scrape_service.py index 21da8e3..ed6113d 100644 --- a/src/ciberwebscan/services/scrape_service.py +++ b/src/ciberwebscan/services/scrape_service.py @@ -15,6 +15,7 @@ from ciberwebscan.config.loader import get_config from ciberwebscan.config.models import ScrapingConfig +from ciberwebscan.core.client.ssrf import SSRFBlockedError from ciberwebscan.core.scraping import ( DataExtractor, ScrapeConfig, @@ -299,6 +300,11 @@ def scrape( result.error = str(e) result.error_code = e.code self.logger.error(f"Validation error: {e}") + except SSRFBlockedError as e: + # Proxy/transport blocked by the SSRF policy + result.error = str(e) + result.error_code = "VALIDATION_ERROR" + self.logger.error(f"SSRF policy blocked the scrape: {e}") except ExecutionError as e: result.error = str(e) result.error_code = e.code @@ -457,6 +463,10 @@ def _scrape_static(self, url: str, options: ScrapeOptions) -> ScrapeResult: except ExecutionError: raise + except SSRFBlockedError: + # Forbidden proxy/transport: propagate to the service boundary + # as a security decision, not a wrapped execution failure. + raise except Exception as e: raise ExecutionError(f"Static scraping failed: {e}") from e @@ -466,6 +476,7 @@ def _scrape_dynamic(self, url: str, options: ScrapeOptions) -> ScrapeResult: from ciberwebscan.core.scraping import BrowserType, DynamicScrapeConfig dyn_cfg = self.config.dynamic + proxy_cfg = self.app_config.http.proxy config = DynamicScrapeConfig( selector=options.selector or "body", wait_selector=options.wait_for or dyn_cfg.wait_selector, @@ -475,6 +486,7 @@ def _scrape_dynamic(self, url: str, options: ScrapeOptions) -> ScrapeResult: schema=options.schema, extract_forms=options.extract_forms, allow_local=self.config.allow_local, + allow_local_proxy=bool(proxy_cfg.allow_local if proxy_cfg else False), ) # Run the async scrape method safely from sync context @@ -540,6 +552,10 @@ def _scrape_dynamic(self, url: str, options: ScrapeOptions) -> ScrapeResult: except ExecutionError: raise + except SSRFBlockedError: + # Forbidden proxy/transport (e.g. Chromium context proxy): + # propagate to the service boundary as a security decision. + raise except Exception as e: raise ExecutionError(f"Dynamic scraping failed: {e}") from e diff --git a/tests/integration/test_socks5_transport.py b/tests/integration/test_socks5_transport.py new file mode 100644 index 0000000..ee2e9bf --- /dev/null +++ b/tests/integration/test_socks5_transport.py @@ -0,0 +1,156 @@ +"""Real SOCKS5 transport verification. + +A minimal local SOCKS5 relay plus a local HTTP origin prove that httpx +(with ``socksio`` from ``httpx[socks]``) really tunnels through a +``socks5://`` proxy — not just that the URL validates — and that the +SSRF policy still rejects the same endpoint without the local opt-in, +before any connection is attempted. +""" + +from __future__ import annotations + +import contextlib +import socket +import socketserver +import threading +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from ciberwebscan.core.client import HTTPClient +from ciberwebscan.core.client.ssrf import SSRFBlockedError + +pytestmark = [pytest.mark.network, pytest.mark.integration] + + +def _recv_exact(sock: socket.socket, size: int) -> bytes: + """Read exactly *size* bytes (loopback may deliver partial reads).""" + chunks = bytearray() + while len(chunks) < size: + chunk = sock.recv(size - len(chunks)) + if not chunk: + break + chunks.extend(chunk) + return bytes(chunks) + + +class _Socks5Server(socketserver.ThreadingTCPServer): + allow_reuse_address = True + daemon_threads = True + + +class _Socks5Handler(socketserver.BaseRequestHandler): + """Minimal SOCKS5 CONNECT relay (no auth; IPv4/IPv6/domain targets).""" + + def handle(self) -> None: + sock: socket.socket = self.request + header = _recv_exact(sock, 2) + if len(header) < 2 or header[0] != 5: + return + methods = _recv_exact(sock, header[1]) + if b"\x00" not in methods: + sock.sendall(b"\x05\xff") # no acceptable auth method + return + sock.sendall(b"\x05\x00") # no authentication required + + request = _recv_exact(sock, 4) + if len(request) < 4 or request[0] != 5 or request[1] != 1: + return + atyp = request[3] + if atyp == 1: # IPv4 + host = socket.inet_ntoa(_recv_exact(sock, 4)) + elif atyp == 3: # domain name + length = _recv_exact(sock, 1)[0] + host = _recv_exact(sock, length).decode() + elif atyp == 4: # IPv6 + host = socket.inet_ntop(socket.AF_INET6, _recv_exact(sock, 16)) + else: + return + port = int.from_bytes(_recv_exact(sock, 2), "big") + + try: + upstream = socket.create_connection((host, port), timeout=10) + except OSError: + sock.sendall(b"\x05\x01\x00\x01" + b"\x00" * 6) # general failure + return + + sock.sendall( + b"\x05\x00\x00\x01" + socket.inet_aton("127.0.0.1") + (0).to_bytes(2, "big") + ) + + def _pump(source: socket.socket, destination: socket.socket) -> None: + try: + while True: + chunk = source.recv(65536) + if not chunk: + break + destination.sendall(chunk) + except OSError: + pass + finally: + for side in (source, destination): + with contextlib.suppress(OSError): + side.shutdown(socket.SHUT_RDWR) + + pumps = [ + threading.Thread(target=_pump, args=(sock, upstream), daemon=True), + threading.Thread(target=_pump, args=(upstream, sock), daemon=True), + ] + for pump in pumps: + pump.start() + for pump in pumps: + pump.join(timeout=10) + + +class _OriginHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 — http.server naming + body = b"socks5-ok" + self.send_response(200) + self.send_header("Content-Type", "text/plain") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: object) -> None: + """Silence per-request logging.""" + + +@pytest.fixture +def socks5_and_origin() -> Iterator[tuple[int, int]]: + """Start a local SOCKS5 relay and HTTP origin on loopback.""" + origin = ThreadingHTTPServer(("127.0.0.1", 0), _OriginHandler) + socks = _Socks5Server(("127.0.0.1", 0), _Socks5Handler) + for server in (origin, socks): + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + yield socks.server_address[1], origin.server_address[1] + finally: + for server in (origin, socks): + server.shutdown() + server.server_close() + + +class TestSocks5Transport: + """Scenario 12/9 (P4): SOCKS5 works for real or is rejected explicitly.""" + + def test_real_socks5_tunnel(self, socks5_and_origin: tuple[int, int]) -> None: + proxy_port, origin_port = socks5_and_origin + + with HTTPClient( + proxy=f"socks5://127.0.0.1:{proxy_port}", + allow_local_proxy=True, + timeout=10.0, + ) as client: + response = client.get(f"http://127.0.0.1:{origin_port}/") + + assert response.status_code == 200 + assert response.text == "socks5-ok" + + def test_socks5_proxy_rejected_without_opt_in( + self, socks5_and_origin: tuple[int, int] + ) -> None: + proxy_port, _origin_port = socks5_and_origin + + with pytest.raises(SSRFBlockedError): + HTTPClient(proxy=f"socks5://127.0.0.1:{proxy_port}") diff --git a/tests/unit/api/routes/test_config_protected_keys.py b/tests/unit/api/routes/test_config_protected_keys.py new file mode 100644 index 0000000..c4bfad3 --- /dev/null +++ b/tests/unit/api/routes/test_config_protected_keys.py @@ -0,0 +1,217 @@ +"""Remote configuration authorization tests (BUG-014, P2). + +Proxy options (``http.proxy.*``) and every security exception +(``*.allow_local``) are operator-only: they cannot be modified through +``PUT /api/config`` because API keys carry no administrative scope +distinction. The local operator keeps full control via the config file, +environment variables and the CLI (which never pass through this route). +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from ciberwebscan.api.app import create_app +from ciberwebscan.api.routes.config import config_update_touches_protected +from ciberwebscan.services.config_service import ConfigService + +pytestmark = pytest.mark.unit + + +@pytest.fixture +def isolated_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Redirect HOME/USERPROFILE so Path.home() lands in tmp.""" + fake_home = tmp_path / "home" + fake_home.mkdir() + monkeypatch.setenv("HOME", str(fake_home)) + monkeypatch.setenv("USERPROFILE", str(fake_home)) + return fake_home + + +@pytest.fixture +def real_client(isolated_home: Path): + """TestClient WITHOUT auth overrides: real API-key auth via global config.""" + app = create_app() + client = TestClient(app) + yield client + app.dependency_overrides.clear() + + +@pytest.fixture +def auth_headers(isolated_home: Path) -> dict[str, str]: + """Seed a known API key in the global config and return its auth headers.""" + key = "unit-test-api-key-1234567890" + ConfigService().set("api.auth.api_keys", [key]) + return {"X-API-Key": key} + + +class TestProtectedKeysHelper: + """Unit coverage of the protection predicate itself.""" + + @pytest.mark.parametrize( + "path", + [ + "http.proxy", + "http.proxy.http", + "http.proxy.https", + "http.proxy.socks5", + "http.proxy.rotate", + "http.proxy.rotation_interval", + "http.proxy.proxy_list", + "http.proxy.allow_local", + "http.proxy.use_env", + "HTTP.Proxy.HTTP", # case variant + " http.proxy.use_env ", # whitespace variant + "http.proxy.", # trailing dot + "analysis.allow_local", + "scraping.allow_local", + "attack.allow_local", + "allow_local", # bare exception key + ], + ) + def test_protected(self, path: str) -> None: + assert config_update_touches_protected(path, "anything") is True + + @pytest.mark.parametrize( + "path", + [ + "http.timeout.connect", + "http.retry.max_attempts", + "http.tls.verify", + "logging.level", + "api.auth.api_keys", + "scraping.dynamic.enabled", + ], + ) + def test_not_protected(self, path: str) -> None: + assert config_update_touches_protected(path, "anything") is False + + @pytest.mark.parametrize( + "path,value", + [ + # Section updates smuggling protected subkeys + ("http", {"proxy": {"allow_local": True}}), + ("http", {"proxy": {"http": "http://evil.example.com:8080"}}), + ("analysis", {"allow_local": True}), + ("scraping", {"allow_local": True}), + ("attack", {"allow_local": True}), + # Deeply nested batch payloads + ("", {"http": {"proxy": {"use_env": True}}}), + ("", {"analysis": {"allow_local": True}}), + ], + ) + def test_batch_updates_protected(self, path: str, value: dict) -> None: + assert config_update_touches_protected(path, value) is True + + def test_batch_update_with_only_allowed_keys(self) -> None: + assert ( + config_update_touches_protected( + "http", {"timeout": {"connect": 15.0}, "http2": True} + ) + is False + ) + + +class TestRemoteUpdateRejected: + """Scenarios 19-23: remote writes to protected keys always fail closed.""" + + @pytest.mark.parametrize( + "path,value", + [ + ("http.proxy.http", "http://proxy.example.com:8080"), + ("http.proxy.socks5", "socks5://proxy.example.com:1080"), + ("http.proxy.rotate", True), + ("http.proxy.proxy_list", ["http://proxy.example.com:8080"]), + ("http.proxy.allow_local", True), + ("http.proxy.use_env", True), + ("HTTP.Proxy.Allow_Local", True), # case variant + ("analysis.allow_local", True), + ("scraping.allow_local", True), + ("attack.allow_local", True), + # Batch/section updates + ("http", {"proxy": {"allow_local": True}}), + ("analysis", {"allow_local": True}), + ], + ) + def test_protected_update_returns_403( + self, + real_client: TestClient, + auth_headers: dict[str, str], + path: str, + value: object, + ) -> None: + resp = real_client.put( + "/api/config", + json={"path": path, "value": value}, + headers=auth_headers, + ) + assert resp.status_code == 403, resp.text + detail = resp.json()["detail"] + assert "protected" in detail.lower() + # No configuration values or secrets leak back to the caller. + assert "proxy.example.com" not in detail + + def test_protected_update_with_save_flag_rejected( + self, + real_client: TestClient, + auth_headers: dict[str, str], + tmp_path: Path, + ) -> None: + resp = real_client.put( + "/api/config", + json={"path": "http.proxy.allow_local", "value": True, "save": True}, + headers=auth_headers, + ) + assert resp.status_code == 403, resp.text + + def test_value_not_applied_after_403( + self, real_client: TestClient, auth_headers: dict[str, str] + ) -> None: + from ciberwebscan.config.loader import get_config + + real_client.put( + "/api/config", + json={"path": "http.proxy.allow_local", "value": True}, + headers=auth_headers, + ) + proxy = get_config().http.proxy + assert proxy is None or proxy.allow_local is False + + def test_allowed_keys_still_updatable( + self, real_client: TestClient, auth_headers: dict[str, str] + ) -> None: + resp = real_client.put( + "/api/config", + json={"path": "http.timeout.connect", "value": 15.0}, + headers=auth_headers, + ) + assert resp.status_code == 200, resp.text + assert resp.json()["data"]["value"] == 15.0 + + def test_unauthenticated_request_still_rejected( + self, real_client: TestClient + ) -> None: + resp = real_client.put( + "/api/config", + json={"path": "http.proxy.allow_local", "value": True}, + ) + assert resp.status_code == 401 + + def test_reset_of_protected_section_still_allowed( + self, real_client: TestClient, auth_headers: dict[str, str] + ) -> None: + """Reset restores safe defaults, so it is not a protected write.""" + from ciberwebscan.config.loader import get_config + + ConfigService().set("http.proxy.allow_local", True) + resp = real_client.post( + "/api/config/reset", + json={"path": "http.proxy"}, + headers=auth_headers, + ) + assert resp.status_code == 200, resp.text + proxy = get_config().http.proxy + assert proxy is None or proxy.allow_local is False diff --git a/tests/unit/config/test_proxy_config_policy.py b/tests/unit/config/test_proxy_config_policy.py new file mode 100644 index 0000000..beb112e --- /dev/null +++ b/tests/unit/config/test_proxy_config_policy.py @@ -0,0 +1,152 @@ +"""ProxyConfig ingestion-policy tests. + +Every proxy that reaches the configuration model is classified at +ingestion time with the same central policy used at the usage boundary, +so ``proxy_list`` as a string and as a list behave identically. Invalid +environment overrides degrade the whole config to defaults with an +explicit error log — never to an unvalidated proxy. +""" + +from __future__ import annotations + +import logging + +import pytest +from pydantic import ValidationError + +from ciberwebscan.config.loader import get_config, reset_config +from ciberwebscan.config.models import ProxyConfig + +pytestmark = pytest.mark.unit + + +class TestProxyListConsistency: + """Scenario 11: string and list forms are equally policed.""" + + def test_list_with_loopback_rejected(self) -> None: + with pytest.raises(ValidationError, match="SSRF"): + ProxyConfig(proxy_list=["http://127.0.0.1:8080"]) + + def test_string_with_loopback_rejected(self) -> None: + with pytest.raises(ValidationError, match="SSRF"): + ProxyConfig(proxy_list="http://127.0.0.1:8080") + + def test_string_and_list_reject_the_same_entry(self) -> None: + entry = "http://169.254.169.254" + with pytest.raises(ValidationError): + ProxyConfig(proxy_list=[entry]) + with pytest.raises(ValidationError): + ProxyConfig(proxy_list=entry) + + def test_mixed_string_list_with_loopback_rejected(self) -> None: + with pytest.raises(ValidationError): + ProxyConfig( + proxy_list="http://proxy.example.com:8080, http://10.0.0.5:3128" + ) + + def test_string_and_list_accepted_with_opt_in(self) -> None: + from_list = ProxyConfig(proxy_list=["http://127.0.0.1:8080"], allow_local=True) + from_string = ProxyConfig( + proxy_list="http://127.0.0.1:8080, http://192.168.1.10:3128", + allow_local=True, + ) + assert from_list.proxy_list == ["http://127.0.0.1:8080"] + assert from_string.proxy_list == [ + "http://127.0.0.1:8080", + "http://192.168.1.10:3128", + ] + + def test_public_list_accepted(self) -> None: + config = ProxyConfig( + proxy_list=["http://proxy.example.com:8080"], allow_local=False + ) + assert config.proxy_list == ["http://proxy.example.com:8080"] + + +class TestIndividualProxyFields: + """Scenario 12: http/https/socks5 fields go through the same policy.""" + + @pytest.mark.parametrize( + "kwargs", + [ + {"http": "http://127.0.0.1:8080"}, + {"https": "http://10.0.0.5:3128"}, + {"socks5": "socks5://127.0.0.1:1080"}, + ], + ) + def test_private_proxy_fields_rejected_by_default(self, kwargs: dict) -> None: + with pytest.raises(ValidationError, match="SSRF"): + ProxyConfig(**kwargs) + + @pytest.mark.parametrize( + "kwargs", + [ + {"http": "http://127.0.0.1:8080"}, + {"https": "http://10.0.0.5:3128"}, + {"socks5": "socks5://192.168.1.10:1080"}, + ], + ) + def test_private_proxy_fields_accepted_with_opt_in(self, kwargs: dict) -> None: + config = ProxyConfig(allow_local=True, **kwargs) + assert config.allow_local is True + + @pytest.mark.parametrize("allow_local", [False, True]) + def test_imds_rejected_even_with_opt_in(self, allow_local: bool) -> None: + with pytest.raises(ValidationError, match="SSRF"): + ProxyConfig(allow_local=allow_local, http="http://169.254.169.254") + + def test_public_fields_accepted(self) -> None: + config = ProxyConfig( + http="http://proxy.example.com:8080", + socks5="socks5://proxy.example.com:1080", + ) + assert config.http is not None + assert config.socks5 == "socks5://proxy.example.com:1080" + + def test_defaults_have_no_proxy_and_safe_flags(self) -> None: + config = ProxyConfig() + assert config.http is None + assert config.https is None + assert config.socks5 is None + assert config.proxy_list is None + assert config.allow_local is False + assert config.use_env is False + + +class TestEnvironmentOverrides: + """Scenario 17: invalid config degrades loudly, never silently.""" + + def test_env_enables_local_proxy(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CIBERWEBSCAN_HTTP_PROXY_ALLOW_LOCAL", "true") + monkeypatch.setenv("CIBERWEBSCAN_HTTP_PROXY_HTTP", "http://127.0.0.1:8080") + reset_config() + proxy = get_config().http.proxy + assert proxy is not None + assert proxy.allow_local is True + assert str(proxy.http).startswith("http://127.0.0.1:8080") + + def test_env_rejects_imds_even_with_local_opt_in( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + monkeypatch.setenv("CIBERWEBSCAN_HTTP_PROXY_ALLOW_LOCAL", "true") + monkeypatch.setenv("CIBERWEBSCAN_HTTP_PROXY_HTTP", "http://169.254.169.254:80") + reset_config() + with caplog.at_level(logging.ERROR, logger="ciberwebscan.config.loader"): + config = get_config() + # Fallback to defaults: no proxy at all, with an explicit error log. + assert config.http.proxy is None + assert any( + "Invalid configuration" in record.message for record in caplog.records + ) + + def test_env_rejects_private_proxy_without_opt_in( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + monkeypatch.setenv("CIBERWEBSCAN_HTTP_PROXY_HTTP", "http://10.0.0.5:3128") + reset_config() + with caplog.at_level(logging.ERROR, logger="ciberwebscan.config.loader"): + config = get_config() + assert config.http.proxy is None + assert any( + "Invalid configuration" in record.message for record in caplog.records + ) diff --git a/tests/unit/core/client/test_http_client_proxy_boundary.py b/tests/unit/core/client/test_http_client_proxy_boundary.py new file mode 100644 index 0000000..8d926a4 --- /dev/null +++ b/tests/unit/core/client/test_http_client_proxy_boundary.py @@ -0,0 +1,220 @@ +"""HTTPClient proxy-boundary tests. + +``HTTPClient.__init__`` is the single enforcement point for every HTTP +client (CLI, API, config fallback, rotation, helper clients): the +effective proxy is classified *before* ``httpx.Client`` is constructed, +so a forbidden proxy never reaches a transport. The ``httpx.Client`` +factory is patched to record whether a transport would have been built — +that flag is the evidence of "no connection attempted". +""" + +from __future__ import annotations + +from typing import Any + +import httpx +import pytest + +from ciberwebscan.config.loader import get_config +from ciberwebscan.config.models import ProxyConfig +from ciberwebscan.core.client import HTTPClient +from ciberwebscan.core.client.ssrf import SSRFBlockedError + +pytestmark = pytest.mark.unit + + +@pytest.fixture +def client_factory(monkeypatch: pytest.MonkeyPatch): + """Patch httpx.Client, recording construction and capturing kwargs.""" + captured: dict[str, Any] = {"constructed": False, "kwargs": {}} + real_client_cls = httpx.Client + + def factory(**kwargs: Any) -> httpx.Client: + captured["constructed"] = True + captured["kwargs"] = kwargs + return real_client_cls(**kwargs) + + monkeypatch.setattr(httpx, "Client", factory) + return captured + + +def _legacy_config_proxy(**overrides: Any) -> ProxyConfig: + """Build a ProxyConfig bypassing model validation. + + Simulates configuration that reached the singleton through a legacy + path (or a fixture) without going through the ingestion validator, so + the usage boundary is what gets exercised. + """ + fields: dict[str, Any] = { + "http": None, + "https": None, + "socks5": None, + "rotate": False, + "rotation_interval": 10, + "proxy_list": None, + "allow_local": False, + "use_env": False, + } + fields.update(overrides) + return ProxyConfig.model_construct(**fields) + + +class TestForbiddenProxyNeverReachesTransport: + """Scenarios 1-4, 13: blocked before any transport exists.""" + + @pytest.mark.parametrize( + "proxy", + [ + "http://127.0.0.1:8080", + "http://10.0.0.5:3128", + "http://169.254.169.254", + "http://[::1]:1080", + "http://2130706433:8080", + "socks5://127.0.0.1:1080", + ], + ) + def test_loopback_private_imds_proxy_blocked_at_init( + self, proxy: str, client_factory: dict[str, Any] + ) -> None: + with pytest.raises(SSRFBlockedError): + HTTPClient(proxy=proxy) + assert client_factory["constructed"] is False + + def test_config_fallback_proxy_is_validated( + self, client_factory: dict[str, Any] + ) -> None: + """Scenario 14: a persistent config proxy is checked before use.""" + get_config().http.proxy = _legacy_config_proxy( + https="http://169.254.169.254:80" + ) + with pytest.raises(SSRFBlockedError): + HTTPClient() + assert client_factory["constructed"] is False + + def test_credentials_never_echoed(self) -> None: + with pytest.raises(SSRFBlockedError) as exc: + HTTPClient(proxy="http://user:S3cr3t@127.0.0.1:8080") + assert "S3cr3t" not in str(exc.value) + + +class TestEnvironmentProxies: + """Scenarios 15-16: trust_env is explicit and env proxies are policed.""" + + def test_env_proxies_ignored_by_default( + self, client_factory: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.1:8080") + monkeypatch.setenv("HTTPS_PROXY", "http://169.254.169.254") + client = HTTPClient() + assert client_factory["constructed"] is True + assert client_factory["kwargs"]["trust_env"] is False + assert client_factory["kwargs"]["proxy"] is None + client.close() + + def test_env_proxy_validated_when_enabled( + self, client_factory: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.1:8080") + with pytest.raises(SSRFBlockedError): + HTTPClient(use_env_proxies=True) + assert client_factory["constructed"] is False + + def test_lowercase_env_variant_validated_when_enabled( + self, client_factory: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("all_proxy", "socks5://127.0.0.1:1080") + with pytest.raises(SSRFBlockedError): + HTTPClient(use_env_proxies=True) + assert client_factory["constructed"] is False + + def test_public_env_proxy_accepted_when_enabled( + self, client_factory: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("HTTP_PROXY", "http://proxy.example.com:8080") + client = HTTPClient(use_env_proxies=True) + assert client_factory["constructed"] is True + assert client_factory["kwargs"]["trust_env"] is True + client.close() + + def test_config_use_env_flag_enables_env_honouring( + self, client_factory: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("HTTP_PROXY", "http://169.254.169.254") + get_config().http.proxy = _legacy_config_proxy(use_env=True) + with pytest.raises(SSRFBlockedError): + HTTPClient() + assert client_factory["constructed"] is False + + def test_explicit_proxy_supersedes_env( + self, client_factory: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + """With an explicit (validated) proxy httpx never consults env.""" + monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.1:8080") + client = HTTPClient(proxy="http://proxy.example.com:8080", use_env_proxies=True) + assert client_factory["constructed"] is True + assert client_factory["kwargs"]["proxy"] == "http://proxy.example.com:8080" + client.close() + + +class TestLocalProxyOptIn: + """Scenario 10 + independence (scenario 18) at the boundary.""" + + def test_allow_local_proxy_permits_loopback( + self, client_factory: dict[str, Any] + ) -> None: + client = HTTPClient(proxy="http://127.0.0.1:8080", allow_local_proxy=True) + assert client_factory["constructed"] is True + client.close() + + def test_allow_local_proxy_still_blocks_imds( + self, client_factory: dict[str, Any] + ) -> None: + with pytest.raises(SSRFBlockedError): + HTTPClient(proxy="http://169.254.169.254:80", allow_local_proxy=True) + assert client_factory["constructed"] is False + + def test_config_allow_local_proxy_opt_in( + self, client_factory: dict[str, Any] + ) -> None: + get_config().http.proxy = ProxyConfig( + allow_local=True, http="http://127.0.0.1:8080" + ) + client = HTTPClient() + assert client_factory["constructed"] is True + client.close() + + def test_target_allow_local_does_not_permit_local_proxy( + self, client_factory: dict[str, Any] + ) -> None: + with pytest.raises(SSRFBlockedError): + HTTPClient(proxy="http://127.0.0.1:8080", ssrf_guard=True, allow_local=True) + assert client_factory["constructed"] is False + + +class TestSocks5AndNoProxyClients: + """Scenarios 9, 29: SOCKS5 is really supported; plain clients unchanged.""" + + def test_socks5_proxy_client_is_constructible( + self, client_factory: dict[str, Any] + ) -> None: + """httpx[socks] (socksio) is installed: no 'requires socksio' error.""" + client = HTTPClient(proxy="socks5://proxy.example.com:1080") + assert client_factory["constructed"] is True + client.close() + + def test_client_without_proxy_still_works( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + real_client_cls = httpx.Client + + def factory(**kwargs: Any) -> httpx.Client: + kwargs["transport"] = httpx.MockTransport( + lambda request: httpx.Response(200, text="ok") + ) + return real_client_cls(**kwargs) + + monkeypatch.setattr(httpx, "Client", factory) + client = HTTPClient() + response = client.get("http://example.com/") + assert response.status_code == 200 + client.close() diff --git a/tests/unit/core/client/test_proxy.py b/tests/unit/core/client/test_proxy.py index db50b53..5a8a3cc 100644 --- a/tests/unit/core/client/test_proxy.py +++ b/tests/unit/core/client/test_proxy.py @@ -59,7 +59,8 @@ def test_local_proxy_blocked_by_default(self): """Local proxies should be blocked by default.""" with pytest.raises(ProxyValidationError) as exc: parse_proxy("http://localhost:8080") - assert "Local proxies are not allowed" in str(exc.value) + assert "ssrf" in str(exc.value).lower() + assert "blocked" in str(exc.value).lower() with pytest.raises(ProxyValidationError): parse_proxy("http://127.0.0.1:8080") @@ -81,7 +82,8 @@ def test_private_ip_blocked_by_default(self): """Private IP proxies should be blocked by default.""" with pytest.raises(ProxyValidationError) as exc: parse_proxy("http://192.168.1.1:8080") - assert "Private network proxies are not allowed" in str(exc.value) + assert "ssrf" in str(exc.value).lower() + assert "blocked" in str(exc.value).lower() with pytest.raises(ProxyValidationError): parse_proxy("http://10.0.0.1:8080") diff --git a/tests/unit/core/client/test_ssrf_proxy.py b/tests/unit/core/client/test_ssrf_proxy.py new file mode 100644 index 0000000..3dd88ec --- /dev/null +++ b/tests/unit/core/client/test_ssrf_proxy.py @@ -0,0 +1,196 @@ +"""Proxy-endpoint SSRF policy tests. + +``check_proxy_url`` classifies the endpoint the client actually connects +to (the proxy), independently from the request-target policy in +``check_url``. Hostname answers come from the shared fake DNS resolver +fixture in ``tests/conftest.py``; no real resolution or connection ever +happens here. +""" + +from __future__ import annotations + +import pytest + +from ciberwebscan.core.client.ssrf import ( + PROXY_ALLOWED_SCHEMES, + SSRFBlockedError, + check_proxy_url, + check_url, +) + +pytestmark = pytest.mark.unit + +# Loopback/private/link-local/CGNAT plus non-canonical IPv4 notations: +# every one of these must be rejected before any socket is opened. +BLOCKED_BY_DEFAULT = [ + "http://127.0.0.1:8080", + "http://localhost:8080", + "http://localtest.me:8080", # fake DNS -> 127.0.0.1 + "http://10.0.0.5:3128", + "http://192.168.1.10:8080", + "http://172.16.0.1:8080", + "http://169.254.169.254", # link-local / cloud IMDS + "http://metadata.google.internal", # fake DNS -> 169.254.169.254 + "http://100.64.0.1:8080", # CGNAT shared address space + "http://0.0.0.0:8080", # unspecified / "this network" + "http://[::1]:8080", + "http://[fe80::1]:8080", + "http://[fc00::1]:1080", # IPv6 ULA + "http://[::]:8080", # unspecified IPv6 + "socks5://127.0.0.1:1080", + "socks5://10.0.0.5:1080", + "http://2130706433:8080", # decimal loopback + "http://0x7f000001:8080", # hex loopback + "http://127.1:8080", # short-form loopback + "http://0177.0.0.1:8080", # octal loopback + "http://[::ffff:127.0.0.1]:8080", # IPv4-mapped loopback + "http://[::ffff:169.254.169.254]:80", # IPv4-mapped IMDS +] + +# Accepted once the operator explicitly authorizes local proxies. +ALLOWED_WITH_LOCAL = [ + "http://127.0.0.1:8080", + "http://localhost:8080", + "http://192.168.1.10:8080", + "http://10.0.0.5:3128", + "http://172.16.0.1:3128", + "http://[::1]:1080", + "socks5://localhost:1080", + "http://user:pass@192.168.1.10:8080", +] + +# Infrastructure ranges: blocked even with local proxies authorized. +STILL_BLOCKED_WITH_LOCAL = [ + "http://169.254.169.254", # IMDS + "http://metadata.google.internal", # DNS -> IMDS + "http://100.64.0.1:8080", # CGNAT + "http://0.0.0.0:8080", # unspecified + "http://[fe80::1]:8080", # IPv6 link-local + "http://[::]:8080", # unspecified IPv6 + "http://224.0.0.1:8080", # multicast + "http://240.0.0.1:8080", # reserved + "http://[::ffff:169.254.169.254]:80", # IPv4-mapped IMDS +] + +PUBLICLY_VALID = [ + "http://proxy.example.com:8080", + "https://proxy.example.com:8443", + "socks5://proxy.example.com:1080", + "http://93.184.216.34:8080", + "http://user:pass@proxy.example.com:3128", +] + + +class TestProxyBlockedByDefault: + """Scenarios 1-6: local/private/metadata proxies are rejected.""" + + @pytest.mark.parametrize("proxy", BLOCKED_BY_DEFAULT) + def test_blocked_by_default(self, proxy: str) -> None: + with pytest.raises(SSRFBlockedError) as exc: + check_proxy_url(proxy) + assert "ssrf" in str(exc.value).lower() + assert "blocked" in str(exc.value).lower() + + def test_hostname_resolving_to_private_is_blocked(self) -> None: + """Scenario 5: DNS name that resolves to a forbidden address.""" + with pytest.raises(SSRFBlockedError, match="localtest\\.me"): + check_proxy_url("http://localtest.me:8080") + + def test_dns_failure_is_fail_closed(self) -> None: + """Scenario 6: unresolvable proxy hosts are blocked, not allowed.""" + with pytest.raises(SSRFBlockedError, match="could not be resolved"): + check_proxy_url("http://nonexistent.invalid:8080") + + +class TestProxyPubliclyValid: + """Scenario 7: public proxies keep working.""" + + @pytest.mark.parametrize("proxy", PUBLICLY_VALID) + def test_public_proxy_accepted(self, proxy: str) -> None: + check_proxy_url(proxy) # must not raise + + def test_public_hostname_with_dns_is_accepted(self) -> None: + check_proxy_url("http://public.example.com:8080") + + +class TestProxyCredentials: + """Scenario 8: credentials are accepted but never echoed back.""" + + def test_credentials_do_not_break_validation(self) -> None: + check_proxy_url("http://user:pass@proxy.example.com:3128") + + def test_credentials_not_leaked_in_error(self) -> None: + with pytest.raises(SSRFBlockedError) as exc: + check_proxy_url("http://user:S3cr3t@127.0.0.1:8080") + message = str(exc.value) + assert "S3cr3t" not in message + assert "user:" not in message + + +class TestProxySchemesAndFormat: + """Scenario 9: only actually-supported schemes are accepted.""" + + def test_allowed_schemes(self) -> None: + assert frozenset({"http", "https", "socks5"}) == PROXY_ALLOWED_SCHEMES + + @pytest.mark.parametrize( + "proxy", + [ + "ftp://proxy.example.com:21", + "file:///etc/passwd", + "gopher://proxy.example.com:70", + "ws://proxy.example.com:80", + ], + ) + def test_unsupported_scheme_blocked(self, proxy: str) -> None: + with pytest.raises(SSRFBlockedError, match="unsupported scheme"): + check_proxy_url(proxy) + + @pytest.mark.parametrize( + "proxy", + [ + "", + " ", + "http://", + "http://proxy.example.com:notaport", + "http://proxy.example.com:99999", + ], + ) + def test_malformed_proxy_blocked(self, proxy: str) -> None: + with pytest.raises(SSRFBlockedError): + check_proxy_url(proxy) + + +class TestProxyLocalOptIn: + """Scenario 10: explicit local proxies work, infrastructure stays blocked.""" + + @pytest.mark.parametrize("proxy", ALLOWED_WITH_LOCAL) + def test_local_proxy_allowed_when_opted_in(self, proxy: str) -> None: + check_proxy_url(proxy, allow_local=True) # must not raise + + @pytest.mark.parametrize("proxy", STILL_BLOCKED_WITH_LOCAL) + def test_infrastructure_stays_blocked_with_opt_in(self, proxy: str) -> None: + with pytest.raises(SSRFBlockedError): + check_proxy_url(proxy, allow_local=True) + + +class TestTargetAndProxyAuthorizationsAreIndependent: + """Scenario 18: neither allow_local opt-in authorizes the other.""" + + def test_target_opt_in_does_not_permit_local_proxy(self) -> None: + check_url("http://127.0.0.1/", allow_local=True) # targets: allowed + with pytest.raises(SSRFBlockedError): + check_proxy_url("http://127.0.0.1:8080") # proxies: still blocked + + def test_proxy_opt_in_does_not_permit_local_target(self) -> None: + check_proxy_url("http://127.0.0.1:8080", allow_local=True) # proxies ok + with pytest.raises(SSRFBlockedError): + check_url("http://127.0.0.1/") # targets: still blocked + + def test_default_target_policy_unchanged(self) -> None: + """BUG-004 behaviour is preserved for targets.""" + with pytest.raises(SSRFBlockedError): + check_url("http://169.254.169.254/") + with pytest.raises(SSRFBlockedError): + check_url("http://2130706433/") + check_url("http://public.example.com/") diff --git a/tests/unit/core/scraping/test_dynamic.py b/tests/unit/core/scraping/test_dynamic.py index dab9c5c..7563cbd 100644 --- a/tests/unit/core/scraping/test_dynamic.py +++ b/tests/unit/core/scraping/test_dynamic.py @@ -13,6 +13,7 @@ import pytest from bs4 import BeautifulSoup +from ciberwebscan.core.client.ssrf import SSRFBlockedError from ciberwebscan.core.scraping.dynamic import ( BrowserType, DynamicScrapeConfig, @@ -1016,3 +1017,130 @@ async def run_test(): result = asyncio.run(run_test()) assert result.success is True + + +class TestDynamicScraperProxyPolicy: + """Playwright proxy-boundary tests (BUG-014, scenarios 24-26). + + Chromium opens its own connection outside ``HTTPClient``, so the + scraper classifies the proxy before creating the browser context. + """ + + def test_private_proxy_blocked_before_context_creation(self): + mock_pw, mock_browser, mock_context, _ = _create_mock_playwright( + "" + ) + scraper = _create_scraper_with_mocked_browser(mock_pw) + config = DynamicScrapeConfig(selector="body", proxy="http://169.254.169.254:80") + + with pytest.raises(SSRFBlockedError): + asyncio.run(scraper._create_context(config)) + mock_browser.new_context.assert_not_called() + + def test_loopback_proxy_blocked_without_opt_in(self): + mock_pw, mock_browser, _, _ = _create_mock_playwright("") + scraper = _create_scraper_with_mocked_browser(mock_pw) + scraper._browser = mock_browser + config = DynamicScrapeConfig(selector="body", proxy="http://127.0.0.1:8080") + + with pytest.raises(SSRFBlockedError): + asyncio.run(scraper._create_context(config)) + mock_browser.new_context.assert_not_called() + + def test_local_proxy_allowed_with_opt_in(self): + mock_pw, mock_browser, mock_context, _ = _create_mock_playwright( + "" + ) + scraper = _create_scraper_with_mocked_browser(mock_pw) + scraper._browser = mock_browser + config = DynamicScrapeConfig( + selector="body", + proxy="http://127.0.0.1:8080", + allow_local_proxy=True, + ) + + asyncio.run(scraper._create_context(config)) + _, kwargs = mock_browser.new_context.call_args + assert kwargs["proxy"] == {"server": "http://127.0.0.1:8080"} + + def test_imds_stays_blocked_with_opt_in(self): + mock_pw, mock_browser, _, _ = _create_mock_playwright("") + scraper = _create_scraper_with_mocked_browser(mock_pw) + scraper._browser = mock_browser + config = DynamicScrapeConfig( + selector="body", + proxy="http://169.254.169.254:80", + allow_local_proxy=True, + ) + + with pytest.raises(SSRFBlockedError): + asyncio.run(scraper._create_context(config)) + mock_browser.new_context.assert_not_called() + + def test_target_allow_local_does_not_permit_local_proxy(self): + mock_pw, mock_browser, _, _ = _create_mock_playwright("") + scraper = _create_scraper_with_mocked_browser(mock_pw) + scraper._browser = mock_browser + config = DynamicScrapeConfig( + selector="body", proxy="http://10.0.0.5:3128", allow_local=True + ) + + with pytest.raises(SSRFBlockedError): + asyncio.run(scraper._create_context(config)) + mock_browser.new_context.assert_not_called() + + def test_setup_page_registers_http_and_websocket_guards(self): + mock_pw, _, _, mock_page = _create_mock_playwright("") + scraper = _create_scraper_with_mocked_browser(mock_pw) + config = DynamicScrapeConfig(selector="body") + + asyncio.run(scraper._setup_page(mock_page, config)) + mock_page.route.assert_awaited_once() + mock_page.route_web_socket.assert_awaited_once() + + +class TestDynamicScraperWebsocketGuard: + """Routed WebSockets are validated before connect_to_server() (BUG-014). + + Playwright's ``route_web_socket`` keeps a routed socket unconnected + until ``connect_to_server`` is called, so a blocked endpoint never + reaches the network. These tests drive the guard handler directly + with mock WebSocketRoute objects. + """ + + def _run_handler(self, ws, *, allow_local: bool = False): + scraper = DynamicScraper() + asyncio.run(scraper._handle_websocket_route(ws, allow_local=allow_local)) + + def test_private_websocket_blocked(self): + ws = AsyncMock() + ws.url = "ws://169.254.169.254/socket" + + self._run_handler(ws) + ws.close.assert_awaited_once() + assert ws.close.call_args.kwargs["code"] == 1008 + ws.connect_to_server.assert_not_awaited() + + def test_loopback_websocket_blocked_by_default(self): + ws = AsyncMock() + ws.url = "ws://localhost:8000/socket" + + self._run_handler(ws) + ws.close.assert_awaited_once() + ws.connect_to_server.assert_not_awaited() + + def test_public_websocket_connected(self): + ws = AsyncMock() + ws.url = "wss://public.example.com/socket" + + self._run_handler(ws) + ws.connect_to_server.assert_awaited_once() + ws.close.assert_not_awaited() + + def test_loopback_websocket_allowed_with_target_opt_in(self): + ws = AsyncMock() + ws.url = "ws://127.0.0.1:8000/socket" + + self._run_handler(ws, allow_local=True) + ws.connect_to_server.assert_awaited_once() + ws.close.assert_not_awaited() diff --git a/tests/unit/services/test_ssrf_entries.py b/tests/unit/services/test_ssrf_entries.py index b77d43f..27c78cb 100644 --- a/tests/unit/services/test_ssrf_entries.py +++ b/tests/unit/services/test_ssrf_entries.py @@ -167,3 +167,109 @@ def test_check_target_url_opt_in(self) -> None: service._check_target_url("http://169.254.169.254/", allow_local=True) == "http://169.254.169.254/" ) + + +class TestProxyEntrySSRF: + """Malicious per-request proxies are rejected coherently. + + A forbidden proxy must surface as a validation-class error (never as + success, an ambiguous failure or an unexpected 500), for every + service that can receive a proxy from CLI/API/rotation. + """ + + @pytest.mark.parametrize( + "proxy", + [ + "http://127.0.0.1:8080", + "http://169.254.169.254", + "http://10.0.0.5:3128", + ], + ) + def test_scrape_rejects_malicious_proxy(self, proxy: str) -> None: + service = ScrapeService() + + result = service.scrape( + ScrapeOptions( + url="http://public.example.com/", selector="body", proxy=proxy + ) + ) + + assert result.success is False + assert result.error_code == "VALIDATION_ERROR" + assert "ssrf" in (result.error or "").lower() + + def test_analyze_rejects_malicious_proxy(self) -> None: + service = AnalyzeService() + + result = service.analyze( + AnalyzeOptions( + url="http://public.example.com/", + ssl=False, + proxy="http://169.254.169.254", + ) + ) + + assert result.success is False + assert result.error_code == "VALIDATION_ERROR" + assert "ssrf" in (result.error or "").lower() + + def test_attack_rejects_malicious_proxy(self) -> None: + service = AttackService() + service.app_config.attack.enabled = True + service.app_config.attack.whitelist = [] + + with pytest.raises(ValidationError) as exc_info: + service.attack( + AttackOptions( + url="http://public.example.com/", + user_consent=True, + xss=True, + proxy="http://127.0.0.1:8080", + ) + ) + + assert "ssrf" in str(exc_info.value).lower() + + def test_quick_scan_rejects_malicious_proxy(self) -> None: + service = QuickService() + + with pytest.raises(ValidationError) as exc_info: + service.quick_scan( + QuickOptions( + url="http://public.example.com/", + preset="low", + proxy="http://127.0.0.1:8080", + ) + ) + + assert "ssrf" in str(exc_info.value).lower() + + def test_scrape_config_proxy_is_also_policed(self) -> None: + """A forbidden proxy from persistent config fails the same way. + + ``model_construct`` simulates configuration that reached the + singleton without the ingestion validator (legacy state), so the + usage boundary inside the service is what gets exercised. + """ + from ciberwebscan.config.loader import get_config + from ciberwebscan.config.models import ProxyConfig + + get_config().http.proxy = ProxyConfig.model_construct( + http="http://169.254.169.254:80", + https=None, + socks5=None, + rotate=False, + rotation_interval=10, + proxy_list=None, + allow_local=False, + use_env=False, + ) + service = ScrapeService() + + result = service.scrape( + ScrapeOptions(url="http://public.example.com/", selector="body") + ) + + assert result.success is False + assert result.error_code == "VALIDATION_ERROR" + assert "ssrf" in (result.error or "").lower() diff --git a/uv.lock b/uv.lock index 127b3f3..78ec07f 100644 --- a/uv.lock +++ b/uv.lock @@ -192,7 +192,7 @@ dependencies = [ { name = "beautifulsoup4" }, { name = "click" }, { name = "cryptography" }, - { name = "httpx", extra = ["http2"] }, + { name = "httpx", extra = ["http2", "socks"] }, { name = "lxml" }, { name = "orjson" }, { name = "packaging" }, @@ -228,7 +228,7 @@ requires-dist = [ { name = "click", specifier = "==8.5.0" }, { name = "cryptography", specifier = "==50.0.0" }, { name = "fastapi", marker = "extra == 'api'", specifier = "==0.140.0" }, - { name = "httpx", extras = ["http2"], specifier = "==0.28.1" }, + { name = "httpx", extras = ["http2", "socks"], specifier = "==0.28.1" }, { name = "lxml", specifier = "==6.1.1" }, { name = "orjson", specifier = "==3.11.9" }, { name = "packaging", specifier = "==24.0" }, @@ -655,6 +655,9 @@ wheels = [ http2 = [ { name = "h2" }, ] +socks = [ + { name = "socksio" }, +] [[package]] name = "hyperframe" @@ -1492,6 +1495,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755, upload-time = "2023-10-24T04:13:38.866Z" }, ] +[[package]] +name = "socksio" +version = "1.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f8/5c/48a7d9495be3d1c651198fd99dbb6ce190e2274d0f28b9051307bdec6b85/socksio-1.0.0.tar.gz", hash = "sha256:f88beb3da5b5c38b9890469de67d0cb0f9d494b78b106ca1845f96c10b91c4ac", size = 19055, upload-time = "2020-04-17T15:50:34.664Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/37/c3/6eeb6034408dac0fa653d126c9204ade96b819c936e136c5e8a6897eee9c/socksio-1.0.0-py3-none-any.whl", hash = "sha256:95dc1f15f9b34e8d7b16f06d74b8ccf48f609af32ab33c608d08761c5dcbb1f3", size = 12763, upload-time = "2020-04-17T15:50:31.878Z" }, +] + [[package]] name = "soupsieve" version = "2.9.2"