77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from app.checker import EndpointChecker
|
|
from app.config import Settings
|
|
from app.models import MonitorState
|
|
from app.security import UnsafeTarget
|
|
|
|
|
|
async def public_resolver(host: str, port: int) -> list[str]:
|
|
return ["93.184.216.34"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_http_results_and_redirect_are_checked() -> None:
|
|
seen: list[str] = []
|
|
|
|
async def resolver(host: str, port: int) -> list[str]:
|
|
seen.append(host)
|
|
return ["93.184.216.34"]
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.host == "example.com":
|
|
return httpx.Response(302, headers={"location": "https://other.example/final"})
|
|
return httpx.Response(503)
|
|
|
|
checker = EndpointChecker(Settings(), transport=httpx.MockTransport(handler), resolver=resolver)
|
|
result = await checker.check("id", "https://example.com/start?token=secret")
|
|
assert result.state is MonitorState.DOWN
|
|
assert result.http_status == 503
|
|
assert result.latency_ms is not None
|
|
assert seen == ["example.com", "other.example"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_transport_error_becomes_error_status() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectTimeout("timed out", request=request)
|
|
|
|
checker = EndpointChecker(Settings(), transport=httpx.MockTransport(handler), resolver=public_resolver)
|
|
result = await checker.check("id", "https://example.com")
|
|
assert result.state is MonitorState.ERROR
|
|
assert "timed out" in (result.error or "")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_private_dns_is_blocked_before_transport() -> None:
|
|
called = False
|
|
|
|
async def private_resolver(host: str, port: int) -> list[str]:
|
|
return ["127.0.0.1"]
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
nonlocal called
|
|
called = True
|
|
return httpx.Response(200)
|
|
|
|
checker = EndpointChecker(Settings(), transport=httpx.MockTransport(handler), resolver=private_resolver)
|
|
with pytest.raises(UnsafeTarget):
|
|
await checker.check("id", "https://internal.example")
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_redirect_to_private_address_is_blocked() -> None:
|
|
calls = 0
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
nonlocal calls
|
|
calls += 1
|
|
return httpx.Response(302, headers={"location": "http://127.0.0.1/admin"})
|
|
|
|
checker = EndpointChecker(Settings(), transport=httpx.MockTransport(handler), resolver=public_resolver)
|
|
with pytest.raises(UnsafeTarget):
|
|
await checker.check("id", "https://example.com")
|
|
assert calls == 1
|