68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
import json
|
|
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from monitor_service.checker import EndpointChecker, SecurityCheckError
|
|
from monitor_service.config import Settings
|
|
|
|
|
|
async def public_resolver(host: str, port: int) -> list[str]:
|
|
return ["93.184.216.34"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_success_latency_and_query_redaction(caplog: pytest.LogCaptureFixture) -> None:
|
|
transport = httpx.MockTransport(lambda request: httpx.Response(200, request=request))
|
|
async with httpx.AsyncClient(transport=transport) as client:
|
|
checker = EndpointChecker(client, Settings(), public_resolver)
|
|
with caplog.at_level(logging.INFO, logger="monitor_service.checker"):
|
|
result = await checker.check("id", "https://example.com/path?token=very-secret")
|
|
assert result.status.state == "up"
|
|
assert result.status.http_status == 200
|
|
assert result.status.latency_ms is not None
|
|
assert "very-secret" not in caplog.text
|
|
event = json.loads(caplog.records[-1].message)
|
|
assert event["url"] == "https://example.com/path"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_transport_failure_maps_to_down() -> None:
|
|
def fail(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectTimeout("late", request=request)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(fail)) as client:
|
|
result = await EndpointChecker(client, Settings(), public_resolver).check("id", "https://example.com")
|
|
assert result.status.state == "down"
|
|
assert result.status.error == "ConnectTimeout"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.1", "169.254.169.254", "::1"])
|
|
async def test_dns_private_addresses_are_blocked(address: str) -> None:
|
|
async def resolver(host: str, port: int) -> list[str]:
|
|
return [address]
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200))) as client:
|
|
checker = EndpointChecker(client, Settings(), resolver)
|
|
with pytest.raises(SecurityCheckError, match="non-public"):
|
|
await checker.check("id", "http://attacker.example")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_redirect_target_is_resolved_and_blocked() -> None:
|
|
seen: list[str] = []
|
|
|
|
async def resolver(host: str, port: int) -> list[str]:
|
|
seen.append(host)
|
|
return ["127.0.0.1"] if host == "internal.example" else ["93.184.216.34"]
|
|
|
|
def redirect(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(302, headers={"location": "http://internal.example/admin"}, request=request)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(redirect)) as client:
|
|
with pytest.raises(SecurityCheckError):
|
|
await EndpointChecker(client, Settings(), resolver).check("id", "https://public.example")
|
|
assert seen == ["public.example", "internal.example"]
|