chore: import upstream snapshot with attribution

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")