|
37 | 37 | serve, |
38 | 38 | ) |
39 | 39 |
|
40 | | -from .harness import FakeWeb, connect, raw_request |
| 40 | +from .harness import PUBLIC_IP, FakeWeb, connect, raw_request |
41 | 41 |
|
42 | 42 | PAGE = "https://example.com/page" |
43 | 43 | ROBOTS = "https://example.com/robots.txt" |
@@ -842,6 +842,34 @@ async def test_unresolvable_host_is_left_to_the_transport(web: FakeWeb) -> None: |
842 | 842 | assert web.lookups == ["only-the-proxy-knows.example"] |
843 | 843 |
|
844 | 844 |
|
| 845 | +@pytest.mark.parametrize( |
| 846 | + "url, connected_host", |
| 847 | + [ |
| 848 | + # IDNA 2003 (the codec getaddrinfo applies to a Unicode name) maps ß |
| 849 | + # and final sigma to ss and σ; IDNA 2008, which httpx connects with, |
| 850 | + # keeps them. The guard must resolve the name the connection uses. |
| 851 | + ("http://faß.example.com/", "xn--fa-hia.example.com"), |
| 852 | + ("http://xn--fa-hia.example.com/", "xn--fa-hia.example.com"), |
| 853 | + ("http://σοφος.example.com/", "xn--0xaajbq.example.com"), |
| 854 | + ], |
| 855 | +) |
| 856 | +async def test_guard_resolves_the_name_the_connection_uses( |
| 857 | + web: FakeWeb, url: str, connected_host: str |
| 858 | +) -> None: |
| 859 | + web.dns["fass.example.com"] = [PUBLIC_IP] |
| 860 | + web.dns["σοφοσ.example.com"] = [PUBLIC_IP] |
| 861 | + web.dns[connected_host] = ["127.0.0.1"] |
| 862 | + web.add(f"http://{connected_host}/", plain("internal-secret")) |
| 863 | + async with connect(ignore_robots_txt=True) as (session, _): |
| 864 | + result = await call(session, {"url": url}) |
| 865 | + assert wire(result)["isError"] is True |
| 866 | + assert f": {connected_host} resolves to 127.0.0.1, which is not a public" in ( |
| 867 | + text_of(result) |
| 868 | + ) |
| 869 | + assert web.lookups == [connected_host] |
| 870 | + assert web.requests == [] |
| 871 | + |
| 872 | + |
845 | 873 | async def test_allow_private_ips_follows_redirect_to_private_address( |
846 | 874 | web: FakeWeb, |
847 | 875 | ) -> None: |
|
0 commit comments