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.

app/checker.py

@@ -1,117 +1,125 @@
 import asyncio
 import ipaddress
-import json
-import logging
 import socket
+from dataclasses import dataclass
 from time import perf_counter
-from urllib.parse import urljoin, urlsplit, urlunsplit
+from urllib.parse import urljoin, urlsplit
 
 import aiohttp
 from aiohttp.abc import AbstractResolver
 
-from app.models import CurrentStatus, State, utcnow
-from app.settings import Settings
 
-logger = logging.getLogger("endpoint_monitor.checker")
-
-
-class SecurityError(Exception):
+class UnsafeTargetError(Exception):
     pass
 
 
-class StaticResolver(AbstractResolver):
-    def __init__(self, addresses: list[str]) -> None:
-        self.addresses = addresses
+class CheckTimeoutError(Exception):
+    def __init__(self, latency_ms: float) -> None:
+        self.latency_ms = latency_ms
 
-    async def resolve(self, host: str, port: int = 0, family: int = socket.AF_INET):
-        del family
-        return [{"hostname": host, "host": address, "port": port,
-                 "family": socket.AF_INET6 if ":" in address else socket.AF_INET,
-                 "proto": socket.IPPROTO_TCP, "flags": socket.AI_NUMERICHOST}
-                for address in self.addresses]
+
+class CheckNetworkError(Exception):
+    def __init__(self, latency_ms: float) -> None:
+        self.latency_ms = latency_ms
+
+
+class TooManyRedirectsError(Exception):
+    def __init__(self, latency_ms: float) -> None:
+        self.latency_ms = latency_ms
+
+
+@dataclass(frozen=True)
+class HttpOutcome:
+    status_code: int
+    latency_ms: float
+    final_url: str
+
+
+class SafeResolver(AbstractResolver):
+    async def resolve(
+        self, host: str, port: int = 0, family: int = socket.AF_INET
+    ) -> list[dict[str, object]]:
+        try:
+            infos = await asyncio.to_thread(
+                socket.getaddrinfo, host, port, family, socket.SOCK_STREAM
+            )
+        except socket.gaierror as exc:
+            raise OSError("DNS resolution failed") from exc
+        results: list[dict[str, object]] = []
+        seen: set[str] = set()
+        for resolved_family, _, proto, _, sockaddr in infos:
+            address = sockaddr[0]
+            ip = ipaddress.ip_address(address)
+            if not ip.is_global:
+                raise UnsafeTargetError("target resolves to a non-public address")
+            if address not in seen:
+                seen.add(address)
+                results.append({
+                    "hostname": host, "host": address, "port": port,
+                    "family": resolved_family, "proto": proto, "flags": 0,
+                })
+        if not results:
+            raise OSError("DNS returned no addresses")
+        return results
 
     async def close(self) -> None:
         return None
 
-
-def redact_url(raw: str) -> str:
-    try:
-        parts = urlsplit(raw)
-        host = parts.hostname or "invalid"
-        if ":" in host and not host.startswith("["):
-            host = f"[{host}]"
-        port = f":{parts.port}" if parts.port else ""
-        query = "REDACTED" if parts.query else ""
-        return urlunsplit((parts.scheme, host + port, parts.path, query, ""))
-    except (TypeError, ValueError):
-        return "invalid-url"
-
-
-def event(name: str, **fields: object) -> None:
-    logger.info(json.dumps({"event": name, **fields}, separators=(",", ":")))
+    async def validate_url(self, url: str) -> None:
+        parsed = urlsplit(url)
+        if parsed.scheme not in {"http", "https"} or not parsed.hostname:
+            raise UnsafeTargetError("only absolute HTTP(S) URLs are allowed")
+        if parsed.username is not None or parsed.password is not None:
+            raise UnsafeTargetError("URL userinfo is not allowed")
+        try:
+            port = parsed.port or (443 if parsed.scheme == "https" else 80)
+        except ValueError as exc:
+            raise UnsafeTargetError("invalid target port") from exc
+        await self.resolve(parsed.hostname, port, socket.AF_UNSPEC)
 
 
 class EndpointChecker:
-    def __init__(self, settings: Settings) -> None:
-        self.settings = settings
+    def __init__(self, timeout_seconds: float, max_redirects: int) -> None:
+        self.timeout_seconds = timeout_seconds
+        self.max_redirects = max_redirects
+        self.resolver = SafeResolver()
 
