diff --git a/.env.example b/.env.example index b6a6fa4..9de2fd4 100644 --- a/.env.example +++ b/.env.example @@ -53,6 +53,12 @@ CIBERWEBSCAN_CACHE_ENABLED=true # -> cache.enabled CIBERWEBSCAN_API_AUTH_SERVER_SECRET= # -> api.auth.server_secret (HMAC for logs only) CIBERWEBSCAN_DOWNLOAD_REQUIRE_SAME_USER=true # -> download.require_same_user (token bound to issuing key) +# Reverse proxies (client identification). Empty = X-Forwarded-For always ignored. +# Declare proxy endpoints only; the proxy must overwrite or append (rightmost) +# X-Forwarded-For. The API runs Uvicorn with proxy_headers=False, so +# FORWARDED_ALLOW_IPS has no effect on client identification. +# CIBERWEBSCAN_API_TRUSTED_PROXIES=127.0.0.1 # -> api.trusted_proxies (IPs/CIDRs) + # Analysis / CVE API keys # NVD_API_KEY / VULNERS_API_KEY are read directly by the CVE clients, or use: # CIBERWEBSCAN_ANALYSIS_CVE_NVD_API_KEY=... diff --git a/AGENTS.md b/AGENTS.md index 8b47687..43a7759 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -203,7 +203,7 @@ X-API-Key: ### Middleware - **RequestLoggingMiddleware**: Logs method, path, status, duration for every request -- **RateLimitingMiddleware**: In-memory sliding window per IP, returns 429 with `Retry-After` header +- **RateLimitingMiddleware**: In-memory per-client counter in a fixed 60s window, returns 429 with `Retry-After` header. The client identity is the connection peer; `X-Forwarded-For` is honoured only when that peer is declared in `api.trusted_proxies` (empty by default = header always ignored), resolved right-to-left hop by hop (`api/client_ip.py`, shared with auth log attribution) ### Models @@ -424,6 +424,7 @@ def test_analyze_route(mock_service_class): | API routes | `src/ciberwebscan/api/routes/` | | API models | `src/ciberwebscan/api/models/requests.py`, `responses.py` | | API auth | `src/ciberwebscan/api/auth.py` | +| Client IP resolver | `src/ciberwebscan/api/client_ip.py` | | API middleware | `src/ciberwebscan/api/middleware.py` | | API download helper | `src/ciberwebscan/api/helpers/download_helper.py` | | Pydantic config | `src/ciberwebscan/config/models.py` | diff --git a/docs/API.md b/docs/API.md index 9a5725e..49e96dc 100644 --- a/docs/API.md +++ b/docs/API.md @@ -922,6 +922,8 @@ each use: `GET /api/download/{token}`, so toggling it applies to the next download request. - **API auth `server_secret`** (`api.auth.server_secret`): read per request but used only for log identifiers; the auto-generated default is fixed for the lifetime of the process. +- **Trusted proxies** (`api.trusted_proxies`): re-read on every request when resolving the client + IP for rate limiting and auth log attribution, so proxies can be declared or removed at runtime. - **HTTP client settings** (timeouts, retries, rate limits): apply to HTTP clients created after the change; operations already in flight keep the settings they started with. - **Scan defaults** (scraping, analysis, attack options): apply to operations started after the @@ -1005,6 +1007,9 @@ CIBERWEBSCAN_API_AUTH_SERVER_SECRET="" CIBERWEBSCAN_API_RATE_LIMIT_ENABLED=true CIBERWEBSCAN_API_RATE_LIMIT_REQUESTS_PER_MINUTE=60 +# Client identification (reverse proxies allowed to set X-Forwarded-For) +CIBERWEBSCAN_API_TRUSTED_PROXIES="127.0.0.1" + # CORS CIBERWEBSCAN_API_CORS_ORIGINS="http://localhost:3000,https://example.com" CIBERWEBSCAN_API_CORS_ALLOW_CREDENTIALS=true @@ -1027,6 +1032,68 @@ The API includes rate limiting to prevent abuse: - **Configurable** via `CIBERWEBSCAN_API_RATE_LIMIT_REQUESTS_PER_MINUTE` - **429 status code** returned when exceeded - **Retry-After header** included in rate limit responses +- **Per-client counter**: each client identity has its own bucket (see + [Client identification](#client-identification-and-trusted-proxies) below) + +### Client identification and trusted proxies + +Clients are identified by their **connection peer address**. The +`X-Forwarded-For` header is **ignored by default** (secure default) and is only +honoured when the peer address is declared in `api.trusted_proxies`: + +```bash +CIBERWEBSCAN_API_TRUSTED_PROXIES="127.0.0.1" # single proxy +CIBERWEBSCAN_API_TRUSTED_PROXIES="10.0.0.0/8,::1" # CIDR and IPv6 accepted +``` + +Rules applied (`ciberwebscan.api.client_ip.resolve_client_ip`, shared by the +rate limiter and auth log attribution): + +1. Peer missing or not parseable as an IP (e.g. unix socket) → that fallback + identity is used and `X-Forwarded-For` is never consulted. +2. `api.trusted_proxies` empty (default) → the header is ignored. +3. Peer outside the trusted list → the header is ignored, even when the list + is non-empty. +4. Peer is a trusted proxy → the chain is walked **from right to left**; the + first hop not vouching for by a trusted proxy is the client. Every hop is + validated with `ipaddress` (IPv4, IPv6 and IPv4-mapped forms are + normalized); a malformed hop stops the walk and the last trusted hop wins. + Entries are validated as IP addresses or CIDR networks at configuration + time and stored in canonical network form. + +Reverse proxy requirements for correct attribution: + +- The proxy must **overwrite** `X-Forwarded-For` or **append** (rightmost) the + address it observed. Transparent/pass-through proxies that do not touch the + header must **not** be declared. +- Declare only the proxy's own endpoint addresses (`/32` / `/128`), never a + range that also contains end clients — a client inside a trusted range could + forge a chain that the walk would believe. + +Effect of not configuring it behind a reverse proxy: all clients share the +proxy's peer address, so they share a single rate-limit bucket (60 rpm global) +and logs show the proxy IP. Spoofing is not possible either way. + +Interaction with Uvicorn's proxy handling: both launch paths of this API +(`ciberwebscan api` and `python -m ciberwebscan.api.app`) start Uvicorn with +`proxy_headers=False`, so Uvicorn never rewrites `request.client` from +forwarded headers and `FORWARDED_ALLOW_IPS` is irrelevant — the resolver +above is the single authority for client identity. Deployments that start +the app under a different ASGI server (or a custom Uvicorn invocation) with +forwarded-header trust enabled — especially `FORWARDED_ALLOW_IPS=*`, which +installs the _leftmost_, client-controlled header value as the peer — void +this policy and are not supported. The `Forwarded` header (RFC 7239) is not +honoured by this API. + +Invalid values: `api.trusted_proxies` entries must be valid IP addresses or +CIDR networks. Catch-all networks (`0.0.0.0/0`, `::/0`, and IPv4-mapped +ranges covering the whole mapped IPv4 space), IPv4-mapped networks wider +than `/96`, and IPv6 zone identifiers (`fe80::1%eth0`) are rejected: they +would either trust every peer as a proxy or over-match silently. Any +invalid value at startup is reported loudly and, like any +invalid configuration value, makes the loader fall back to the default +configuration (see [Configuration](CONFIGURATION.md)); `PUT /api/config` and +`ciberwebscan config set` validate without mutating. ### CORS diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 60083e0..bc65a84 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -77,6 +77,7 @@ Examples that work out of the box: - `CIBERWEBSCAN_ANALYSIS_CVE_NVD_API_KEY` → `analysis.cve.nvd_api_key` - `CIBERWEBSCAN_CACHE_MAX_SIZE_MB` → `cache.max_size_mb` - `CIBERWEBSCAN_API_RATE_LIMIT_REQUESTS_PER_MINUTE` → `api.rate_limit.requests_per_minute` +- `CIBERWEBSCAN_API_TRUSTED_PROXIES` → `api.trusted_proxies` (comma-separated IPs/CIDRs) - `CIBERWEBSCAN_API_CORS_ORIGINS` → `api.cors_origins` - `CIBERWEBSCAN_API_CORS_ALLOW_CREDENTIALS` → `api.cors_allow_credentials` - `CIBERWEBSCAN_API_CORS_ALLOW_METHODS` → `api.cors_allow_methods` @@ -556,6 +557,7 @@ Configure the FastAPI server, authentication, and CORS settings. "enabled": true, "requests_per_minute": 60 }, + "trusted_proxies": [], "cors_origins": ["http://localhost:3000"], "cors_allow_credentials": true, "cors_allow_methods": ["GET", "POST", "PUT", "DELETE", "PATCH"], @@ -575,6 +577,7 @@ Configure the FastAPI server, authentication, and CORS settings. | `api.auth.server_secret` | `""` | Secret for HMAC log obfuscation only (see below) | | `api.rate_limit.enabled` | `true` | Rate limiting enabled by default | | `api.rate_limit.requests_per_minute` | `60` | Max requests per minute (1-10000) | +| `api.trusted_proxies` | `[]` | Reverse proxies allowed to set `X-Forwarded-For` | | `api.cors_origins` | `[]` | Allowed CORS origins (empty = no cross-origin) | | `api.cors_allow_credentials` | `false` | Allow credentials in cross-origin requests | | `api.cors_allow_methods` | see below | HTTP methods allowed for CORS | @@ -585,6 +588,54 @@ Configure the FastAPI server, authentication, and CORS settings. **Default `cors_allow_headers`:** `["Authorization", "Content-Type", "X-API-Key"]` +#### `api.trusted_proxies` (client identification) + +Clients of the API are identified by their **connection peer address**. The +`X-Forwarded-For` header is ignored unless the peer is declared in +`api.trusted_proxies`, so the default (`[]`) is the safe choice for direct +exposure: no header can forge a client identity, bypass the rate limiter, or +consume the quota attributed to another address. + +- **Accepts**: IP addresses (IPv4/IPv6) or CIDR networks, as a list or a + comma-separated string. Entries are validated with `ipaddress` and stored in + canonical network form (`127.0.0.1` → `127.0.0.1/32`); host bits in a CIDR + are masked with a warning (`10.0.0.1/8` → `10.0.0.0/8`); IPv4-mapped IPv6 + networks (`::ffff:10.0.0.0/104`) are converted to IPv4 (`10.0.0.0/8`). +- **Invalid values** (e.g. `not-an-ip`, `1.2.3.4/33`, catch-all networks + like `0.0.0.0/0` or `::/0`, IPv4-mapped ranges covering the whole mapped + IPv4 space such as `::ffff:10.0.0.0/96`, and IPv6 zone identifiers like + `fe80::1%eth0`) raise a validation + error. Via `PUT /api/config` or `config set` this means a 400 / non-zero + exit without mutating the configuration; at startup (config file or + environment), the loader's pre-existing policy applies: it logs an error and + runs on the **entire default configuration**. +- **Read per request** — changes apply without restarting the server (see + [Runtime changes vs restart-required + settings](#runtime-changes-vs-restart-required-settings)). +- **Trust model**: declare only the proxy endpoints (e.g. `127.0.0.1`, + `10.0.0.1/32`). A range that also contains end clients lets a client inside + it forge a forwarded chain that the right-to-left walk would believe. Only + proxies that **overwrite** or **append** (rightmost) `X-Forwarded-For` may be + declared; pass-through proxies must not be. +- **Behind a reverse proxy without configuring this**: every request is + attributed to the proxy's own address, so all users share a single + rate-limit bucket and logs name the proxy — no spoofing, but quota sharing. +- **Uvicorn interaction**: both launch paths (`ciberwebscan api` and + `python -m ciberwebscan.api.app`) start Uvicorn with `proxy_headers=False`, + so Uvicorn never rewrites `request.client` from forwarded headers and + `FORWARDED_ALLOW_IPS` is irrelevant — the `api.trusted_proxies` policy is + the single authority. Running the app under a different ASGI server with + forwarded-header trust enabled (especially `FORWARDED_ALLOW_IPS=*`, which + installs the client-controlled leftmost header value as the peer) voids + this policy and is not supported. The `Forwarded` header is ignored. + +```yaml +api: + trusted_proxies: + - 127.0.0.1 + - 10.0.0.0/8 +``` + #### `api.auth.server_secret` (log obfuscation only) `api.auth.server_secret` (default `""`) is used **only** to obfuscate API keys in logs: log identifiers are derived from HMAC-SHA256 of the key under this secret. It takes **no part in authorization** — the authenticated identity (`GET /api/auth/me`) and download-token ownership are derived from a stable key identifier instead, so downloads work regardless of this value. @@ -880,7 +931,8 @@ configuration object in place, so every `get_config()` consumer sees them immedi - **Applied at next use (no restart):** `api.auth.api_keys` (API authentication is checked per request), `api.auth.server_secret` (log obfuscation only, read per request — the auto-generated - default stays stable until the process restarts) and `download.require_same_user` (checked on + default stays stable until the process restarts), `api.trusted_proxies` (client identity for + rate limiting and auth logs is resolved per request), and `download.require_same_user` (checked on every `GET /api/download/{token}`), HTTP client settings used when a client is created for a new operation, and scraping/analysis/attack defaults read per scan. - **Require a restart:** `api.cors.*`, `api.rate_limit.*`, `download.cleanup_interval_seconds`, diff --git a/src/ciberwebscan/api/app.py b/src/ciberwebscan/api/app.py index 22e04ef..edcc56e 100644 --- a/src/ciberwebscan/api/app.py +++ b/src/ciberwebscan/api/app.py @@ -153,4 +153,5 @@ async def general_exception_handler( port=app_config.api.port, reload=True, log_level=app_config.logging.level.lower(), + proxy_headers=False, ) diff --git a/src/ciberwebscan/api/auth.py b/src/ciberwebscan/api/auth.py index b932cb7..63a9a70 100644 --- a/src/ciberwebscan/api/auth.py +++ b/src/ciberwebscan/api/auth.py @@ -14,6 +14,7 @@ from fastapi.security import APIKeyHeader from pydantic import BaseModel +from ciberwebscan.api.client_ip import resolve_client_ip from ciberwebscan.config.loader import get_config from ciberwebscan.utils.logging import _resolve_server_secret, mask_key_for_logging @@ -147,11 +148,8 @@ async def verify_api_key( def _get_client_ip(request: Request) -> str: - """Extract client IP from request, handling proxies.""" - forwarded = request.headers.get("X-Forwarded-For", "").split(",")[0].strip() - if forwarded: - return forwarded - return request.client.host if request.client else "unknown" + """Resolve client IP for log attribution using the shared proxy policy.""" + return resolve_client_ip(request) async def get_current_user( diff --git a/src/ciberwebscan/api/client_ip.py b/src/ciberwebscan/api/client_ip.py new file mode 100644 index 0000000..52681c9 --- /dev/null +++ b/src/ciberwebscan/api/client_ip.py @@ -0,0 +1,137 @@ +""" +Shared client IP resolution for the API layer. + +Client identity comes from the connection peer (``request.client``). +``X-Forwarded-For`` is only consulted when the peer belongs to the declared +``api.trusted_proxies`` list, in which case the forwarded chain is walked from +right to left until the first address that a trusted proxy did not vouch for. +With no trusted proxies configured (the default) the header is always ignored. +""" + +from __future__ import annotations + +import ipaddress +import logging +from collections.abc import Callable + +from fastapi import Request + +from ciberwebscan.config.loader import get_config + +logger = logging.getLogger(__name__) + +_UNKNOWN_CLIENT = "unknown" + +_IPv4Address = ipaddress.IPv4Address +_IPv6Address = ipaddress.IPv6Address +_IPv4Network = ipaddress.IPv4Network +_IPv6Network = ipaddress.IPv6Network + +_ClientAddress = _IPv4Address | _IPv6Address +_ClientNetwork = _IPv4Network | _IPv6Network + + +def _unwrap_mapped(address: _ClientAddress) -> _ClientAddress: + """Return the IPv4 form of an IPv4-mapped IPv6 address (else itself).""" + if isinstance(address, _IPv6Address) and address.ipv4_mapped is not None: + return address.ipv4_mapped + return address + + +def _parse_ip(value: str) -> _ClientAddress | None: + """Parse a single hop, or return None when it is not a valid IP.""" + try: + return _unwrap_mapped(ipaddress.ip_address(value.strip())) + except ValueError: + return None + + +def _trusted_networks() -> list[_ClientNetwork]: + """Parse ``api.trusted_proxies`` for this request (fail closed).""" + try: + entries = get_config().api.trusted_proxies + except Exception: + # Any unexpected config failure must not turn into a 500 or an + # open header: empty list = X-Forwarded-For stays ignored. + logger.warning( + "Could not load api.trusted_proxies; ignoring X-Forwarded-For", + exc_info=True, + ) + return [] + networks: list[_ClientNetwork] = [] + for entry in entries: + try: + networks.append(ipaddress.ip_network(str(entry), strict=False)) + except ValueError: + logger.warning("Ignoring invalid api.trusted_proxies entry: %r", entry) + return networks + + +def _forwarded_for(request: Request) -> str: + """Join all X-Forwarded-For header lines (Starlette semantics).""" + headers = request.headers + # Starlette's Headers exposes getlist(); mapping-style test doubles + # may only provide get(). + getlist: Callable[[str], list[str]] | None = getattr(headers, "getlist", None) + if getlist is not None: + return ", ".join(getlist("x-forwarded-for")) + value = headers.get("X-Forwarded-For", "") + return value if isinstance(value, str) else "" + + +def resolve_client_ip(request: Request) -> str: + """ + Resolve the client identity for rate limiting and log attribution. + + Args: + request: The incoming request. + + Returns: + The client IP as a string. The connection peer wins whenever + ``X-Forwarded-For`` cannot be trusted: empty ``api.trusted_proxies`` + (default), peer outside that list, or peer not parseable as an IP + (in which case the shared ``"unknown"`` bucket is used rather than + echoing a non-IP peer string). Behind trusted proxies the chain is + validated hop by hop from right to left and a malformed entry stops + the walk (last trusted hop wins). + """ + peer = request.client.host if request.client else None + peer_ip = _parse_ip(peer) if isinstance(peer, str) and peer else None + if peer_ip is None: + # No usable peer (unix socket, test client, missing scope): never + # consult a client-controlled header for identity, and never echo + # an arbitrary peer string (it could be client-controlled under a + # misconfigured upstream proxy). + return _UNKNOWN_CLIENT + + forwarded = _forwarded_for(request) + if not forwarded.strip(): + return str(peer_ip) + + trusted = _trusted_networks() + if not trusted: + logger.debug( + "Ignoring X-Forwarded-For: no trusted proxies configured (%s)", peer_ip + ) + return str(peer_ip) + if not any(peer_ip in network for network in trusted): + logger.debug( + "Ignoring X-Forwarded-For: peer %s is not a trusted proxy", peer_ip + ) + return str(peer_ip) + + # Peer is a declared proxy: walk the chain right to left (appended hops + # are written by our proxies; the client controls only the left end). + result = str(peer_ip) + for entry in reversed(forwarded.split(",")): + entry = entry.strip() + if not entry: + break + hop = _parse_ip(entry) + if hop is None: + # Malformed hop: stop before touching client-controlled data. + break + if not any(hop in network for network in trusted): + return str(hop) + result = str(hop) + return result diff --git a/src/ciberwebscan/api/middleware.py b/src/ciberwebscan/api/middleware.py index f8d0c3d..bea8159 100644 --- a/src/ciberwebscan/api/middleware.py +++ b/src/ciberwebscan/api/middleware.py @@ -14,6 +14,8 @@ from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware +from ciberwebscan.api.client_ip import resolve_client_ip + logger = logging.getLogger(__name__) @@ -56,20 +58,19 @@ def __init__(self, app, requests_per_minute: int): self.window = 0 async def dispatch(self, request: Request, call_next) -> Response: - # Identificación de cliente (Prioriza X-Forwarded-For) - client_ip = request.headers.get("X-Forwarded-For", "").split(",")[ - 0 - ].strip() or (request.client.host if request.client else "unknown") + # Client identification: X-Forwarded-For is trusted only if the + # connection peer is a declared reverse proxy (api.trusted_proxies). + client_ip = resolve_client_ip(request) now = time.time() current_window = int(now // 60) - # Rotación de ventana + # Rotate the window. if current_window != self.window: self.counts.clear() self.window = current_window - # Verificación de límite + # Check the rate limit. if self.counts[client_ip] >= self.limit: retry_after = 60 - int(now % 60) logger.warning(f"Rate limit exceeded: {client_ip}") diff --git a/src/ciberwebscan/cli/commands/api.py b/src/ciberwebscan/cli/commands/api.py index 2d4a277..db353d8 100644 --- a/src/ciberwebscan/cli/commands/api.py +++ b/src/ciberwebscan/cli/commands/api.py @@ -75,6 +75,7 @@ def run_api( port=server_port, reload=reload, log_level=log_level, + proxy_headers=False, ) except ImportError: print_error( diff --git a/src/ciberwebscan/config/models.py b/src/ciberwebscan/config/models.py index 8f2fac1..ebb21eb 100644 --- a/src/ciberwebscan/config/models.py +++ b/src/ciberwebscan/config/models.py @@ -8,10 +8,14 @@ from __future__ import annotations +import ipaddress +import logging from typing import Annotated, Literal from pydantic import BaseModel, Field, HttpUrl, field_validator +logger = logging.getLogger(__name__) + # ============================================================================= # HTTP Client Configuration # ============================================================================= @@ -371,6 +375,101 @@ class APIConfig(BaseModel): port: Annotated[int, Field(ge=1, le=65535)] = 8000 auth: APIAuthConfig = Field(default_factory=lambda: APIAuthConfig()) rate_limit: APIRateLimitConfig = Field(default_factory=lambda: APIRateLimitConfig()) + trusted_proxies: list[str] = Field( + default=[], + description=( + "IP addresses or CIDR networks of the reverse proxies allowed to " + "set X-Forwarded-For. Empty list (default) = the header is always " + "ignored and clients are identified by their connection peer. " + "Declare proxy endpoint addresses only; ranges that also contain " + "clients would let them forge the forwarded chain." + ), + ) + + @field_validator("trusted_proxies", mode="before") + @classmethod + def parse_trusted_proxies(cls, v: object) -> list[str]: + """Validate proxy entries as IPv4/IPv6 addresses or CIDR networks.""" + if isinstance(v, str): + items: list = v.split(",") + elif isinstance(v, (list, tuple)): + items = list(v) + else: + # Scalar (e.g. a numeric env var value) — still validated below. + items = [v] + + networks: list[str] = [] + for item in items: + entry = str(item).strip() + if not entry: + continue + if "%" in entry: + raise ValueError( + f"IPv6 zone identifiers are not allowed in " + f"api.trusted_proxies: {entry!r}; declare the proxy by " + f"routable address/CIDR" + ) + try: + network = ipaddress.ip_network(entry, strict=False) + except ValueError as exc: + raise ValueError( + f"Invalid IP address or CIDR network in " + f"api.trusted_proxies: {entry!r}" + ) from exc + + converted_from_mapped = False + address_part = entry.split("/", 1)[0] + address = ipaddress.ip_address(address_part) + if isinstance(address, ipaddress.IPv6Address): + mapped = address.ipv4_mapped + if mapped is not None: + # /96 covers the entire mapped IPv4 space, i.e. every + # IPv4 peer — never accept it as a "proxy range". + if network.prefixlen <= 96: + raise ValueError( + f"IPv4-mapped IPv6 network {entry!r} in " + f"api.trusted_proxies covers the whole mapped " + f"IPv4 range; use IPv4 notation instead" + ) + network = ipaddress.IPv4Network( + (mapped, network.prefixlen - 96), + strict=False, + ) + converted_from_mapped = True + + canonical = str(network) + if network.prefixlen == 0: + raise ValueError( + f"Catch-all network {entry!r} in api.trusted_proxies " + f"would trust every peer as a proxy and defeat client " + f"identification; declare the proxy addresses explicitly" + ) + if canonical != entry: + if converted_from_mapped: + logger.debug( + "api.trusted_proxies entry %r (IPv4-mapped) normalized to %r", + entry, + canonical, + ) + else: + try: + ipaddress.ip_network(entry, strict=True) + except ValueError: + logger.warning( + "api.trusted_proxies entry %r has host bits set; " + "using network %r", + entry, + canonical, + ) + else: + logger.debug( + "api.trusted_proxies entry %r normalized to %r", + entry, + canonical, + ) + networks.append(canonical) + + return networks # CORS — secure defaults cors_origins: list[str] = Field( diff --git a/tests/unit/api/test_auth.py b/tests/unit/api/test_auth.py index 780ca11..3b0321f 100644 --- a/tests/unit/api/test_auth.py +++ b/tests/unit/api/test_auth.py @@ -14,12 +14,14 @@ from fastapi.testclient import TestClient from ciberwebscan.api.auth import ( + _get_client_ip, _secure_compare_key, generate_api_key, get_auth_config, verify_api_key, ) from ciberwebscan.api.routes.auth import router as auth_router +from ciberwebscan.config.models import APIConfig, AppConfig from ciberwebscan.export.models import ( AnalysisReport, AttackResult, @@ -404,6 +406,47 @@ async def test_different_keys_get_different_identifiers(self): assert first.identifier != second.identifier +# ============================================================================= +# Client IP Attribution Tests +# ============================================================================= + + +class TestClientIPAttribution: + """Auth logs must attribute the IP under the shared trust policy.""" + + def test_forwarded_for_ignored_for_untrusted_peer(self): + request = _create_mock_request(client_ip="10.0.0.9") + request.headers = {"X-Forwarded-For": "203.0.113.7"} + + with patch("ciberwebscan.api.client_ip.get_config", return_value=AppConfig()): + assert _get_client_ip(request) == "10.0.0.9" + + def test_forwarded_for_ignored_when_list_configured_but_peer_untrusted( + self, + ): + request = _create_mock_request(client_ip="10.0.0.9") + request.headers = {"X-Forwarded-For": "203.0.113.7"} + config = AppConfig(api=APIConfig(trusted_proxies=["10.0.0.50"])) + + with patch("ciberwebscan.api.client_ip.get_config", return_value=config): + assert _get_client_ip(request) == "10.0.0.9" + + def test_forwarded_for_resolved_behind_trusted_proxy(self): + request = _create_mock_request(client_ip="10.0.0.50") + request.headers = {"X-Forwarded-For": "203.0.113.7"} + config = AppConfig(api=APIConfig(trusted_proxies=["10.0.0.50"])) + + with patch("ciberwebscan.api.client_ip.get_config", return_value=config): + assert _get_client_ip(request) == "203.0.113.7" + + def test_missing_forwarded_for_keeps_peer(self): + request = _create_mock_request(client_ip="10.0.0.50") + config = AppConfig(api=APIConfig(trusted_proxies=["10.0.0.50"])) + + with patch("ciberwebscan.api.client_ip.get_config", return_value=config): + assert _get_client_ip(request) == "10.0.0.50" + + # ============================================================================= # Auth Endpoint Tests # ============================================================================= diff --git a/tests/unit/api/test_client_ip.py b/tests/unit/api/test_client_ip.py new file mode 100644 index 0000000..ad378d5 --- /dev/null +++ b/tests/unit/api/test_client_ip.py @@ -0,0 +1,249 @@ +""" +Tests for the shared client IP resolver. + +Covers peer fallback, the trusted-proxy policy, right-to-left chain +resolution, malformed hops, and IPv4/IPv6/CIDR/IPv4-mapped handling. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from starlette.requests import Request + +from ciberwebscan.api.client_ip import resolve_client_ip +from ciberwebscan.config.models import APIConfig, AppConfig + +pytestmark = pytest.mark.unit + +# ============================================================================= +# Helpers +# ============================================================================= + + +def _request( + peer: tuple[str, int] | None = None, + forwarded_for: str | None = None, + extra_forwarded_lines: list[str] | None = None, +) -> Request: + """Build a bare ASGI Request with an explicit peer and XFF chain.""" + headers: list[tuple[bytes, bytes]] = [] + if forwarded_for is not None: + headers.append((b"x-forwarded-for", forwarded_for.encode())) + for line in extra_forwarded_lines or []: + headers.append((b"x-forwarded-for", line.encode())) + scope = { + "type": "http", + "method": "GET", + "path": "/", + "headers": headers, + "client": peer, + } + return Request(scope) + + +def _with_config(config: AppConfig): + return patch("ciberwebscan.api.client_ip.get_config", return_value=config) + + +def _trusted(*entries: str) -> AppConfig: + return AppConfig(api=APIConfig(trusted_proxies=list(entries))) + + +# ============================================================================= +# Peer Fallback / Secure Default +# ============================================================================= + + +class TestPeerFallback: + """Without trusted proxies the peer is the only identity.""" + + def test_peer_without_forwarded_for(self): + assert resolve_client_ip(_request(("10.0.0.1", 50000))) == "10.0.0.1" + + def test_forwarded_for_ignored_with_default_config(self): + request = _request(("10.0.0.1", 50000), "203.0.113.7") + with _with_config(AppConfig()): + assert resolve_client_ip(request) == "10.0.0.1" + + def test_empty_forwarded_for_header(self): + request = _request(("10.0.0.1", 50000), "") + assert resolve_client_ip(request) == "10.0.0.1" + + def test_whitespace_only_forwarded_for_header(self): + request = _request(("10.0.0.1", 50000), " ") + assert resolve_client_ip(request) == "10.0.0.1" + + def test_missing_client_scope_falls_back_to_unknown(self): + request = _request(None, "203.0.113.7") + assert resolve_client_ip(request) == "unknown" + + def test_unparseable_peer_never_trusts_forwarded_for(self): + request = _request(("testclient", 50000), "203.0.113.7") + assert resolve_client_ip(request) == "unknown" + + def test_empty_peer_host_falls_back_to_unknown(self): + request = _request(("", 50000), "203.0.113.7") + assert resolve_client_ip(request) == "unknown" + + +# ============================================================================= +# Trusted Proxy Policy +# ============================================================================= + + +class TestTrustedProxyPolicy: + """X-Forwarded-For is honoured only for declared proxy peers.""" + + def test_empty_peer_ignores_header_even_when_list_configured(self): + """A missing peer address collapses to the shared 'unknown' bucket.""" + request = _request(("", 50000), "203.0.113.7") + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "unknown" + + def test_peer_outside_configured_list_is_not_trusted(self): + request = _request(("192.168.9.9", 50000), "203.0.113.7") + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "192.168.9.9" + + def test_cidr_entry_matches_peer_inside_range(self): + request = _request(("10.4.5.6", 50000), "203.0.113.7") + with _with_config(_trusted("10.0.0.0/8")): + assert resolve_client_ip(request) == "203.0.113.7" + + def test_cidr_entry_rejects_peer_outside_range(self): + request = _request(("11.4.5.6", 50000), "203.0.113.7") + with _with_config(_trusted("10.0.0.0/8")): + assert resolve_client_ip(request) == "11.4.5.6" + + def test_ipv6_cidr_entry_matches_peer_inside_range(self): + request = _request(("2001:db8::7", 50000), "203.0.113.7") + with _with_config(_trusted("2001:db8::/32")): + assert resolve_client_ip(request) == "203.0.113.7" + + def test_ipv6_cidr_entry_rejects_peer_outside_range(self): + request = _request(("2001:db9::7", 50000), "203.0.113.7") + with _with_config(_trusted("2001:db8::/32")): + assert resolve_client_ip(request) == "2001:db9::7" + + def test_trusted_peer_without_header_keeps_peer(self): + request = _request(("10.0.0.50", 50000)) + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "10.0.0.50" + + def test_invalid_config_entries_are_ignored_at_runtime(self): + """Fail closed if an invalid entry reaches the resolver.""" + broken = SimpleNamespace(api=SimpleNamespace(trusted_proxies=["garbage"])) + with patch("ciberwebscan.api.client_ip.get_config", return_value=broken): + request = _request(("10.0.0.9", 50000), "203.0.113.7") + assert resolve_client_ip(request) == "10.0.0.9" + + +# ============================================================================= +# Chain Resolution (right to left) +# ============================================================================= + + +class TestChainResolution: + """Validated traversal from the nearest hop towards the client.""" + + def test_rightmost_untrusted_hop_is_the_client(self): + request = _request(("10.0.0.50", 50000), "1.1.1.1, 2.2.2.2") + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "2.2.2.2" + + def test_trusted_hops_are_skipped(self): + request = _request(("10.0.0.50", 50000), "203.0.113.5, 10.0.0.49, 10.0.0.48") + with _with_config(_trusted("10.0.0.50", "10.0.0.49", "10.0.0.48")): + assert resolve_client_ip(request) == "203.0.113.5" + + def test_fully_trusted_chain_returns_leftmost_trusted_hop(self): + request = _request(("10.0.0.50", 50000), "10.0.0.5, 10.0.0.6") + with _with_config(_trusted("10.0.0.50", "10.0.0.5", "10.0.0.6")): + assert resolve_client_ip(request) == "10.0.0.5" + + def test_malformed_hop_stops_the_walk(self): + """A bad hop never routes the walk into client-controlled data.""" + request = _request(("10.0.0.50", 50000), "1.1.1.1, garbage") + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "10.0.0.50" + + def test_empty_entry_stops_the_walk(self): + request = _request(("10.0.0.50", 50000), "1.1.1.1, ") + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "10.0.0.50" + + def test_port_suffixed_hop_is_malformed(self): + request = _request(("10.0.0.50", 50000), "1.1.1.1, 10.0.0.5:8080") + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "10.0.0.50" + + def test_duplicate_header_lines_are_joined(self): + """Starlette exposes repeated XFF lines via getlist; join them. + + The forged identity sits on the first line only: a get()-based + resolver would return the trusted hop 10.0.0.49 instead. + """ + request = _request( + ("10.0.0.50", 50000), + forwarded_for="10.0.0.49", + extra_forwarded_lines=["203.0.113.5"], + ) + with _with_config(_trusted("10.0.0.50", "10.0.0.49")): + assert resolve_client_ip(request) == "203.0.113.5" + + +# ============================================================================= +# Address Families / Normalization +# ============================================================================= + + +class TestAddressFamilies: + """IPv4, IPv6, CIDR and IPv4-mapped addresses behave consistently.""" + + def test_ipv6_peer(self): + request = _request(("::1", 50000), "2001:db8::1") + assert resolve_client_ip(request) == "::1" + + def test_ipv6_trusted_proxy_resolves_ipv6_client(self): + request = _request(("::1", 50000), "2001:db8::5, ::1") + with _with_config(_trusted("::1")): + assert resolve_client_ip(request) == "2001:db8::5" + + def test_ipv4_mapped_peer_is_normalized(self): + request = _request(("::ffff:10.0.0.9", 50000)) + assert resolve_client_ip(request) == "10.0.0.9" + + def test_ipv4_mapped_peer_matches_ipv4_cidr(self): + request = _request(("::ffff:10.0.0.9", 50000), "8.8.8.8") + with _with_config(_trusted("10.0.0.0/8")): + assert resolve_client_ip(request) == "8.8.8.8" + + def test_ipv4_mapped_hop_is_normalized(self): + request = _request(("10.0.0.50", 50000), "2.2.2.2, ::ffff:8.8.8.8") + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "8.8.8.8" + + def test_ipv6_hop_is_not_matched_against_ipv4_network(self): + request = _request(("10.0.0.50", 50000), "2001:db8::1") + with _with_config(_trusted("10.0.0.50")): + assert resolve_client_ip(request) == "2001:db8::1" + + +# ============================================================================= +# Default Config Contract +# ============================================================================= + + +class TestDefaultConfiguration: + """The shipped configuration must be the safe one.""" + + def test_trusted_proxies_default_is_empty(self): + assert AppConfig().api.trusted_proxies == [] + + def test_default_config_ignores_forwarded_for(self): + request = _request(("10.0.0.9", 50000), "203.0.113.7") + with _with_config(AppConfig()): + assert resolve_client_ip(request) == "10.0.0.9" diff --git a/tests/unit/api/test_middleware.py b/tests/unit/api/test_middleware.py index fdadad1..b1727d2 100644 --- a/tests/unit/api/test_middleware.py +++ b/tests/unit/api/test_middleware.py @@ -7,6 +7,7 @@ from __future__ import annotations import logging +from unittest.mock import patch import pytest from fastapi import FastAPI @@ -18,7 +19,7 @@ add_rate_limiting_middleware, add_request_logging_middleware, ) -from ciberwebscan.config.models import APIConfig +from ciberwebscan.config.models import APIConfig, APIRateLimitConfig, AppConfig pytestmark = pytest.mark.unit @@ -140,6 +141,16 @@ def test_log_record_has_extra_fields( assert isinstance(log_record.duration, float) +def _find_rate_limiter(app: FastAPI) -> RateLimitingMiddleware: + """Find the RateLimitingMiddleware instance in the built stack.""" + current = app.middleware_stack + while hasattr(current, "app"): + if isinstance(current, RateLimitingMiddleware): + return current + current = current.app + raise AssertionError("Could not access middleware instance") + + # ============================================================================= # RateLimitingMiddleware Tests # ============================================================================= @@ -197,68 +208,100 @@ def test_429_response_has_retry_after_header( assert 0 < retry_after <= 60 def test_different_clients_have_separate_limits(self, app: FastAPI): - """Test that different clients have independent rate limits.""" + """Test that different connection peers have independent rate limits.""" add_rate_limiting_middleware(app, requests_per_minute=2) - client = TestClient(app) + client_a = TestClient(app, client=("10.0.0.1", 50000)) + client_b = TestClient(app, client=("10.0.0.2", 50000)) - # Client 1 makes 2 requests + # Client A makes 2 requests for _ in range(2): - response = client.get("/test", headers={"X-Forwarded-For": "10.0.0.1"}) + response = client_a.get("/test") assert response.status_code == 200 - # Client 1 should be blocked - response = client.get("/test", headers={"X-Forwarded-For": "10.0.0.1"}) + # Client A should be blocked + response = client_a.get("/test") assert response.status_code == 429 - # Client 2 should still be allowed + # Client B should still be allowed for _ in range(2): - response = client.get("/test", headers={"X-Forwarded-For": "10.0.0.2"}) + response = client_b.get("/test") assert response.status_code == 200 - def test_x_forwarded_for_header_used_for_client_identification(self, app: FastAPI): - """Test that X-Forwarded-For header is used for client identification.""" + def test_rotating_forwarded_for_does_not_bypass_rate_limit(self, app: FastAPI): + """X-Forwarded-For must not mint fresh buckets (secure default).""" add_rate_limiting_middleware(app, requests_per_minute=2) - client = TestClient(app) + client = TestClient(app, client=("10.0.0.9", 50000)) - # Make requests from "client1" via X-Forwarded-For - for _ in range(2): - response = client.get("/test", headers={"X-Forwarded-For": "192.168.1.1"}) - assert response.status_code == 200 + with patch("ciberwebscan.api.client_ip.get_config", return_value=AppConfig()): + for i in range(2): + response = client.get( + "/test", headers={"X-Forwarded-For": f"192.168.1.{i + 1}"} + ) + assert response.status_code == 200 - # 3rd request from same "client" should be blocked - response = client.get("/test", headers={"X-Forwarded-For": "192.168.1.1"}) - assert response.status_code == 429 + # 3rd request with yet another forged hop: still limited + response = client.get("/test", headers={"X-Forwarded-For": "192.168.1.99"}) + assert response.status_code == 429 - # Request from different "client" should be allowed - response = client.get("/test", headers={"X-Forwarded-For": "192.168.1.2"}) - assert response.status_code == 200 + # Counter is keyed by the connection peer, not by the header + limiter = _find_rate_limiter(app) + assert set(limiter.counts) == {"10.0.0.9"} - def test_x_forwarded_for_uses_first_ip(self, app: FastAPI): - """Test that first IP in X-Forwarded-For chain is used.""" + def test_forwarded_for_resolved_right_to_left_behind_trusted_proxy( + self, app: FastAPI + ): + """Behind a trusted proxy the identity is the rightmost hop.""" add_rate_limiting_middleware(app, requests_per_minute=2) - client = TestClient(app) + client = TestClient(app, client=("10.0.0.50", 50000)) + config = AppConfig(api=APIConfig(trusted_proxies=["10.0.0.50"])) + + with patch("ciberwebscan.api.client_ip.get_config", return_value=config): + # The proxy appends the real client on the right; the left side + # is attacker-controlled and must not create new buckets. + for _ in range(2): + response = client.get( + "/test", + headers={"X-Forwarded-For": "1.1.1.1, 172.16.0.7"}, + ) + assert response.status_code == 200 - # Make requests with chained X-Forwarded-For - for _ in range(2): response = client.get( "/test", - headers={"X-Forwarded-For": "10.0.0.1, 192.168.1.1, 172.16.0.1"}, + headers={"X-Forwarded-For": "2.2.2.2, 172.16.0.7"}, ) + assert response.status_code == 429 + + # A different client behind the same proxy has its own bucket + response = client.get("/test", headers={"X-Forwarded-For": "172.16.0.8"}) assert response.status_code == 200 - # Should be blocked based on first IP (10.0.0.1) - response = client.get( - "/test", - headers={"X-Forwarded-For": "10.0.0.1, 192.168.1.1"}, - ) - assert response.status_code == 429 + limiter = _find_rate_limiter(app) + assert "172.16.0.7" in limiter.counts + assert "1.1.1.1" not in limiter.counts - # Different first IP should work - response = client.get( - "/test", - headers={"X-Forwarded-For": "10.0.0.2, 192.168.1.1"}, - ) - assert response.status_code == 200 + def test_multi_proxy_chain_resolves_left_of_trusted_hops(self, app: FastAPI): + """All trusted hops are skipped; the first untrusted hop wins. + + A forged leftmost hop (9.9.9.9) proves the chain is not resolved + from the client-controlled end. + """ + add_rate_limiting_middleware(app, requests_per_minute=2) + client = TestClient(app, client=("10.0.0.50", 50000)) + config = AppConfig(api=APIConfig(trusted_proxies=["10.0.0.50", "10.0.0.49"])) + chain = "9.9.9.9, 203.0.113.9, 10.0.0.49" + + with patch("ciberwebscan.api.client_ip.get_config", return_value=config): + for _ in range(2): + response = client.get("/test", headers={"X-Forwarded-For": chain}) + assert response.status_code == 200 + + # Same client identity (203.0.113.9, skipping the trusted hop) + # -> limited, and the proxies themselves get no bucket of their own + response = client.get("/test", headers={"X-Forwarded-For": chain}) + assert response.status_code == 429 + + limiter = _find_rate_limiter(app) + assert set(limiter.counts) == {"203.0.113.9"} def test_rate_limiting_logs_warning_on_exceeded( self, client_rate_limiting: TestClient, caplog: pytest.LogCaptureFixture @@ -341,6 +384,129 @@ def test_window_rotation_allows_new_requests(self, app: FastAPI): assert response.status_code == 200 +# ============================================================================= +# Security Regressions (client identity / X-Forwarded-For trust) +# ============================================================================= + + +class TestForwardedForSecurity: + """The forwarded chain must never become a client-controlled identity.""" + + def test_forged_forwarded_for_does_not_consume_victim_quota(self, app: FastAPI): + """Flooding with a forged victim hop must not exhaust her bucket.""" + add_rate_limiting_middleware(app, requests_per_minute=2) + attacker = TestClient(app, client=("10.0.0.9", 50000)) + victim = TestClient(app, client=("172.16.0.7", 50000)) + + with patch("ciberwebscan.api.client_ip.get_config", return_value=AppConfig()): + for _ in range(2): + assert ( + attacker.get( + "/test", headers={"X-Forwarded-For": "172.16.0.7"} + ).status_code + == 200 + ) + # Attacker exhausts his own (peer) quota... + assert ( + attacker.get( + "/test", headers={"X-Forwarded-For": "172.16.0.7"} + ).status_code + == 429 + ) + # ...and never created a bucket under the forged victim hop + limiter = _find_rate_limiter(app) + assert set(limiter.counts) == {"10.0.0.9"} + + # The victim is unaffected + assert victim.get("/test").status_code == 200 + assert limiter.counts["172.16.0.7"] == 1 + + def test_forged_victim_hop_behind_trusted_proxy_hits_real_client( + self, app: FastAPI + ): + """A proxy-appended chain credits the proxy-seen hop, not the left.""" + add_rate_limiting_middleware(app, requests_per_minute=2) + client = TestClient(app, client=("10.0.0.50", 50000)) + config = AppConfig(api=APIConfig(trusted_proxies=["10.0.0.50"])) + + with patch("ciberwebscan.api.client_ip.get_config", return_value=config): + # Attacker is 10.0.0.9 (appended by the proxy); left hop forged + for _ in range(2): + assert ( + client.get( + "/test", + headers={"X-Forwarded-For": "172.16.0.7, 10.0.0.9"}, + ).status_code + == 200 + ) + assert ( + client.get( + "/test", headers={"X-Forwarded-For": "172.16.0.7, 10.0.0.9"} + ).status_code + == 429 + ) + # Bucket under the attacker's real hop only, not the forged victim + limiter = _find_rate_limiter(app) + assert "10.0.0.9" in limiter.counts + assert "172.16.0.7" not in limiter.counts + + # The real client 172.16.0.7 still has her own bucket + assert ( + client.get( + "/test", headers={"X-Forwarded-For": "172.16.0.7"} + ).status_code + == 200 + ) + + def test_configured_trusted_proxies_do_not_trust_every_peer(self, app: FastAPI): + """A non-empty list must not make arbitrary peers trusted.""" + add_rate_limiting_middleware(app, requests_per_minute=2) + client = TestClient(app, client=("10.0.0.9", 50000)) + config = AppConfig(api=APIConfig(trusted_proxies=["10.0.0.50"])) + + with patch("ciberwebscan.api.client_ip.get_config", return_value=config): + for i in range(2): + assert ( + client.get( + "/test", headers={"X-Forwarded-For": f"192.168.1.{i + 1}"} + ).status_code + == 200 + ) + assert ( + client.get( + "/test", headers={"X-Forwarded-For": "192.168.1.50"} + ).status_code + == 429 + ) + + limiter = _find_rate_limiter(app) + assert set(limiter.counts) == {"10.0.0.9"} + + def test_create_app_rate_limits_rotating_forwarded_for(self): + """Full application (default config): rotating hops cannot bypass.""" + from ciberwebscan.api.app import create_app + + config = AppConfig( + api=APIConfig(rate_limit=APIRateLimitConfig(requests_per_minute=5)) + ) + with ( + patch("ciberwebscan.api.app.get_config", return_value=config), + patch("ciberwebscan.api.client_ip.get_config", return_value=config), + ): + app = create_app() + client = TestClient(app, client=("10.0.0.9", 50000)) + + statuses = [ + client.get( + "/health", headers={"X-Forwarded-For": f"203.0.113.{i}"} + ).status_code + for i in range(7) + ] + + assert statuses[:5] == [200] * 5 + assert statuses[5:] == [429, 429] + + # ============================================================================= # Helper Function Tests # ============================================================================= @@ -467,11 +633,12 @@ def test_rate_limiting_empty_x_forwarded_for(self, app: FastAPI): def test_endpoint(): return {"status": "ok"} - client = TestClient(app) + client = TestClient(app, client=("10.0.0.9", 50000)) - # Empty X-Forwarded-For should fall back to client IP + # Empty X-Forwarded-For should fall back to the connection peer response = client.get("/test", headers={"X-Forwarded-For": ""}) assert response.status_code == 200 + assert set(_find_rate_limiter(app).counts) == {"10.0.0.9"} def test_rate_limiting_whitespace_only_x_forwarded_for(self, app: FastAPI): """Test rate limiting with whitespace-only X-Forwarded-For header.""" @@ -481,11 +648,12 @@ def test_rate_limiting_whitespace_only_x_forwarded_for(self, app: FastAPI): def test_endpoint(): return {"status": "ok"} - client = TestClient(app) + client = TestClient(app, client=("10.0.0.9", 50000)) - # Whitespace-only X-Forwarded-For should fall back to client IP + # Whitespace-only X-Forwarded-For should fall back to the connection peer response = client.get("/test", headers={"X-Forwarded-For": " "}) assert response.status_code == 200 + assert set(_find_rate_limiter(app).counts) == {"10.0.0.9"} # ============================================================================= diff --git a/tests/unit/config/test_loader.py b/tests/unit/config/test_loader.py index afaaad4..274553e 100644 --- a/tests/unit/config/test_loader.py +++ b/tests/unit/config/test_loader.py @@ -116,6 +116,42 @@ def test_api_rate_limit_requests_per_minute(self, loader, monkeypatch): assert loader.config.api.rate_limit.requests_per_minute == 120 +# ============================================================================= +# api.trusted_proxies +# ============================================================================= + + +class TestTrustedProxiesEnv: + """CIBERWEBSCAN_API_TRUSTED_PROXIES must map onto api.trusted_proxies.""" + + def test_comma_separated_list(self, loader, monkeypatch): + _set( + monkeypatch, + "CIBERWEBSCAN_API_TRUSTED_PROXIES", + "127.0.0.1,10.0.0.0/8", + ) + assert loader.config.api.trusted_proxies == [ + "127.0.0.1/32", + "10.0.0.0/8", + ] + + def test_single_value_without_comma(self, loader, monkeypatch): + _set(monkeypatch, "CIBERWEBSCAN_API_TRUSTED_PROXIES", "::1") + assert loader.config.api.trusted_proxies == ["::1/128"] + + def test_default_without_env_is_empty(self, loader, monkeypatch): + monkeypatch.delenv("CIBERWEBSCAN_API_TRUSTED_PROXIES", raising=False) + assert loader.config.api.trusted_proxies == [] + + def test_invalid_value_is_reported_not_silently_dropped(self, loader, monkeypatch): + _set(monkeypatch, "CIBERWEBSCAN_API_TRUSTED_PROXIES", "not-an-ip") + config = loader.config + assert loader.validation_error is not None + # Pre-existing loader policy: invalid values fall back to defaults + # loudly (ERROR log + validation_error), never a partial config. + assert config.api.trusted_proxies == [] + + # ============================================================================= # Double-underscore convention # ============================================================================= diff --git a/tests/unit/config/test_models.py b/tests/unit/config/test_models.py index 9633329..80b73b3 100644 --- a/tests/unit/config/test_models.py +++ b/tests/unit/config/test_models.py @@ -203,3 +203,70 @@ def test_cors_empty_origins_with_credentials(self): cfg = APIConfig(cors_origins=[], cors_allow_credentials=True) assert cfg.cors_origins == [] assert cfg.cors_allow_credentials is True + + +# ============================================================================= +# api.trusted_proxies (client identification) +# ============================================================================= + + +class TestAPIConfigTrustedProxies: + """Proxy allow-list: default safe, explicit validation of entries.""" + + def test_default_is_empty(self): + """Empty list = X-Forwarded-For always ignored (secure default).""" + assert APIConfig().trusted_proxies == [] + + def test_accepts_ipv4_ipv6_and_cidr_entries(self): + cfg = APIConfig(trusted_proxies=["127.0.0.1", "::1", "10.0.0.0/8"]) + assert cfg.trusted_proxies == ["127.0.0.1/32", "::1/128", "10.0.0.0/8"] + + def test_accepts_comma_separated_string(self): + cfg = APIConfig(trusted_proxies="127.0.0.1, 10.0.0.0/8") + assert cfg.trusted_proxies == ["127.0.0.1/32", "10.0.0.0/8"] + + def test_empty_entries_are_dropped(self): + cfg = APIConfig(trusted_proxies=["", " ", "10.0.0.1"]) + assert cfg.trusted_proxies == ["10.0.0.1/32"] + + def test_host_bits_are_masked_with_warning(self, caplog): + import logging + + with caplog.at_level(logging.WARNING, logger="ciberwebscan.config.models"): + cfg = APIConfig(trusted_proxies=["10.0.0.1/8"]) + assert cfg.trusted_proxies == ["10.0.0.0/8"] + assert any("host bits" in r.message for r in caplog.records) + + def test_ipv4_mapped_network_converted_to_ipv4(self): + cfg = APIConfig(trusted_proxies=["::ffff:10.0.0.0/104"]) + assert cfg.trusted_proxies == ["10.0.0.0/8"] + + def test_invalid_entries_rejected(self): + for bad in ["not-an-ip", "1.2.3.4/33", "999.999.999.999", "10.0.0.0/8x"]: + with pytest.raises(ValidationError): + APIConfig(trusted_proxies=[bad]) + + def test_scalar_non_ip_value_rejected(self): + with pytest.raises(ValidationError): + APIConfig(trusted_proxies=10) + + def test_ipv4_mapped_prefix_below_96_rejected(self): + """A mapped entry wider than /96 would over-match; reject loudly.""" + with pytest.raises(ValidationError): + APIConfig(trusted_proxies=["::ffff:10.0.0.0/64"]) + + def test_ipv4_mapped_prefix_96_rejected(self): + """A mapped /96 covers every IPv4 peer; reject as catch-all.""" + with pytest.raises(ValidationError): + APIConfig(trusted_proxies=["::ffff:10.0.0.0/96"]) + + def test_catch_all_networks_rejected(self): + """0.0.0.0/0 or ::/0 would trust every peer; reject loudly.""" + for bad in ["0.0.0.0/0", "::/0"]: + with pytest.raises(ValidationError): + APIConfig(trusted_proxies=[bad]) + + def test_ipv6_zone_identifiers_rejected(self): + """Zone IDs never appear on real peers; they over-match silently.""" + with pytest.raises(ValidationError): + APIConfig(trusted_proxies=["fe80::1%eth0"])