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.

src/monitor_service/checker.py

@@ -0,0 +1,94 @@
+import asyncio
+import ipaddress
+import logging
+import socket
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from time import perf_counter
+from urllib.parse import urljoin, urlsplit
+
+import httpx
+
+from .config import Settings
+from .logging import log_event, redact_url
+from .models import CurrentStatus, MonitorState, utc_now
+
+Resolver = Callable[[str, int], Awaitable[list[str]]]
+logger = logging.getLogger("monitor_service.checker")
+
+
+class SecurityCheckError(ValueError):
+    pass
+
+
+@dataclass(frozen=True)
+class CheckOutcome:
+    final_url: str
+    status: CurrentStatus
+
+
+async def resolve_host(host: str, port: int) -> list[str]:
+    loop = asyncio.get_running_loop()
+    records = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    return sorted({record[4][0].split("%", 1)[0] for record in records})
+
+
+def _address_is_allowed(value: str) -> bool:
+    try:
+        return ipaddress.ip_address(value).is_global
+    except ValueError:
+        return False
+
+
+class EndpointChecker:
+    def __init__(self, client: httpx.AsyncClient, settings: Settings, resolver: Resolver = resolve_host) -> None:
+        self.client = client
+        self.settings = settings
+        self.resolver = resolver
+
+    async def validate_target(self, url: str) -> None:
+        parsed = urlsplit(url)
+        if parsed.scheme not in {"http", "https"} or not parsed.hostname:
+            raise SecurityCheckError("only absolute HTTP(S) targets are allowed")
+        if parsed.username is not None or parsed.password is not None:
+            raise SecurityCheckError("URL credentials are not allowed")
+        port = parsed.port or (443 if parsed.scheme == "https" else 80)
+        try:
+            addresses = await self.resolver(parsed.hostname, port)
+        except (OSError, UnicodeError) as exc:
+            raise SecurityCheckError("target DNS resolution failed") from exc
+        if not addresses:
+            raise SecurityCheckError("target DNS resolution returned no addresses")
+        if any(not _address_is_allowed(address) for address in addresses):
+            raise SecurityCheckError("target resolves to a non-public address")
+
+    async def check(self, monitor_id: object, url: str) -> CheckOutcome:
+        current = url
+        started = perf_counter()
+        try:
+            for redirect_count in range(self.settings.max_redirects + 1):
+                await self.validate_target(current)
+                async with self.client.stream("GET", current, follow_redirects=False) as response:
+                    if response.is_redirect and response.headers.get("location"):
+                        if redirect_count >= self.settings.max_redirects:
+                            raise httpx.TooManyRedirects("redirect limit exceeded")
+                        current = urljoin(current, response.headers["location"])
+                        continue
+                    elapsed = round((perf_counter() - started) * 1000, 2)
+                    state = MonitorState.up if 200 <= response.status_code < 400 else MonitorState.down
+                    status = CurrentStatus(state=state, checked_at=utc_now(), latency_ms=elapsed,
+                                           http_status=response.status_code)
+                    log_event(logger, "check_complete", monitor_id=monitor_id, url=current,
+                              state=state, latency_ms=elapsed, http_status=response.status_code)
+                    return CheckOutcome(redact_url(current), status)
+            raise httpx.TooManyRedirects("redirect limit exceeded")
+        except SecurityCheckError:
+            raise
+        except (httpx.HTTPError, TimeoutError) as exc:
+            elapsed = round((perf_counter() - started) * 1000, 2)
+            message = type(exc).__name__
+            status = CurrentStatus(state=MonitorState.down, checked_at=utc_now(), latency_ms=elapsed,
+                                   error=message)
+            log_event(logger, "check_failed", monitor_id=monitor_id, url=current,
+                      state=MonitorState.down, latency_ms=elapsed, error=message)
+            return CheckOutcome(redact_url(current), status)