44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
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()
|