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.

tests/test_checker.py

@@ -1,68 +1,109 @@
+import json
 import logging
-import socket
-from unittest.mock import AsyncMock, patch
 
 import httpx
 import pytest
 
-from app.checker import check_url
+from app.checker import EndpointChecker, UnsafeTarget
 from app.config import Settings
-from app.security import UnsafeDestination, validate_destination
+from app.logging import StructuredJsonFormatter, redact_url
-
-PUBLIC_DNS = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))]
 
 
-@pytest.mark.asyncio
+async def public_resolver(host: str, port: int) -> list[str]:
-async def test_success_and_redirect_targets_are_each_validated():
+    return ["93.184.216.34"]
-    transport = httpx.MockTransport(lambda request: (
-        httpx.Response(302, headers={"location": "https://target.example/final"})
-        if request.url.host == "start.example" else httpx.Response(200)
-    ))
-    validator = AsyncMock()
-    with patch("app.checker.validate_destination", validator):
-        result = await check_url("https://start.example/a?token=x", 1, Settings(), transport)
-    assert result.state == "up"
-    assert validator.await_count == 2
-    assert validator.await_args_list[1].args[0] == "https://target.example/final"
 
 
-@pytest.mark.asyncio
+def checker(handler, resolver=public_resolver, **settings):
-async def test_private_dns_answer_is_blocked():
+    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
-    loop = __import__("asyncio").get_running_loop()
+    return EndpointChecker(client, Settings(**settings), resolver), client
-    with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=[
-        (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 80))
-    ])):
-        with pytest.raises(UnsafeDestination, match="non-public"):
-            await validate_destination("http://internal.example")
 
 
-@pytest.mark.asyncio
+async def test_http_status_and_latency():
-async def test_redirect_to_private_literal_is_blocked():
+    service, client = checker(lambda request: httpx.Response(503, request=request))
-    transport = httpx.MockTransport(lambda request: httpx.Response(
+    try:
-        302, headers={"location": "http://169.254.169.254/latest"}
+        result = await service.check("https://example.com/path")
-    ))
+    finally:
-    loop = __import__("asyncio").get_running_loop()
+        await client.aclose()
-    with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=PUBLIC_DNS)):
+    assert result.state == "down"
-        with pytest.raises(UnsafeDestination):
+    assert result.status_code == 503
-            await check_url("https://example.com", 1, Settings(), transport)
+    assert result.latency_ms is not None and result.latency_ms >= 0
 
 
-@pytest.mark.asyncio
+async def test_timeout_is_sanitized_error_result():
-async def test_transport_error_maps_to_error_state():
+    def timeout(request):
-    def fail(request):
+        raise httpx.ReadTimeout("secret low-level detail", request=request)
-        raise httpx.ConnectError("no", request=request)
+    service, client = checker(timeout)
-    transport = httpx.MockTransport(fail)
+    try:
-    with patch("app.checker.validate_destination", new=AsyncMock()):
+        result = await service.check("https://example.com/?token=secret")
-        result = await check_url("https://example.com", 1, Settings(), transport)
+    finally:
+        await client.aclose()
     assert result.state == "error"
-    assert result.error == "ConnectError"
+    assert result.error == "request timed out"
+    assert "secret" not in result.error
 
 
-@pytest.mark.asyncio
+async def test_dns_blocks_private_answer_before_outbound_request():
-async def test_logs_redact_query(caplog):
+    calls = 0
-    transport = httpx.MockTransport(lambda request: httpx.Response(200))
+    def handler(request):
-    with patch("app.checker.validate_destination", new=AsyncMock()), caplog.at_level(logging.INFO):
+        nonlocal calls
-        await check_url("https://example.com/a?api_key=supersecret", 1, Settings(), transport)
+        calls += 1
-    text = caplog.text
+        return httpx.Response(200, request=request)
-    assert "supersecret" not in text
+    async def private_resolver(host: str, port: int) -> list[str]:
-    assert "REDACTED" in text
+        return ["10.0.0.8"]
+    service, client = checker(handler, private_resolver)
+    try:
+        with pytest.raises(UnsafeTarget, match="non-public"):
+            await service.check("http://internal.invalid/")
+    finally:
+        await client.aclose()
+    assert calls == 0
+
+
+async def test_mixed_dns_answers_are_blocked():
+    async def mixed(host: str, port: int) -> list[str]:
+        return ["93.184.216.34", "127.0.0.1"]
+    service, client = checker(lambda request: httpx.Response(200, request=request), mixed)
+    try:
+        with pytest.raises(UnsafeTarget):
+            await service.check("https://example.com")
+    finally:
+        await client.aclose()
+
+
+async def test_redirect_destination_is_resolved_and_blocked():
+    calls = []
+    def handler(request):
+        calls.append(str(request.url))
+        return httpx.Response(302, headers={"Location": "http://127.0.0.1/admin"}, request=request)
+    async def resolver(host: str, port: int) -> list[str]:
+        return ["127.0.0.1"] if host == "127.0.0.1" else ["93.184.216.34"]
+    service, client = checker(handler, resolver)
+    try:
+        with pytest.raises(UnsafeTarget):
+            await service.check("https://example.com/start")
+    finally:
+        await client.aclose()
+    assert calls == ["https://example.com/start"]
+
+
+async def test_redirect_limit_is_error():
+    def handler(request):
+        return httpx.Response(302, headers={"Location": "/again"}, request=request)
+    service, client = checker(handler, max_redirects=1)
+    try:
+        result = await service.check("https://example.com/start")
+    finally:
+        await client.aclose()
+    assert result.state == "error"
+    assert result.error == "redirect limit exceeded"
+
+
+def test_url_redaction_and_structured_log_are_inspectable():
+    safe = redact_url("https://user:password@example.com/path?token=secret#private")
+    assert "user" not in safe and "password" not in safe and "secret" not in safe
+    record = logging.LogRecord("checker", logging.INFO, __file__, 1,
+                               "done", (), None)
+    record.target = safe
+    payload = json.loads(StructuredJsonFormatter().format(record))
+    assert payload["target"] == "https://example.com/path?<redacted>#<redacted>"
+    assert payload["message"] == "done"