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/endpoint_monitor/checker.py

@@ -0,0 +1,107 @@
+import asyncio
+import ipaddress
+import logging
+import socket
+from collections.abc import Awaitable, Callable
+from datetime import UTC, datetime
+from time import perf_counter
+from urllib.parse import urljoin, urlsplit
+
+import httpx
+
+from .logging import redact_url
+from .models import CheckResult, MonitorStatus
+
+Resolver = Callable[[str, int], Awaitable[list[str]]]
+logger = logging.getLogger(__name__)
+
+
+class TargetBlockedError(ValueError):
+    pass
+
+
+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] for record in records})
+
+
+def _validate_ip(value: str) -> None:
+    try:
+        address = ipaddress.ip_address(value)
+    except ValueError as exc:
+        raise TargetBlockedError("DNS returned an invalid address") from exc
+    if not address.is_global:
+        raise TargetBlockedError("target resolves to a non-public address")
+
+
+async def validate_target(url: str, resolver: Resolver) -> None:
+    parsed = urlsplit(url)
+    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
+        raise TargetBlockedError("only absolute HTTP(S) URLs are allowed")
+    if parsed.username is not None or parsed.password is not None:
+        raise TargetBlockedError("URL user-info is not allowed")
+    port = parsed.port or (443 if parsed.scheme == "https" else 80)
+    try:
+        literal = ipaddress.ip_address(parsed.hostname)
+    except ValueError:
+        try:
+            answers = await resolver(parsed.hostname, port)
+        except (OSError, asyncio.TimeoutError) as exc:
+            raise TargetBlockedError("target DNS resolution failed") from exc
+        if not answers:
+            raise TargetBlockedError("target DNS returned no addresses")
+        for answer in answers:
+            _validate_ip(answer)
+    else:
+        _validate_ip(str(literal))
+
+
+class EndpointChecker:
+    def __init__(self, timeout: float, max_redirects: int, resolver: Resolver = resolve_host, transport: httpx.AsyncBaseTransport | None = None) -> None:
+        self.timeout = timeout
+        self.max_redirects = max_redirects
+        self.resolver = resolver
+        self.transport = transport
+
+    async def check(self, url: str, monitor_id: str) -> CheckResult:
+        started = perf_counter()
+        checked_at = datetime.now(UTC)
+        current = url
+        try:
+            async with asyncio.timeout(self.timeout):
+                async with httpx.AsyncClient(follow_redirects=False, timeout=self.timeout, transport=self.transport) as client:
+                    for hop in range(self.max_redirects + 1):
+                        await validate_target(current, self.resolver)
+                        response = await client.get(current, headers={"user-agent": "endpoint-monitor/1.0"})
+                        if response.is_redirect:
+                            location = response.headers.get("location")
+                            if not location:
+                                raise TargetBlockedError("redirect has no location")
+                            if hop >= self.max_redirects:
+                                raise TargetBlockedError("redirect limit exceeded")
+                            current = urljoin(current, location)
+                            continue
+                        status = MonitorStatus.UP if 200 <= response.status_code < 400 else MonitorStatus.DOWN
+                        result = self._result(status, checked_at, started, response.status_code, current, None)
+                        self._log(monitor_id, result)
+                        return result
+            raise RuntimeError("unreachable")
+        except TargetBlockedError:
+            raise
+        except TimeoutError:
+            result = self._result(MonitorStatus.ERROR, checked_at, started, None, current, "request timed out")
+        except httpx.HTTPError as exc:
+            result = self._result(MonitorStatus.ERROR, checked_at, started, None, current, f"transport error: {type(exc).__name__}")
+        except (OSError, ValueError) as exc:
+            result = self._result(MonitorStatus.ERROR, checked_at, started, None, current, f"check error: {type(exc).__name__}")
+        self._log(monitor_id, result)
+        return result
+
+    @staticmethod
+    def _result(status: MonitorStatus, checked_at: datetime, started: float, http_status: int | None, url: str, detail: str | None) -> CheckResult:
+        return CheckResult(status=status, checked_at=checked_at, latency_ms=round((perf_counter() - started) * 1000, 3), http_status=http_status, final_url=redact_url(url), detail=detail)
+
+    @staticmethod
+    def _log(monitor_id: str, result: CheckResult) -> None:
+        logger.info("check_completed", extra={"monitor_id": monitor_id, "status": result.status.value, "latency_ms": result.latency_ms, "url": result.final_url})