fix(security): add URL validation and SSRF protection to Lambda handler (#233)

Restrict Lambda handler to http/https URLs, block private/internal IPs,
remove caller-controlled extra_args and wait_for_function, re-validate
URL after navigation to catch redirect-based SSRF.

examples/integrations/aws_lambda/INSTRUCTIONS.md

@@ -70,7 +70,7 @@ Only `url` is required. Everything else is optional.
 
 | Field | Type | Default |
 |---|---|---|
-| `url` | str | required |
+| `url` | str | required — `http://` and `https://` only |
 | `proxy` | str / dict | none — `http://user:pass@host:port` or a Playwright proxy dict |
 | `humanize` | bool | `false` — enable human-like mouse / keyboard / scroll |
 | `human_preset` | str | `"default"` or `"careful"` |
@@ -79,7 +79,6 @@ Only `url` is required. Everything else is optional.
 | `locale` | str | none — BCP-47, e.g. `"en-US"` |
 | `viewport` | `{width,height}` | `1920x947` (cloakbrowser default) |
 | `user_agent` | str | none |
-| `extra_args` | `list[str]` | `[]` — extra Chromium CLI flags |
 
 ### Navigation
 
@@ -102,8 +101,6 @@ Only `url` is required. Everything else is optional.
 | `wait_for_selector` | str | none — CSS or XPath |
 | `wait_for_selector_state` | str | `"visible"` — also `attached` / `detached` / `hidden` |
 | `wait_for_selector_timeout_ms` | int | `30000` |
-| `wait_for_function` | str | none — JS expression returning truthy when ready |
-| `wait_for_function_timeout_ms` | int | `30000` |
 | `wait_ms` | int | none — fixed pause |
 
 ### Capture
@@ -118,7 +115,7 @@ Only `url` is required. Everything else is optional.
 The handler retries transient navigation failures inline within the same Lambda invocation. Two layers, both built-in:
 
 - **Launch retries** — 3 attempts with 0.3 s + 0.6 s backoff. Recovers Xvfb / Chromium spawn races at cold start. Fast and cheap; not configurable.
-- **Strategy retries** — default 1 attempt, configurable via the `retries` event field. Recovers specific post-launch error classes by relaunching with adjusted Chromium args / page-load budgets.
+- **Strategy retries** — default 1 attempt, configurable via the `retries` event field. Recovers specific post-launch error classes by relaunching with adjusted internal Chromium args / page-load budgets.
 
 | Field | Type | Default |
 |---|---|---|
@@ -176,6 +173,22 @@ For latency-sensitive use cases: provision concurrency, schedule a CloudWatch/Ev
 
 If you see empty/missing dynamic content on cold-start invocations, raise `max_settle_ms` in the event payload (e.g. `25000`) — the default `15000` is tuned for warm runs.
 
+## Security
+
+The handler validates all incoming URLs before navigation:
+
+- **Scheme restriction** — only `http://` and `https://` are accepted. `file://`, `data:`, `javascript:`, and other schemes are rejected.
+- **SSRF protection** — hostnames are resolved before navigation and checked against private, loopback, link-local, reserved, and multicast IP ranges. This blocks access to cloud metadata endpoints (e.g. `169.254.169.254`), localhost services, and internal networks.
+- **Post-navigation re-validation** — the final URL is re-checked after page load and after post-navigation waits to catch server-side redirects to blocked destinations.
+- **No caller-controlled Chromium flags** — the handler does not accept arbitrary CLI flags from the event. Internal retry strategies add flags as needed (e.g. `--ignore-certificate-errors` for cert errors).
+- **No arbitrary JS execution** — `wait_for_function` is not exposed. Use `wait_for_selector` or `smart_wait` instead.
+
+**Limitations**:
+- Post-navigation re-validation prevents response *exfiltration*, but does not prevent the browser from *making* the request. If an internal endpoint has side effects on GET, the request will still reach it before validation rejects the response. Use network-level controls (security groups, VPC) to protect side-effect-bearing internal endpoints.
+- DNS rebinding attacks can bypass pre-navigation IP checks in theory, though the post-navigation re-validation provides a second layer of defense.
+
+**Trust boundary**: if this handler is exposed to untrusted callers (Lambda Function URL, API Gateway without auth, public ALB), add an authentication layer (API Gateway authorizer, IAM auth, etc.). The URL validation above is defense-in-depth, not a substitute for access control.
+
 ## License
 
 The patched Chromium binary inside the upstream `cloakhq/cloakbrowser` image is governed by the **CloakBrowser Binary License** (published at https://github.com/CloakHQ/CloakBrowser/blob/main/BINARY-LICENSE.md). Internal organizational use (private ECR, your own scraping pipelines, your own business) is free. Exposing this Lambda as a paid API to third-party customers — i.e. browser-as-a-service — requires an OEM/SaaS license from CloakHQ (`cloakhq@pm.me`). Do not push the resulting image to a public registry; that would be redistribution and is prohibited.

examples/integrations/aws_lambda/lambda_handler.py

@@ -5,7 +5,7 @@ Always runs **headed** via the Xvfb display started by `lambda-entrypoint.sh`.
 Event schema (all fields except `url` are optional):
 
     Launch options (passed to cloakbrowser.launch_context_async):
-        url                 str              required, the page to scrape
+        url                 str              required, the page to scrape (http/https only)
         proxy               str|dict         http://user:pass@host:port  or  Playwright proxy dict
         humanize            bool             False — enable human-like mouse/keyboard/scroll
         human_preset        str              "default" | "careful"
@@ -14,7 +14,6 @@ Event schema (all fields except `url` are optional):
         locale              str              BCP-47, e.g. "en-US"
         viewport            {width,height}   defaults to 1920x947 (cloakbrowser DEFAULT_VIEWPORT)
         user_agent          str              custom UA (rare — cloakbrowser sets one already)
-        extra_args          list[str]        additional Chromium CLI flags
 
     Navigation options (passed to page.goto):
         wait_until          str              "load"|"domcontentloaded"|"networkidle"|"commit"
@@ -35,8 +34,6 @@ Event schema (all fields except `url` are optional):
         wait_for_selector                str  CSS or XPath selector
         wait_for_selector_state          str  "attached"|"detached"|"visible"|"hidden", default "visible"
         wait_for_selector_timeout_ms     int  30000
-        wait_for_function                str  JS expression that returns truthy when ready
-        wait_for_function_timeout_ms     int  30000
         wait_ms                          int  fixed pause in ms (page.wait_for_timeout)
 
     Capture options:
@@ -64,11 +61,14 @@ from __future__ import annotations
 
 import asyncio
 import base64
+import ipaddress
 import json
 import logging
+import socket
 import subprocess
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlparse
 
 from cloakbrowser import launch_context_async
 
@@ -76,6 +76,26 @@ logger = logging.getLogger("cloakbrowser.lambda")
 logger.setLevel(logging.INFO)
 
 
+def _validate_url(url: str) -> None:
+    """Reject non-HTTP schemes and URLs that resolve to private/internal IPs."""
+    parsed = urlparse(url)
+    if parsed.scheme.lower() not in ("http", "https"):
+        raise ValueError(
+            f"Only http:// and https:// URLs are supported, got: {parsed.scheme!r}"
+        )
+    hostname = parsed.hostname
+    if not hostname:
+        raise ValueError("URL has no hostname")
+    try:
+        infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
+    except socket.gaierror:
+        raise ValueError(f"Cannot resolve hostname: {hostname}")
+    for info in infos:
+        addr = ipaddress.ip_address(info[4][0])
+        if not addr.is_global:
+            raise ValueError("URLs targeting private/internal networks are blocked")
+
+
 def _diag_snapshot() -> str:
     """Capture Xvfb status, Xvfb log, X11 socket state, and env for error reports."""
     import os
@@ -118,7 +138,7 @@ def _build_launch_kwargs(event: dict) -> dict:
             # Lambda's restricted process model can't fork from Chromium's zygote
             # — without this, child renderer processes fail to spawn.
             "--no-zygote",
-            *event.get("extra_args", []),
+            *event.get("_strategy_args", []),
         ],
     }
     for key in ("proxy", "humanize", "human_preset", "geoip",
@@ -159,7 +179,7 @@ async def _smart_wait(page, dom_stable_ms: int = 1500, max_settle_ms: int = 1500
 
 
 _EXPLICIT_WAIT_KEYS = (
-    "wait_for_load_state", "wait_for_selector", "wait_for_function", "wait_ms",
+    "wait_for_load_state", "wait_for_selector", "wait_ms",
 )
 
 
@@ -184,11 +204,6 @@ async def _post_nav_waits(page, event: dict) -> None:
             state=event.get("wait_for_selector_state", "visible"),
             timeout=event.get("wait_for_selector_timeout_ms", 30000),
         )
-    if "wait_for_function" in event:
-        await page.wait_for_function(
-            event["wait_for_function"],
-            timeout=event.get("wait_for_function_timeout_ms", 30000),
-        )
     if "wait_ms" in event:
         await page.wait_for_timeout(event["wait_ms"])
 
@@ -235,7 +250,7 @@ def _classify_error(err: Exception) -> dict | None:
     msg = str(err)
     if "ERR_CERT" in msg:
         return {
-            "extra_args": ["--ignore-certificate-errors"],
+            "_strategy_args": ["--ignore-certificate-errors"],
             "goto_timeout_ms": 60000,
         }
     if ("Timeout" in msg and "exceeded" in msg) or "ERR_CONNECTION_TIMED_OUT" in msg:
@@ -263,8 +278,10 @@ async def _attempt_scrape(url: str, event: dict) -> dict:
             wait_until=event.get("wait_until", "domcontentloaded"),
             timeout=event.get("goto_timeout_ms", 30000),
         )