-    async def _resolve(self, host: str, port: int) -> list[str]:
-        loop = asyncio.get_running_loop()
-        answers = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM,
-                                         proto=socket.IPPROTO_TCP)
-        addresses = sorted({answer[4][0] for answer in answers})
-        if not addresses:
-            raise OSError("no addresses")
-        for raw in addresses:
-            address = ipaddress.ip_address(raw)
-            if address.ipv4_mapped is not None:
-                address = address.ipv4_mapped
-            if not address.is_global:
-                raise SecurityError("destination is not globally routable")
-        return addresses
-
-    async def _request_once(self, url: str, addresses: list[str]) -> tuple[int, str | None]:
-        timeout = aiohttp.ClientTimeout(total=self.settings.request_timeout_seconds,
-                                        connect=self.settings.connect_timeout_seconds)
-        connector = aiohttp.TCPConnector(resolver=StaticResolver(addresses), use_dns_cache=False)
-        async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session:
-            async with session.get(url, allow_redirects=False) as response:
-                return response.status, response.headers.get("Location")
-
-    async def check(self, raw_url: str) -> CurrentStatus:
+    async def check(self, url: str) -> HttpOutcome:
         started = perf_counter()
-        current = raw_url
-        event("check_started", url=redact_url(raw_url))
+        timeout = aiohttp.ClientTimeout(total=self.timeout_seconds)
+        connector = aiohttp.TCPConnector(
+            resolver=self.resolver, ttl_dns_cache=0, force_close=True
+        )
         try:
-            for redirects in range(self.settings.max_redirects + 1):
-                parsed = urlsplit(current)
-                if parsed.scheme not in {"http", "https"} or not parsed.hostname:
-                    raise SecurityError("unsupported redirect destination")
-                if parsed.username is not None or parsed.password is not None:
-                    raise SecurityError("URL credentials are not allowed")
-                port = parsed.port or (443 if parsed.scheme == "https" else 80)
-                addresses = await self._resolve(parsed.hostname, port)
-                code, location = await self._request_once(current, addresses)
-                if code in {301, 302, 303, 307, 308} and location is not None:
-                    if redirects >= self.settings.max_redirects:
-                        raise SecurityError("redirect limit exceeded")
-                    current = urljoin(current, location)
-                    continue
-                state = State.UP if 200 <= code < 400 else State.DOWN
-                result = CurrentStatus(state=state, checked_at=utcnow(),
-                                       latency_ms=(perf_counter() - started) * 1000,
-                                       http_status=code, final_url=redact_url(current))
-                event("check_finished", url=redact_url(raw_url), state=state,
-                      http_status=code)
-                return result
-            raise SecurityError("redirect limit exceeded")
-        except SecurityError:
-            category = "blocked_destination"
-        except asyncio.TimeoutError:
-            category = "timeout"
-        except (aiohttp.ClientError, OSError, ValueError):
-            category = "transport_error"
-        event("check_failed", url=redact_url(raw_url), error=category)
-        return CurrentStatus(state=State.ERROR, checked_at=utcnow(),
-                             latency_ms=(perf_counter() - started) * 1000,
-                             error=category)
+            async with aiohttp.ClientSession(
+                timeout=timeout, connector=connector, trust_env=False
+            ) as session:
+                current = url
+                for hop in range(self.max_redirects + 1):
+                    await self.resolver.validate_url(current)
+                    async with session.get(current, allow_redirects=False) as response:
+                        if response.status in {301, 302, 303, 307, 308}:
+                            location = response.headers.get("Location")
+                            if location is None:
+                                return HttpOutcome(
+                                    response.status, self._elapsed(started), current
+                                )
+                            if hop == self.max_redirects:
+                                raise TooManyRedirectsError(self._elapsed(started))
+                            current = urljoin(current, location)
+                            continue
+                        return HttpOutcome(
+                            response.status, self._elapsed(started), current
+                        )
+        except UnsafeTargetError:
+            raise
+        except TooManyRedirectsError:
+            raise
+        except (asyncio.TimeoutError, TimeoutError) as exc:
+            raise CheckTimeoutError(self._elapsed(started)) from exc
+        except aiohttp.ClientError as exc:
+            raise CheckNetworkError(self._elapsed(started)) from exc
+
+    @staticmethod
+    def _elapsed(started: float) -> float:
+        return round((perf_counter() - started) * 1000, 3)