93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
import json
|
|
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.checker import EndpointChecker
|
|
from app.config import Settings
|
|
from app.logging import JsonFormatter
|
|
from app.main import app
|
|
from app.models import State
|
|
from app.security import DestinationRejected, redacted_url
|
|
|
|
|
|
async def public_resolver(host: str, port: int) -> list[str]:
|
|
return ["93.184.216.34"]
|
|
|
|
|
|
async def private_resolver(host: str, port: int) -> list[str]:
|
|
return ["127.0.0.1"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_success_down_and_transport_error() -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/ok":
|
|
return httpx.Response(204)
|
|
if request.url.path == "/down":
|
|
return httpx.Response(503)
|
|
raise httpx.ConnectError("secret host detail", request=request)
|
|
|
|
worker = EndpointChecker(Settings(), public_resolver, httpx.MockTransport(handler))
|
|
assert (await worker.check("1", "https://example.com/ok?token=x")).state is State.UP
|
|
assert (await worker.check("1", "https://example.com/down")).state is State.DOWN
|
|
failed = await worker.check("1", "https://example.com/fail")
|
|
assert failed.state is State.ERROR
|
|
assert failed.error == "outbound request failed: ConnectError"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_private_dns_and_private_redirect_are_blocked() -> None:
|
|
worker = EndpointChecker(Settings(), private_resolver, httpx.MockTransport(lambda r: httpx.Response(200)))
|
|
with pytest.raises(DestinationRejected):
|
|
await worker.check("1", "http://example.com")
|
|
|
|
calls = 0
|
|
|
|
async def changing_resolver(host: str, port: int) -> list[str]:
|
|
return ["127.0.0.1"] if host == "internal.test" else ["93.184.216.34"]
|
|
|
|
async def redirect(request: httpx.Request) -> httpx.Response:
|
|
nonlocal calls
|
|
calls += 1
|
|
return httpx.Response(302, headers={"location": "http://internal.test/admin"})
|
|
|
|
worker = EndpointChecker(Settings(), changing_resolver, httpx.MockTransport(redirect))
|
|
with pytest.raises(DestinationRejected):
|
|
await worker.check("1", "https://example.com/start")
|
|
assert calls == 1
|
|
|
|
|
|
def test_api_check_updates_status_and_rejection(client: TestClient) -> None:
|
|
created = client.post(
|
|
"/v1/monitors", json={"name": "site", "url": "https://example.com/ok"}
|
|
).json()
|
|
app.state.checker = EndpointChecker(
|
|
Settings(), public_resolver, httpx.MockTransport(lambda r: httpx.Response(200))
|
|
)
|
|
response = client.post(f"/v1/monitors/{created['id']}/check")
|
|
assert response.status_code == 200
|
|
assert response.json()["state"] == "up"
|
|
assert client.get(f"/v1/monitors/{created['id']}/status").json()["status_code"] == 200
|
|
|
|
app.state.checker = EndpointChecker(
|
|
Settings(), private_resolver, httpx.MockTransport(lambda r: httpx.Response(200))
|
|
)
|
|
response = client.post(f"/v1/monitors/{created['id']}/check")
|
|
assert response.status_code == 400
|
|
assert client.get(f"/v1/monitors/{created['id']}/status").json()["state"] == "error"
|
|
|
|
|
|
def test_url_and_json_log_redaction() -> None:
|
|
safe = redacted_url("https://user:pass@example.com/a?token=secret#fragment")
|
|
assert safe == "https://example.com/a?<redacted>"
|
|
record = logging.LogRecord("x", logging.INFO, "", 0, "done", (), None)
|
|
record.url = safe
|
|
payload = json.loads(JsonFormatter().format(record))
|
|
serialized = json.dumps(payload)
|
|
assert "secret" not in serialized
|
|
assert "pass" not in serialized
|
|
assert payload["url"].endswith("?<redacted>")
|