|
| 1 | +""" |
| 2 | +Behavioral tests for the 0.8.8 non-breaking security patch. |
| 3 | +
|
| 4 | +Each fix here is backward-compatible: features keep working, only the unsafe |
| 5 | +behavior is closed. (The full secure-by-default redesign is the later major.) |
| 6 | +
|
| 7 | +Covers: |
| 8 | + - SSRF blocklist gaps closed (NAT64 / 6to4 / :: / v4-mapped, not-is_global) |
| 9 | + + opaque error (no resolved IP leak) |
| 10 | + - output_path symlink/TOCTOU hardening (realpath containment + O_NOFOLLOW) |
| 11 | + with the feature kept |
| 12 | + - request-supplied LLM base_url ignored (key-exfil vector) |
| 13 | + - env:SECRET_KEY exfil gadget blocked in LLMConfig (provider keys still work) |
| 14 | + - CRLF-safe logging |
| 15 | + - webhook header sanitization |
| 16 | +""" |
| 17 | + |
| 18 | +import os |
| 19 | +import socket |
| 20 | +import sys |
| 21 | + |
| 22 | +import pytest |
| 23 | + |
| 24 | +DOCKER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| 25 | +if DOCKER_DIR not in sys.path: |
| 26 | + sys.path.insert(0, DOCKER_DIR) |
| 27 | + |
| 28 | + |
| 29 | +def _patch_dns(monkeypatch, ip): |
| 30 | + def fake(host, port=None, *a, **k): |
| 31 | + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, port or 0))] |
| 32 | + monkeypatch.setattr(socket, "getaddrinfo", fake) |
| 33 | + |
| 34 | + |
| 35 | +class TestSsrfGapsClosed: |
| 36 | + @pytest.mark.parametrize("ip", [ |
| 37 | + "169.254.169.254", # metadata |
| 38 | + "127.0.0.1", "10.0.0.5", "192.168.1.1", "100.64.0.1", |
| 39 | + "::1", "::", # v6 loopback + unspecified (was a gap) |
| 40 | + "::ffff:169.254.169.254", # v4-mapped metadata |
| 41 | + "64:ff9b::a9fe:a9fe", # NAT64 -> 169.254.169.254 (was a gap) |
| 42 | + "2002:a9fe:a9fe::1", # 6to4 embedding 169.254.169.254 (was a gap) |
| 43 | + ]) |
| 44 | + def test_internal_blocked(self, monkeypatch, ip): |
| 45 | + import utils |
| 46 | + _patch_dns(monkeypatch, ip) |
| 47 | + with pytest.raises(ValueError): |
| 48 | + utils.validate_webhook_url("http://target.example/cb") |
| 49 | + |
| 50 | + @pytest.mark.parametrize("ip", ["8.8.8.8", "1.1.1.1"]) |
| 51 | + def test_public_allowed(self, monkeypatch, ip): |
| 52 | + import utils |
| 53 | + _patch_dns(monkeypatch, ip) |
| 54 | + utils.validate_webhook_url("http://target.example/cb") # no raise |
| 55 | + |
| 56 | + def test_error_is_opaque(self, monkeypatch): |
| 57 | + import utils |
| 58 | + _patch_dns(monkeypatch, "169.254.169.254") |
| 59 | + with pytest.raises(ValueError) as e: |
| 60 | + utils.validate_webhook_url("http://target.example/") |
| 61 | + assert "169.254" not in str(e.value) # no resolved-IP leak |
| 62 | + |
| 63 | + |
| 64 | +class TestOutputPathHardening: |
| 65 | + def test_symlink_escape_rejected(self, monkeypatch, tmp_path): |
| 66 | + import utils |
| 67 | + allowed = tmp_path / "outputs" |
| 68 | + allowed.mkdir() |
| 69 | + monkeypatch.setattr(utils, "ALLOWED_OUTPUT_DIR", str(allowed)) |
| 70 | + # Plant a symlinked subdir that points outside the allowed dir. |
| 71 | + outside = tmp_path / "outside" |
| 72 | + outside.mkdir() |
| 73 | + (allowed / "evil").symlink_to(outside) |
| 74 | + from fastapi import HTTPException |
| 75 | + with pytest.raises(HTTPException): |
| 76 | + utils.validate_output_path("evil/pwned.png") # realpath escapes |
| 77 | + |
| 78 | + def test_normal_path_ok(self, monkeypatch, tmp_path): |
| 79 | + import utils |
| 80 | + allowed = tmp_path / "outputs" |
| 81 | + allowed.mkdir() |
| 82 | + monkeypatch.setattr(utils, "ALLOWED_OUTPUT_DIR", str(allowed)) |
| 83 | + p = utils.validate_output_path("sub/shot.png") |
| 84 | + assert p.startswith(str(allowed)) |
| 85 | + |
| 86 | + def test_write_refuses_symlink_final_component(self, monkeypatch, tmp_path): |
| 87 | + import utils |
| 88 | + allowed = tmp_path / "outputs" |
| 89 | + allowed.mkdir() |
| 90 | + target = allowed / "shot.png" |
| 91 | + # Pre-plant a symlink at the final path pointing at a secret. |
| 92 | + secret = tmp_path / "secret" |
| 93 | + secret.write_text("SECRET") |
| 94 | + target.symlink_to(secret) |
| 95 | + with pytest.raises(OSError): |
| 96 | + utils.write_output_file(str(target), b"data") # O_NOFOLLOW refuses |
| 97 | + |
| 98 | + |
| 99 | +class TestLlmBaseUrlIgnored: |
| 100 | + def test_request_base_url_not_honored_in_source(self): |
| 101 | + # base_url from the request must never be passed to the LLM call. |
| 102 | + with open(os.path.join(DOCKER_DIR, "api.py")) as f: |
| 103 | + src = f.read() |
| 104 | + assert "base_url or get_llm_base_url" not in src |
| 105 | + assert "base_url=get_llm_base_url(config" in src |
| 106 | + |
| 107 | + |
| 108 | +class TestEnvSecretGuard: |
| 109 | + def test_env_secret_key_blocked(self): |
| 110 | + from crawl4ai.async_configs import LLMConfig |
| 111 | + with pytest.raises(ValueError): |
| 112 | + LLMConfig(api_token="env:SECRET_KEY") |
| 113 | + |
| 114 | + @pytest.mark.parametrize("name", ["REDIS_PASSWORD", "CRAWL4AI_API_TOKEN", "MY_PRIVATE_KEY"]) |
| 115 | + def test_other_secrets_blocked(self, name): |
| 116 | + from crawl4ai.async_configs import LLMConfig |
| 117 | + with pytest.raises(ValueError): |
| 118 | + LLMConfig(api_token=f"env:{name}") |
| 119 | + |
| 120 | + def test_provider_key_still_works(self, monkeypatch): |
| 121 | + from crawl4ai.async_configs import LLMConfig |
| 122 | + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-123") |
| 123 | + cfg = LLMConfig(provider="openai/gpt-4o-mini", api_token="env:OPENAI_API_KEY") |
| 124 | + assert cfg.api_token == "sk-test-123" # normal provider key unaffected |
| 125 | + |
| 126 | + |
| 127 | +class TestCRLFSafeLogging: |
| 128 | + def test_crlf_stripped(self): |
| 129 | + import logging |
| 130 | + from utils import CRLFSafeFilter |
| 131 | + rec = logging.LogRecord("t", logging.INFO, __file__, 1, |
| 132 | + "url=http://x/\r\nINJECTED login ok", None, None) |
| 133 | + CRLFSafeFilter().filter(rec) |
| 134 | + msg = rec.getMessage() |
| 135 | + assert "\r" not in msg and "\n" not in msg and "INJECTED" in msg |
| 136 | + |
| 137 | + |
| 138 | +class TestWebhookHeaderSanitization: |
| 139 | + def test_crlf_value_rejected(self): |
| 140 | + from webhook import sanitize_webhook_headers |
| 141 | + with pytest.raises(ValueError): |
| 142 | + sanitize_webhook_headers({"X-Foo": "bar\r\nInjected: 1"}) |
| 143 | + |
| 144 | + @pytest.mark.parametrize("bad", ["Host", "Content-Length", "Authorization", "Cookie"]) |
| 145 | + def test_hop_by_hop_denied(self, bad): |
| 146 | + from webhook import sanitize_webhook_headers |
| 147 | + with pytest.raises(ValueError): |
| 148 | + sanitize_webhook_headers({bad: "x"}) |
| 149 | + |
| 150 | + def test_good_header_passes(self): |
| 151 | + from webhook import sanitize_webhook_headers |
| 152 | + assert sanitize_webhook_headers({"X-Trace-Id": "abc"}) == {"X-Trace-Id": "abc"} |
| 153 | + |
| 154 | + def test_schema_rejects_early(self): |
| 155 | + from schemas import WebhookConfig |
| 156 | + import pydantic |
| 157 | + with pytest.raises(pydantic.ValidationError): |
| 158 | + WebhookConfig(webhook_url="https://example.com/cb", webhook_headers={"Host": "evil"}) |
0 commit comments