53 lines
2.1 KiB
Python
53 lines
2.1 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from app.checker import EndpointChecker
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
async def test_blocks_private_dns_before_http_call():
|
|
called = False
|
|
def handler(request):
|
|
nonlocal called
|
|
called = True
|
|
return httpx.Response(200, request=request)
|
|
async def private(host, port):
|
|
return ["127.0.0.1"]
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
result = await EndpointChecker(client, 1, 2, 100, private).check("id", "http://public.test")
|
|
assert result.status == "error"
|
|
assert result.error == "non_public_target"
|
|
assert not called
|
|
|
|
|
|
async def test_blocks_redirect_to_private_target():
|
|
calls = []
|
|
def handler(request):
|
|
calls.append(str(request.url))
|
|
return httpx.Response(302, headers={"location": "http://internal.test/admin"}, request=request)
|
|
async def resolver(host, port):
|
|
return ["10.0.0.1"] if host == "internal.test" else ["93.184.216.34"]
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
result = await EndpointChecker(client, 1, 2, 100, resolver).check("id", "https://example.com")
|
|
assert result.error == "non_public_target"
|
|
assert calls == ["https://example.com"]
|
|
|
|
|
|
async def test_rejects_if_any_dns_answer_is_private():
|
|
async def mixed(host, port):
|
|
return ["93.184.216.34", "169.254.169.254"]
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200))) as client:
|
|
result = await EndpointChecker(client, 1, 1, 100, mixed).check("id", "http://example.com")
|
|
assert result.error == "non_public_target"
|
|
|
|
|
|
async def test_redirect_limit():
|
|
def handler(request):
|
|
return httpx.Response(302, headers={"location": "/again"}, request=request)
|
|
async def public(host, port):
|
|
return ["93.184.216.34"]
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
result = await EndpointChecker(client, 1, 1, 100, public).check("id", "https://example.com")
|
|
assert result.error == "too_many_redirects"
|