booleanbeyond/jobfetch

1"""Egress guard tests. These are the ones that must never regress."""2 3from __future__ import annotations4 5import dataclasses6 7import pytest8 9from app import net as net_mod10from app import security as sec11from app.config import settings as base_settings12 13STRICT = dataclasses.replace(base_settings, allow_private_hosts=False, pin_dns=False)14 15 16@pytest.fixture(autouse=True)17def strict(monkeypatch):18    monkeypatch.setattr(sec, "settings", STRICT)19    monkeypatch.setattr(net_mod, "settings", STRICT)20 21 22@pytest.mark.asyncio23@pytest.mark.parametrize(24    "url",25    [26        "http://127.0.0.1/",27        "http://localhost/admin",28        "http://169.254.169.254/latest/meta-data/",29        "http://metadata.google.internal/computeMetadata/v1/",30        "http://[::1]:80/",31        "http://0.0.0.0/",32        "http://10.0.0.5/",33        "http://192.168.1.1/",34        "http://172.16.0.1/",35        "http://2130706433/",          # decimal 127.0.0.136        "http://0x7f.0x0.0x0.0x1/",    # hex 127.0.0.137        "http://127.1/",               # short form38    ],39)40async def test_private_and_obfuscated_targets_rejected(url):41    with pytest.raises(sec.UnsafeURLError):42        await sec.resolve_and_validate(url)43 44 45@pytest.mark.asyncio46@pytest.mark.parametrize(47    "url",48    [49        "file:///etc/passwd",50        "gopher://evil/",51        "ftp://files.example.com/x",52        "javascript:alert(1)",53        "data:text/html,<h1>x</h1>",54        "http://user:pass@example.com/",55        "http://example.com:22/",56        "",57        "   ",58    ],59)60async def test_bad_schemes_ports_and_credentials_rejected(url):61    with pytest.raises(sec.UnsafeURLError):62        await sec.resolve_and_validate(url)63 64 65def test_normalise_adds_scheme_and_drops_fragment():66    assert sec.normalise_url("example.com/careers#top") == "https://example.com/careers"67    assert sec.normalise_url("  https://a.com  ") == "https://a.com/"68 69 70def test_control_characters_rejected():71    with pytest.raises(sec.UnsafeURLError):72        sec.normalise_url("https://example.com/\nHost: evil")73 74 75@pytest.mark.asyncio76async def test_redirect_hops_are_revalidated(monkeypatch):77    """A public URL that 302s to link-local must be refused at the second hop."""78    budget = net_mod.Budget.start(10)79    client = net_mod.SafeClient(budget)80 81    hops: list[str] = []82 83    async def fake_raw(method, target, **kw):84        hops.append(target.url)85        return net_mod.Response(86            url=target.url,87            requested_url=target.url,88            status=302,89            headers={"location": "http://169.254.169.254/latest/meta-data/"},90            content=b"",91            text="",92            elapsed_ms=1,93        )94 95    async def fake_resolve(url):96        # Delegate to the real guard so the second hop is genuinely checked.97        if "169.254" in url:98            return await sec.resolve_and_validate(url)99        return sec.ValidatedTarget(url, "https", "example.com", 443, ("93.184.216.34",))100 101    monkeypatch.setattr(client, "_raw_request", fake_raw)102    monkeypatch.setattr(net_mod, "resolve_and_validate", fake_resolve)103 104    try:105        with pytest.raises(sec.UnsafeURLError):106            await client.get("https://example.com/start", check_robots=False)107        assert hops == ["https://example.com/start"]108    finally:109        await client.aclose()110 111 112@pytest.mark.asyncio113async def test_budget_stops_runaway_fetching():114    budget = net_mod.Budget.start(10)115    budget.max_fetches = 2116    budget.fetches = 2117    with pytest.raises(net_mod.BudgetExceeded):118        budget.check()119    assert budget.soft_ok() is False120    assert budget.truncated is True121