HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""SSRF guard — pure validator tests (no network: DNS answers are injected)."""2from __future__ import annotations34import pytest56from aiatlas.sdk.fetch import MAX_REDIRECTS, BlockedDestination, validate_destination789@pytest.mark.parametrize("url", [10 "ftp://example.com/x", "file:///etc/passwd", "gopher://example.com", "javascript:alert(1)", "data:text/html,hi",11 "http://localhost/", "http://LOCALHOST:8080/", "https://foo.local/", "http://db.internal/", "http://intranet/", "http://metadata.google.internal/computeMetadata/v1/",12 "http://127.0.0.1/", "http://127.1.2.3/", "http://10.0.0.5/", "http://172.16.4.4/", "http://172.31.255.1/", "http://192.168.2.10/",13 "http://169.254.169.254/latest/meta-data/", "http://169.254.170.2/", "http://100.64.1.1/", "http://100.127.255.255/", "http://0.0.0.0/",14 "http://[::1]/", "http://[fe80::1]/", "http://[fc00::1]/", "http://[fd12:3456::1]/", "http://[::]/", "http://[::ffff:127.0.0.1]/", "http://[::ffff:10.0.0.1]/",15 "http://user:pass@example.com/",16])17def test_blocked_literals(url: str) -> None:18 with pytest.raises(BlockedDestination):19 validate_destination(url, resolved_ips=[])202122@pytest.mark.parametrize("url", ["https://huggingface.co/Qwen/Qwen3-8B", "http://arxiv.org/abs/2505.09388", "https://8.8.8.8/", "https://[2606:4700::6810:84e5]/"])23def test_public_allowed(url: str) -> None:24 validate_destination(url, resolved_ips=["104.16.0.1"])252627def test_dns_rebinding_to_private_is_blocked() -> None:28 with pytest.raises(BlockedDestination):29 validate_destination("https://evil.example.com/", resolved_ips=["10.0.0.9"])30 with pytest.raises(BlockedDestination):31 validate_destination("https://evil.example.com/", resolved_ips=["104.16.0.1", "169.254.169.254"]) # any private answer blocks32 with pytest.raises(BlockedDestination):33 validate_destination("https://evil.example.com/", resolved_ips=["fd00::1"])343536def test_unresolvable_host_is_not_ssrf() -> None:37 # resolution failures surface later as transport errors, not as blocked destinations38 validate_destination("https://does-not-exist.example.com/", resolved_ips=[])394041def test_redirect_cap_constant() -> None:42 assert MAX_REDIRECTS == 5434445async def test_fetcher_get_blocks_before_connecting() -> None:46 from aiatlas.sdk.fetch import Fetcher, FetchError4748 async with Fetcher(robots=False) as f:49 with pytest.raises(FetchError) as exc:50 await f.get("http://169.254.169.254/latest/meta-data/")51 assert "blocked destination" in str(exc.value)52 with pytest.raises(FetchError):53 await f.get("file:///etc/hosts")545556async def test_fetcher_blocks_redirect_to_private(monkeypatch: pytest.MonkeyPatch) -> None:57 """A public host redirecting to an internal address is refused on the redirect hop (uses respx-free httpx MockTransport)."""58 import httpx5960 from aiatlas.sdk import fetch as fetch_mod61 from aiatlas.sdk.fetch import Fetcher, FetchError6263 async def fake_validate(url: str) -> None:64 # public → ok, private literal → raise, exactly like the real validator without DNS65 fetch_mod.validate_destination(url, resolved_ips=["104.16.0.1"])6667 monkeypatch.setattr(fetch_mod, "validate_destination_async", fake_validate)6869 def handler(request: httpx.Request) -> httpx.Response:70 if request.url.host == "public.example.com":71 return httpx.Response(302, headers={"location": "http://169.254.169.254/latest/"})72 return httpx.Response(200, content=b"x" * 100)7374 async with Fetcher(robots=False) as f:75 f._client = httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=False)76 with pytest.raises(FetchError) as exc:77 await f.get("https://public.example.com/start", min_bytes=1)78 assert "blocked destination" in str(exc.value)798081async def test_fetcher_caps_redirects(monkeypatch: pytest.MonkeyPatch) -> None:82 import httpx8384 from aiatlas.sdk import fetch as fetch_mod85 from aiatlas.sdk.fetch import Fetcher, FetchError8687 async def fake_validate(url: str) -> None:88 fetch_mod.validate_destination(url, resolved_ips=["104.16.0.1"])8990 monkeypatch.setattr(fetch_mod, "validate_destination_async", fake_validate)91 n = {"hops": 0}9293 def handler(request: httpx.Request) -> httpx.Response:94 n["hops"] += 195 return httpx.Response(301, headers={"location": f"https://public.example.com/{n['hops']}"})9697 async with Fetcher(robots=False) as f:98 f._client = httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=False)99 with pytest.raises(FetchError) as exc:100 await f.get("https://public.example.com/start", min_bytes=1)101 assert "too many redirects" in str(exc.value)102 assert n["hops"] == MAX_REDIRECTS + 1103