diff --git a/tests/test_security.py b/tests/test_security.py new file mode 100644 index 0000000..54329b7 --- /dev/null +++ b/tests/test_security.py @@ -0,0 +1,43 @@ +import io +import json +import logging + +import httpx + +from app.checker import EndpointChecker +from app.logging import JsonFormatter +from app.models import Monitor, MonitorCreate +from app.security import redacted_url + + +async def test_dns_resolution_blocks_private_address_before_transport() -> None: + calls = 0 + + async def private_resolver(_: str, __: int) -> list[str]: + return ["10.0.0.7"] + + async def transport_handler(_: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response(200) + + monitor = Monitor(**MonitorCreate(name="private", url="https://internal.example").model_dump()) + checker = EndpointChecker( + 1, 2, resolver=private_resolver, transport=httpx.MockTransport(transport_handler) + ) + result = await checker.check(monitor) + assert result.state == "blocked" + assert calls == 0 + + +def test_url_and_json_logs_redact_query_and_credentials() -> None: + assert redacted_url("https://user:pass@example.com/a?token=secret#x") == "https://example.com/a" + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setFormatter(JsonFormatter()) + record = logging.LogRecord("test", logging.INFO, "", 0, "check", (), None) + record.event_data = {"url": redacted_url("https://example.com/a?token=secret")} + handler.emit(record) + payload = json.loads(stream.getvalue()) + assert payload["url"] == "https://example.com/a" + assert "secret" not in stream.getvalue()