decomposer: generate deliverable files for Define the service contract and project architecture for the FastAPI endpoint monitoring service.; Implement the typed monitor CRUD API and concurrency-safe in-memory state according to the service design.; Implement secure on-demand endpoint checks with status updates, latency measurement, robust error handling, and redacted structured logs.; Add operational API endpoints and environment-driven runtime configuration to the monitoring service.; Create automated tests for the monitoring service.; Package the service with Docker and developer documentation.; Validate the complete project.

This commit is contained in:
2026-08-09 15:38:31 +00:00
parent 4da2460dfe
commit f85274d76f

View File

@@ -1,108 +1,67 @@
import asyncio
import json import json
import logging import logging
import socket
import httpx
import pytest import pytest
from app.checker import EndpointChecker, redact_url from monitor_service.checker import EndpointChecker, SecurityCheckError
from app.models import State from monitor_service.config import Settings
from app.settings import Settings
@pytest.fixture async def public_resolver(host: str, port: int) -> list[str]:
def checker() -> EndpointChecker:
return EndpointChecker(Settings(request_timeout_seconds=1, connect_timeout_seconds=1,
max_redirects=2))
@pytest.mark.asyncio
async def test_public_address_is_pinned_for_request(checker: EndpointChecker, monkeypatch) -> None:
async def resolve(_host: str, _port: int) -> list[str]:
return ["93.184.216.34"] return ["93.184.216.34"]
@pytest.mark.anyio
async def test_success_latency_and_query_redaction(caplog: pytest.LogCaptureFixture) -> None:
transport = httpx.MockTransport(lambda request: httpx.Response(200, request=request))
async with httpx.AsyncClient(transport=transport) as client:
checker = EndpointChecker(client, Settings(), public_resolver)
with caplog.at_level(logging.INFO, logger="monitor_service.checker"):
result = await checker.check("id", "https://example.com/path?token=very-secret")
assert result.status.state == "up"
assert result.status.http_status == 200
assert result.status.latency_ms is not None
assert "very-secret" not in caplog.text
event = json.loads(caplog.records[-1].message)
assert event["url"] == "https://example.com/path"
@pytest.mark.anyio
async def test_transport_failure_maps_to_down() -> None:
def fail(request: httpx.Request) -> httpx.Response:
raise httpx.ConnectTimeout("late", request=request)
async with httpx.AsyncClient(transport=httpx.MockTransport(fail)) as client:
result = await EndpointChecker(client, Settings(), public_resolver).check("id", "https://example.com")
assert result.status.state == "down"
assert result.status.error == "ConnectTimeout"
@pytest.mark.anyio
@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.1", "169.254.169.254", "::1"])
async def test_dns_private_addresses_are_blocked(address: str) -> None:
async def resolver(host: str, port: int) -> list[str]:
return [address]
async with httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200))) as client:
checker = EndpointChecker(client, Settings(), resolver)
with pytest.raises(SecurityCheckError, match="non-public"):
await checker.check("id", "http://attacker.example")
@pytest.mark.anyio
async def test_redirect_target_is_resolved_and_blocked() -> None:
seen: list[str] = [] seen: list[str] = []
async def request(_url: str, addresses: list[str]) -> tuple[int, None]:
seen.extend(addresses)
return 200, None
monkeypatch.setattr(checker, "_resolve", resolve)
monkeypatch.setattr(checker, "_request_once", request)
result = await checker.check("https://example.com/?token=secret")
assert result.state == State.UP
assert seen == ["93.184.216.34"]
assert "secret" not in (result.final_url or "")
async def resolver(host: str, port: int) -> list[str]:
seen.append(host)
return ["127.0.0.1"] if host == "internal.example" else ["93.184.216.34"]
@pytest.mark.asyncio def redirect(request: httpx.Request) -> httpx.Response:
@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.1", "169.254.1.1", "::1", "192.0.2.1"]) return httpx.Response(302, headers={"location": "http://internal.example/admin"}, request=request)
async def test_dns_rejects_non_global_answers(checker: EndpointChecker, monkeypatch, address: str) -> None:
loop = asyncio.get_running_loop()
def answers(*_args, **_kwargs):
family = socket.AF_INET6 if ":" in address else socket.AF_INET
return [(family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (address, 443))]
monkeypatch.setattr(loop, "getaddrinfo", answers)
with pytest.raises(Exception, match="globally routable"):
await checker._resolve("attacker.invalid", 443)
async with httpx.AsyncClient(transport=httpx.MockTransport(redirect)) as client:
@pytest.mark.asyncio with pytest.raises(SecurityCheckError):
async def test_mixed_dns_answer_is_rejected(checker: EndpointChecker, monkeypatch) -> None: await EndpointChecker(client, Settings(), resolver).check("id", "https://public.example")
loop = asyncio.get_running_loop() assert seen == ["public.example", "internal.example"]
def answers(*_args, **_kwargs):
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 80))
for ip in ("93.184.216.34", "127.0.0.1")]
monkeypatch.setattr(loop, "getaddrinfo", answers)
with pytest.raises(Exception):
await checker._resolve("rebinding.invalid", 80)
@pytest.mark.asyncio
async def test_redirect_target_is_resolved_and_blocked(checker: EndpointChecker, monkeypatch) -> None:
calls = 0
async def resolve(host: str, _port: int) -> list[str]:
if host == "localhost":
from app.checker import SecurityError
raise SecurityError("destination is not globally routable")
return ["93.184.216.34"]
async def request(_url: str, _addresses: list[str]) -> tuple[int, str]:
nonlocal calls
calls += 1
return 302, "http://localhost/admin?key=secret"
monkeypatch.setattr(checker, "_resolve", resolve)
monkeypatch.setattr(checker, "_request_once", request)
result = await checker.check("https://example.com")
assert result.state == State.ERROR
assert result.error == "blocked_destination"
assert calls == 1
@pytest.mark.asyncio
async def test_timeout_mapping(checker: EndpointChecker, monkeypatch) -> None:
async def resolve(_host: str, _port: int) -> list[str]:
return ["93.184.216.34"]
async def request(_url: str, _addresses: list[str]):
raise asyncio.TimeoutError
monkeypatch.setattr(checker, "_resolve", resolve)
monkeypatch.setattr(checker, "_request_once", request)
result = await checker.check("https://example.com")
assert result.state == State.ERROR and result.error == "timeout"
def test_redaction_removes_credentials_query_and_fragment() -> None:
value = redact_url("https://user:secret@example.com/path?token=abc&x=1#frag")
assert value == "https://example.com/path?REDACTED"
assert all(secret not in value for secret in ("user", "secret", "abc", "frag"))
@pytest.mark.asyncio
async def test_structured_logs_do_not_leak_query(checker: EndpointChecker, monkeypatch, caplog) -> None:
async def resolve(_host: str, _port: int) -> list[str]:
return ["93.184.216.34"]
async def request(_url: str, _addresses: list[str]) -> tuple[int, None]:
return 500, None
monkeypatch.setattr(checker, "_resolve", resolve)
monkeypatch.setattr(checker, "_request_once", request)
with caplog.at_level(logging.INFO, logger="endpoint_monitor.checker"):
result = await checker.check("https://example.com/a?api_key=supersecret")
assert result.state == State.DOWN
assert "supersecret" not in caplog.text
assert all("event" in json.loads(record.message) for record in caplog.records)