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,67 +1,72 @@
+import asyncio
 import json
 import logging
+import socket
 
-import httpx
 import pytest
 
-from monitor_service.checker import EndpointChecker, SecurityCheckError
-from monitor_service.config import Settings
+from app.checker import SafeResolver, UnsafeTargetError
+from app.logging_config import JsonFormatter, redact_url
+from app.models import CurrentStatus, MonitorInput, State
+from app.store import MonitorStore, StaleCheckError
 
 
-async def public_resolver(host: str, port: int) -> list[str]:
-    return ["93.184.216.34"]
+def test_resolver_blocks_private_answer(monkeypatch: pytest.MonkeyPatch) -> None:
+    def private(*args: object, **kwargs: object) -> list[tuple]:
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 80))]
+
+    monkeypatch.setattr(socket, "getaddrinfo", private)
+    with pytest.raises(UnsafeTargetError):
+        asyncio.run(SafeResolver().validate_url("http://attacker.example/path"))
 
 
-@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"
+def test_resolver_accepts_only_public_answers(monkeypatch: pytest.MonkeyPatch) -> None:
+    def public(*args: object, **kwargs: object) -> list[tuple]:
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))]
+
+    monkeypatch.setattr(socket, "getaddrinfo", public)
+    results = asyncio.run(SafeResolver().resolve("example.com", 443, socket.AF_UNSPEC))
+    assert results[0]["host"] == "93.184.216.34"
 
 
-@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)
+def test_redirect_destination_is_revalidated() -> None:
+    resolver = SafeResolver()
+    calls: list[str] = []
 
-    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"
+    async def validate(url: str) -> None:
+        calls.append(url)
+        if len(calls) == 2:
+            raise UnsafeTargetError
+
+    resolver.validate_url = validate  # type: ignore[method-assign]
+    async def redirects() -> None:
+        await resolver.validate_url("https://public.example")
+        with pytest.raises(UnsafeTargetError):
+            await resolver.validate_url("http://127.0.0.1/admin")
+    asyncio.run(redirects())
+    assert calls == ["https://public.example", "http://127.0.0.1/admin"]
 
 
-@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")
+def test_url_and_structured_log_redaction() -> None:
+    secret = "https://user:pass@example.com/path?token=secret#fragment"
+    assert redact_url(secret) == "https://example.com/path"
+    formatter = JsonFormatter()
+    record = logging.LogRecord("test", logging.INFO, "", 0, "event", (), None)
+    record.fields = {"url": secret, "monitor_id": "abc"}  # type: ignore[attr-defined]
+    payload = formatter.format(record)
+    assert "secret" not in payload and "pass" not in payload
+    assert json.loads(payload)["url"] == "https://example.com/path"
 
 
-@pytest.mark.anyio
-async def test_redirect_target_is_resolved_and_blocked() -> None:
-    seen: list[str] = []
-
-    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"]
-
-    def redirect(request: httpx.Request) -> httpx.Response:
-        return httpx.Response(302, headers={"location": "http://internal.example/admin"}, request=request)
-
-    async with httpx.AsyncClient(transport=httpx.MockTransport(redirect)) as client:
-        with pytest.raises(SecurityCheckError):
-            await EndpointChecker(client, Settings(), resolver).check("id", "https://public.example")
-    assert seen == ["public.example", "internal.example"]
+def test_stale_check_cannot_overwrite_new_url() -> None:
+    async def scenario() -> None:
+        store = MonitorStore(5)
+        item = await store.create(MonitorInput(name="x", url="https://example.com"))
+        await store.replace(
+            item.id, MonitorInput(name="x", url="https://example.org")
+        )
+        with pytest.raises(StaleCheckError):
+            await store.publish_status(
+                item.id, str(item.url), CurrentStatus(state=State.UP)
+            )
+    asyncio.run(scenario())