110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
import json
|
|
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.checker import EndpointChecker, UnsafeTarget
|
|
from app.config import Settings
|
|
from app.logging import StructuredJsonFormatter, redact_url
|
|
|
|
|
|
async def public_resolver(host: str, port: int) -> list[str]:
|
|
return ["93.184.216.34"]
|
|
|
|
|
|
def checker(handler, resolver=public_resolver, **settings):
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
return EndpointChecker(client, Settings(**settings), resolver), client
|
|
|
|
|
|
async def test_http_status_and_latency():
|
|
service, client = checker(lambda request: httpx.Response(503, request=request))
|
|
try:
|
|
result = await service.check("https://example.com/path")
|
|
finally:
|
|
await client.aclose()
|
|
assert result.state == "down"
|
|
assert result.status_code == 503
|
|
assert result.latency_ms is not None and result.latency_ms >= 0
|
|
|
|
|
|
async def test_timeout_is_sanitized_error_result():
|
|
def timeout(request):
|
|
raise httpx.ReadTimeout("secret low-level detail", request=request)
|
|
service, client = checker(timeout)
|
|
try:
|
|
result = await service.check("https://example.com/?token=secret")
|
|
finally:
|
|
await client.aclose()
|
|
assert result.state == "error"
|
|
assert result.error == "request timed out"
|
|
assert "secret" not in result.error
|
|
|
|
|
|
async def test_dns_blocks_private_answer_before_outbound_request():
|
|
calls = 0
|
|
def handler(request):
|
|
nonlocal calls
|
|
calls += 1
|
|
return httpx.Response(200, request=request)
|
|
async def private_resolver(host: str, port: int) -> list[str]:
|
|
return ["10.0.0.8"]
|
|
service, client = checker(handler, private_resolver)
|
|
try:
|
|
with pytest.raises(UnsafeTarget, match="non-public"):
|
|
await service.check("http://internal.invalid/")
|
|
finally:
|
|
await client.aclose()
|
|
assert calls == 0
|
|
|
|
|
|
async def test_mixed_dns_answers_are_blocked():
|
|
async def mixed(host: str, port: int) -> list[str]:
|
|
return ["93.184.216.34", "127.0.0.1"]
|
|
service, client = checker(lambda request: httpx.Response(200, request=request), mixed)
|
|
try:
|
|
with pytest.raises(UnsafeTarget):
|
|
await service.check("https://example.com")
|
|
finally:
|
|
await client.aclose()
|
|
|
|
|
|
async def test_redirect_destination_is_resolved_and_blocked():
|
|
calls = []
|
|
def handler(request):
|
|
calls.append(str(request.url))
|
|
return httpx.Response(302, headers={"Location": "http://127.0.0.1/admin"}, request=request)
|
|
async def resolver(host: str, port: int) -> list[str]:
|
|
return ["127.0.0.1"] if host == "127.0.0.1" else ["93.184.216.34"]
|
|
service, client = checker(handler, resolver)
|
|
try:
|
|
with pytest.raises(UnsafeTarget):
|
|
await service.check("https://example.com/start")
|
|
finally:
|
|
await client.aclose()
|
|
assert calls == ["https://example.com/start"]
|
|
|
|
|
|
async def test_redirect_limit_is_error():
|
|
def handler(request):
|
|
return httpx.Response(302, headers={"Location": "/again"}, request=request)
|
|
service, client = checker(handler, max_redirects=1)
|
|
try:
|
|
result = await service.check("https://example.com/start")
|
|
finally:
|
|
await client.aclose()
|
|
assert result.state == "error"
|
|
assert result.error == "redirect limit exceeded"
|
|
|
|
|
|
def test_url_redaction_and_structured_log_are_inspectable():
|
|
safe = redact_url("https://user:password@example.com/path?token=secret#private")
|
|
assert "user" not in safe and "password" not in safe and "secret" not in safe
|
|
record = logging.LogRecord("checker", logging.INFO, __file__, 1,
|
|
"done", (), None)
|
|
record.target = safe
|
|
payload = json.loads(StructuredJsonFormatter().format(record))
|
|
assert payload["target"] == "https://example.com/path?<redacted>#<redacted>"
|
|
assert payload["message"] == "done"
|