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/security.py

@@ -1,85 +1,29 @@
 import asyncio
 import ipaddress
 import socket
-from collections.abc import Awaitable, Callable
+from urllib.parse import urlsplit
-from dataclasses import dataclass
-from urllib.parse import SplitResult, urlsplit, urlunsplit
 
 
-class TargetPolicyError(Exception):
+class UnsafeDestination(ValueError):
     pass
 
 
-@dataclass(frozen=True)
+async def validate_destination(url: str) -> None:
-class ValidatedTarget:
+    parsed = urlsplit(url)
-    url: str
+    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
-    hostname: str
+        raise UnsafeDestination("only HTTP(S) URLs with a hostname are allowed")
-    port: int
+    if parsed.username or parsed.password:
-    addresses: tuple[str, ...]
+        raise UnsafeDestination("URL credentials are not allowed")
-
-
-Resolver = Callable[[str, int], Awaitable[list[tuple[object, ...]]]]
-
-
-def redact_url(value: str) -> str:
-    parsed = urlsplit(value)
-    host = parsed.hostname or "invalid-host"
-    if ":" in host:
-        host = f"[{host}]"
-    port = f":{parsed.port}" if parsed.port else ""
-    return urlunsplit((parsed.scheme.lower(), f"{host}{port}", parsed.path or "/", "", ""))
-
-
-def _public_address(value: str) -> bool:
-    address = ipaddress.ip_address(value)
-    return bool(
-        address.is_global
-        and not address.is_private
-        and not address.is_loopback
-        and not address.is_link_local
-        and not address.is_multicast
-        and not address.is_reserved
-        and not address.is_unspecified
-    )
-
-
-async def system_resolver(host: str, port: int) -> list[tuple[object, ...]]:
-    loop = asyncio.get_running_loop()
-    return await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP)
-
-
-async def validate_target(url: str, resolver: Resolver = system_resolver) -> ValidatedTarget:
     try:
-        parsed: SplitResult = urlsplit(url)
+        literal = ipaddress.ip_address(parsed.hostname)
-        if parsed.scheme.lower() not in {"http", "https"}:
+        addresses = [literal]
-            raise TargetPolicyError("only HTTP and HTTPS targets are allowed")
-        if parsed.username is not None or parsed.password is not None:
-            raise TargetPolicyError("target userinfo is not allowed")
-        if not parsed.hostname:
-            raise TargetPolicyError("target hostname is required")
-        hostname = parsed.hostname.rstrip(".").lower()
-        port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80)
-    except ValueError as exc:
-        raise TargetPolicyError("target URL is invalid") from exc
-
-    addresses: set[str] = set()
-    try:
-        literal = ipaddress.ip_address(hostname.split("%", 1)[0])
-        addresses.add(str(literal))
     except ValueError:
+        loop = asyncio.get_running_loop()
         try:
-            answers = await resolver(hostname, port)
+            infos = await loop.getaddrinfo(parsed.hostname, parsed.port,
-            for answer in answers:
+                                           type=socket.SOCK_STREAM)
-                sockaddr = answer[4]
+        except socket.gaierror as exc:
-                if isinstance(sockaddr, tuple) and sockaddr:
+            raise UnsafeDestination("destination DNS resolution failed") from exc
-                    addresses.add(str(ipaddress.ip_address(str(sockaddr[0]).split("%", 1)[0])))
+        addresses = list({ipaddress.ip_address(info[4][0]) for info in infos})
-        except (OSError, ValueError) as exc:
+    if not addresses or any(not address.is_global for address in addresses):
-            raise TargetPolicyError("target DNS resolution failed") from exc
+        raise UnsafeDestination("destination resolves to a non-public address")
-
-    if not addresses:
-        raise TargetPolicyError("target DNS returned no addresses")
-    if len(addresses) > 32:
-        raise TargetPolicyError("target DNS returned too many addresses")
-    if not all(_public_address(address) for address in addresses):
-        raise TargetPolicyError("target resolves to a non-public address")
-    return ValidatedTarget(url=url, hostname=hostname, port=port, addresses=tuple(sorted(addresses)))