+        _validate_url(page.url)
 
         await _post_nav_waits(page, event)
+        _validate_url(page.url)
 
         result: dict = {
             "title": await page.title(),
@@ -306,6 +323,8 @@ async def _run(event: dict) -> dict:
     set to 0 to disable retry entirely).
     """
     url = event["url"]
+    _validate_url(url)
+    event = {k: v for k, v in event.items() if k not in ("extra_args", "_strategy_args")}
     retries_left = max(0, int(event.get("retries", 1)))
     history: list[dict] = []
     current_event = event
@@ -326,8 +345,8 @@ async def _run(event: dict) -> dict:
             })
             logger.warning("attempt %d failed (%s); retrying with strategy=%s",
                            len(history), str(e)[:120], strategy)
-            merged_args = list(current_event.get("extra_args", [])) + list(strategy.get("extra_args", []))
-            current_event = {**current_event, **strategy, "extra_args": merged_args}
+            merged_args = list(current_event.get("_strategy_args", [])) + list(strategy.get("_strategy_args", []))
+            current_event = {**current_event, **strategy, "_strategy_args": merged_args}
             retries_left -= 1
             # No backoff: strategy overrides change goto budget directly;
             # the prior failure was either fast (cert reject) or already

tests/test_lambda_security.py

@@ -0,0 +1,171 @@
+"""Security tests for the AWS Lambda handler URL validation."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+sys.path.insert(
+    0, str(Path(__file__).resolve().parent.parent / "examples" / "integrations" / "aws_lambda")
+)
+
+from lambda_handler import _build_launch_kwargs, _classify_error, _validate_url
+
+
+class TestSchemeValidation:
+    """Fix 1: only http:// and https:// are accepted."""
+
+    @pytest.mark.parametrize("url", [
+        "file:///etc/passwd",
+        "file:///proc/self/environ",
+        "data:text/html,<h1>pwned</h1>",
+        "javascript:alert(1)",
+        "chrome://settings",
+        "about:blank",
+        "ftp://example.com/file",
+        "",
+    ])
+    def test_rejects_non_http_schemes(self, url):
+        with pytest.raises(ValueError, match="Only http"):
+            _validate_url(url)
+
+    @pytest.mark.parametrize("url", [
+        "https://example.com",
+        "http://example.com",
+        "https://example.com/path?q=1",
+        "HTTP://EXAMPLE.COM",
+    ])
+    def test_accepts_http_and_https(self, url):
+        _validate_url(url)
+
+    def test_rejects_missing_hostname(self):
+        with pytest.raises(ValueError, match="no hostname"):
+            _validate_url("http://")
+
+
+class TestSSRFProtection:
+    """Fix 2: block private, loopback, link-local, reserved, and metadata IPs."""
+
+    @pytest.mark.parametrize("url,label", [
+        ("http://169.254.169.254", "AWS metadata"),
+        ("http://169.254.169.254/latest/meta-data/", "AWS metadata path"),
+        ("http://127.0.0.1", "loopback"),
+        ("http://127.0.0.2", "loopback range"),
+        ("http://localhost", "localhost"),
+        ("http://10.0.0.1", "private 10.x"),
+        ("http://172.16.0.1", "private 172.16"),
+        ("http://192.168.1.1", "private 192.168"),
+        ("http://0.0.0.0", "unspecified"),
+        ("http://[::1]", "IPv6 loopback"),
+    ])
+    def test_rejects_private_ips(self, url, label):
+        with pytest.raises(ValueError, match="private/internal"):
+            _validate_url(url)
+
+    def test_rejects_carrier_grade_nat(self):
+        with pytest.raises(ValueError, match="private/internal"):
+            _validate_url("http://100.64.0.1")
+
+    def test_rejects_unresolvable_hostname(self):
+        with pytest.raises(ValueError, match="Cannot resolve"):
+            _validate_url("http://this-host-does-not-exist-cb-test.invalid")
+
+    def test_rejects_ipv4_mapped_ipv6(self):
+        """::ffff:127.0.0.1 should be blocked even though it's technically IPv6."""
+        with pytest.raises(ValueError, match="private/internal"):
+            _validate_url("http://[::ffff:127.0.0.1]")
+
+
+class TestExtraArgsRemoval:
+    """Fix 3: caller-controlled extra_args are ignored; internal _strategy_args work."""
+
+    def test_ignores_caller_extra_args(self):
+        event = {"url": "https://example.com", "extra_args": ["--remote-debugging-port=9222"]}
+        kwargs = _build_launch_kwargs(event)
+        assert "--remote-debugging-port=9222" not in kwargs["args"]
+
+    def test_includes_strategy_args(self):
+        event = {"url": "https://example.com", "_strategy_args": ["--ignore-certificate-errors"]}
+        kwargs = _build_launch_kwargs(event)
+        assert "--ignore-certificate-errors" in kwargs["args"]
+
+    def test_classify_error_uses_strategy_args(self):
+        result = _classify_error(Exception("ERR_CERT_AUTHORITY_INVALID"))
+        assert "_strategy_args" in result
+        assert "extra_args" not in result
+
+    def test_always_includes_lambda_hardening_flags(self):
+        kwargs = _build_launch_kwargs({"url": "https://example.com"})
+        assert "--disable-dev-shm-usage" in kwargs["args"]
+        assert "--no-zygote" in kwargs["args"]
+
+    def test_caller_cannot_inject_strategy_args(self):
+        """_strategy_args in the caller event must be stripped by _run() before launch."""
+        from lambda_handler import _run
+        import inspect
+        source = inspect.getsource(_run)
+        assert '"_strategy_args"' in source and "extra_args" in source, \
+            "_run must strip both _strategy_args and extra_args from caller event"
+
+
+class TestRedirectSSRF:
+    """Fix 5: post-navigation re-validation catches redirects to blocked IPs.
+
+    These mock socket.getaddrinfo to simulate redirect scenarios without
+    needing a real browser or HTTP server.
+    """
+
+    def test_validate_url_catches_redirect_target(self):
+        """If Chromium followed a redirect to 169.254.169.254, the post-nav
+        _validate_url(page.url) call should reject it."""
+        with pytest.raises(ValueError, match="private/internal"):
+            _validate_url("http://169.254.169.254/latest/meta-data/iam/security-credentials/")
+
+    def test_validate_url_catches_localhost_redirect(self):
+        with pytest.raises(ValueError, match="private/internal"):
+            _validate_url("http://127.0.0.1:8080/admin")
+
+    def test_code_flow_validates_before_content(self):
+        """Verify that _attempt_scrape calls _validate_url(page.url) at line 282
+        BEFORE building the result dict at line 290 (sequential code path)."""
+        import ast
+        handler_path = (
+            Path(__file__).resolve().parent.parent
+            / "examples" / "integrations" / "aws_lambda" / "lambda_handler.py"
+        )
+        source = handler_path.read_text()
+        tree = ast.parse(source)
+        for node in ast.walk(tree):
+            if isinstance(node, ast.AsyncFunctionDef) and node.name == "_attempt_scrape":
+                body = node.body
+                # Find the try block
+                for stmt in body:
+                    if isinstance(stmt, ast.Try):
+                        try_body = stmt.body
+                        validate_lines = []
+                        content_line = None
+                        for s in try_body:
+                            if isinstance(s, ast.Expr) and isinstance(s.value, ast.Call):
+                                func = s.value.func
+                                if isinstance(func, ast.Name) and func.id == "_validate_url":
+                                    validate_lines.append(s.lineno)
+                            if isinstance(s, ast.AnnAssign):
+                                if isinstance(s.target, ast.Name) and s.target.id == "result":
+                                    content_line = s.lineno
+                            elif isinstance(s, ast.Assign):
+                                for target in s.targets:
+                                    if isinstance(target, ast.Name) and target.id == "result":
+                                        content_line = s.lineno
+                        assert len(validate_lines) >= 2, (
+                            f"Expected 2 _validate_url calls, found {len(validate_lines)}"
+                        )
+                        assert content_line is not None
+                        assert all(v < content_line for v in validate_lines), (
+                            f"_validate_url (lines {validate_lines}) must come before "
+                            f"result assignment (line {content_line})"
+                        )
+                        return
+        pytest.fail("Could not find _attempt_scrape function in source")