64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.checker import EndpointChecker
|
|
from app.config import Settings
|
|
from app.main import create_app
|
|
from app.models import MonitorState
|
|
from app.store import MonitorStore
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
async def public_resolver(host: str, port: int) -> list[str]:
|
|
return ["93.184.216.34"]
|
|
|
|
|
|
async def test_check_updates_current_status() -> None:
|
|
transport = httpx.MockTransport(lambda request: httpx.Response(204))
|
|
store = MonitorStore()
|
|
checker = EndpointChecker(Settings(), resolver=public_resolver, transport=transport)
|
|
app = create_app(Settings(), store, checker)
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
|
|
item = (await client.post("/monitors", json={"name": "x", "url": "https://example.com"})).json()
|
|
checked = await client.post(f"/monitors/{item['id']}/check")
|
|
current = await client.get(f"/monitors/{item['id']}/status")
|
|
assert checked.json()["state"] == "up"
|
|
assert current.json()["http_status"] == 204
|
|
|
|
|
|
async def test_dns_and_redirect_ssrf_are_blocked() -> None:
|
|
called = False
|
|
|
|
def redirect(request: httpx.Request) -> httpx.Response:
|
|
nonlocal called
|
|
called = True
|
|
return httpx.Response(302, headers={"location": "http://127.0.0.1/admin"})
|
|
|
|
checker = EndpointChecker(Settings(), resolver=public_resolver, transport=httpx.MockTransport(redirect))
|
|
result = await checker.check("one", "https://example.com")
|
|
assert called and result.state is MonitorState.BLOCKED
|
|
|
|
async def private_resolver(host: str, port: int) -> list[str]:
|
|
return ["10.0.0.4"]
|
|
|
|
called = False
|
|
blocked = EndpointChecker(Settings(), resolver=private_resolver, transport=httpx.MockTransport(redirect))
|
|
result = await blocked.check("two", "https://internal.example")
|
|
assert not called and result.state is MonitorState.BLOCKED
|
|
|
|
|
|
async def test_transport_errors_and_log_redaction(caplog: pytest.LogCaptureFixture) -> None:
|
|
def failure(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("secret?token=visible", request=request)
|
|
|
|
checker = EndpointChecker(Settings(), resolver=public_resolver, transport=httpx.MockTransport(failure))
|
|
with caplog.at_level(logging.INFO, logger="monitor.checker"):
|
|
result = await checker.check("id", "https://example.com/path?token=visible#frag")
|
|
assert result.state is MonitorState.ERROR
|
|
text = caplog.text
|
|
assert "https://example.com/path" in text
|
|
assert "token=visible" not in text and "frag" not in text
|