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_security.py

@@ -1,43 +1,85 @@
-import io
+import asyncio
 import json
 import logging
+import socket
 
-import httpx
+import pytest
 
-from app.checker import EndpointChecker
+from app.checker import EndpointChecker, HopResponse, PinnedResolver
+from app.config import Settings
 from app.logging import JsonFormatter
-from app.models import Monitor, MonitorCreate
+from app.models import HealthState
-from app.security import redacted_url
+from app.security import TargetPolicyError, ValidatedTarget, redact_url, validate_target
 
 
-async def test_dns_resolution_blocks_private_address_before_transport() -> None:
+async def resolver_for(*addresses: str):
-    calls = 0
+    async def resolve(host: str, port: int) -> list[tuple[object, ...]]:
-
+        return [(socket.AF_INET6 if ":" in address else socket.AF_INET, 1, 6, "", (address, port)) for address in addresses]
-    async def private_resolver(_: str, __: int) -> list[str]:
+    return resolve
-        return ["10.0.0.7"]
-
-    async def transport_handler(_: httpx.Request) -> httpx.Response:
-        nonlocal calls
-        calls += 1
-        return httpx.Response(200)
-
-    monitor = Monitor(**MonitorCreate(name="private", url="https://internal.example").model_dump())
-    checker = EndpointChecker(
-        1, 2, resolver=private_resolver, transport=httpx.MockTransport(transport_handler)
-    )
-    result = await checker.check(monitor)
-    assert result.state == "blocked"
-    assert calls == 0
 
 
-def test_url_and_json_logs_redact_query_and_credentials() -> None:
+@pytest.mark.asyncio
-    assert redacted_url("https://user:pass@example.com/a?token=secret#x") == "https://example.com/a"
+@pytest.mark.parametrize("address", ["127.0.0.1", "10.1.2.3", "169.254.169.254", "::1", "fc00::1", "0.0.0.0", "224.0.0.1"])
-    stream = io.StringIO()
+async def test_ip_classification_blocks_non_public_answers(address: str) -> None:
-    handler = logging.StreamHandler(stream)
+    with pytest.raises(TargetPolicyError):
-    handler.setFormatter(JsonFormatter())
+        await validate_target("http://example.test/", await resolver_for(address))
-    record = logging.LogRecord("test", logging.INFO, "", 0, "check", (), None)
+
-    record.event_data = {"url": redacted_url("https://example.com/a?token=secret")}
+
-    handler.emit(record)
+@pytest.mark.asyncio
-    payload = json.loads(stream.getvalue())
+async def test_mixed_dns_answer_is_blocked_and_public_is_pinned() -> None:
-    assert payload["url"] == "https://example.com/a"
+    with pytest.raises(TargetPolicyError):
-    assert "secret" not in stream.getvalue()
+        await validate_target("https://example.test", await resolver_for("93.184.216.34", "127.0.0.1"))
+    target = await validate_target("https://example.test", await resolver_for("93.184.216.34"))
+    assert target.addresses == ("93.184.216.34",)
+    pinned = PinnedResolver(target.hostname, target.addresses)
+    assert (await pinned.resolve("example.test", 443))[0]["host"] == "93.184.216.34"
+    with pytest.raises(OSError):
+        await pinned.resolve("attacker.test", 443)
+
+
+class RedirectFetcher:
+    def __init__(self) -> None:
+        self.calls: list[ValidatedTarget] = []
+
+    async def fetch(self, target: ValidatedTarget, settings: Settings, remaining: float) -> HopResponse:
+        self.calls.append(target)
+        return HopResponse(302, "http://internal.test/admin") if len(self.calls) == 1 else HopResponse(200, None)
+
+
+@pytest.mark.asyncio
+async def test_redirect_hop_is_resolved_and_blocked_before_second_fetch() -> None:
+    seen: list[str] = []
+    async def resolver(host: str, port: int) -> list[tuple[object, ...]]:
+        seen.append(host)
+        address = "93.184.216.34" if host == "public.test" else "127.0.0.1"
+        return [(socket.AF_INET, 1, 6, "", (address, port))]
+    fetcher = RedirectFetcher()
+    result = await EndpointChecker(Settings(), fetcher=fetcher, resolver=resolver).check("http://public.test/start")
+    assert result.error_code == "blocked_target"
+    assert seen == ["public.test", "internal.test"]
+    assert len(fetcher.calls) == 1
+
+
+class SlowFetcher:
+    async def fetch(self, target: ValidatedTarget, settings: Settings, remaining: float) -> HopResponse:
+        await asyncio.sleep(0.1)
+        return HopResponse(200, None)
+
+
+@pytest.mark.asyncio
+async def test_total_timeout_is_mapped_without_leaking_exception() -> None:
+    checker = EndpointChecker(Settings(total_timeout_seconds=0.01, connect_timeout_seconds=0.01, read_timeout_seconds=0.01), fetcher=SlowFetcher(), resolver=await resolver_for("93.184.216.34"))
+    result = await checker.check("https://example.test/?token=secret")
+    assert result.state == HealthState.down
+    assert result.error_code == "timeout"
+    assert result.error_message == "endpoint check timed out"
+    assert result.latency_ms >= 0
+
+
+def test_query_redaction_and_json_log_output() -> None:
+    assert redact_url("https://user:pass@example.com:8443/p?q=secret#frag") == "https://example.com:8443/p"
+    record = logging.LogRecord("x", logging.INFO, "", 1, "done", (), None)
+    record.url = redact_url("https://example.com/path?api_key=hunter2")
+    payload = JsonFormatter().format(record)
+    assert json.loads(payload)["url"] == "https://example.com/path"
+    assert "hunter2" not in payload