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

@@ -0,0 +1,117 @@
+import asyncio
+import ipaddress
+import json
+import logging
+import socket
+from time import perf_counter
+from urllib.parse import urljoin, urlsplit, urlunsplit
+
+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):
+    pass
+
+
+class StaticResolver(AbstractResolver):
+    def __init__(self, addresses: list[str]) -> None:
+        self.addresses = addresses
+
+    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]
+
+    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=(",", ":")))
+
+
+class EndpointChecker:
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+
+    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:
+        started = perf_counter()
+        current = raw_url
+        event("check_started", url=redact_url(raw_url))
+        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)