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

@@ -2,48 +2,84 @@ import asyncio
 import ipaddress
 import socket
 from collections.abc import Awaitable, Callable
-from urllib.parse import urlsplit, urlunsplit
+from dataclasses import dataclass
-
+from urllib.parse import SplitResult, urlsplit, urlunsplit
-Resolver = Callable[[str, int], Awaitable[list[str]]]
 
 
-class DestinationRejected(ValueError):
+class TargetPolicyError(Exception):
     pass
 
 
-async def system_resolver(host: str, port: int) -> list[str]:
+@dataclass(frozen=True)
+class ValidatedTarget:
+    url: str
+    hostname: str
+    port: int
+    addresses: tuple[str, ...]
+
+
+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()
-    records = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    return await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP)
-    return sorted({record[4][0] for record in records})
 
 
-def redacted_url(url: str) -> str:
+async def validate_target(url: str, resolver: Resolver = system_resolver) -> ValidatedTarget:
-    parts = urlsplit(url)
-    host = parts.hostname or ""
-    authority = f"{host}:{parts.port}" if parts.port else host
-    query = "<redacted>" if parts.query else ""
-    return urlunsplit((parts.scheme, authority, parts.path or "/", query, ""))
-
-
-async def require_public_destination(url: str, resolver: Resolver = system_resolver) -> None:
-    parts = urlsplit(url)
-    if parts.scheme not in {"http", "https"} or not parts.hostname:
-        raise DestinationRejected("only HTTP(S) destinations are allowed")
-    if parts.username is not None or parts.password is not None:
-        raise DestinationRejected("URL user information is not allowed")
     try:
-        port = parts.port or (443 if parts.scheme == "https" else 80)
+        parsed: SplitResult = urlsplit(url)
+        if parsed.scheme.lower() not in {"http", "https"}:
+            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 DestinationRejected("invalid destination port") from exc
+        raise TargetPolicyError("target URL is invalid") from exc
+
+    addresses: set[str] = set()
     try:
-        addresses = await resolver(parts.hostname, port)
+        literal = ipaddress.ip_address(hostname.split("%", 1)[0])
-    except (OSError, UnicodeError) as exc:
+        addresses.add(str(literal))
-        raise DestinationRejected("destination DNS resolution failed") from exc
+    except ValueError:
+        try:
+            answers = await resolver(hostname, port)
+            for answer in answers:
+                sockaddr = answer[4]
+                if isinstance(sockaddr, tuple) and sockaddr:
+                    addresses.add(str(ipaddress.ip_address(str(sockaddr[0]).split("%", 1)[0])))
+        except (OSError, ValueError) as exc:
+            raise TargetPolicyError("target DNS resolution failed") from exc
+
     if not addresses:
-        raise DestinationRejected("destination DNS resolution returned no addresses")
+        raise TargetPolicyError("target DNS returned no addresses")
-    try:
+    if len(addresses) > 32:
-        parsed = [ipaddress.ip_address(address) for address in addresses]
+        raise TargetPolicyError("target DNS returned too many addresses")
-    except ValueError as exc:
+    if not all(_public_address(address) for address in addresses):
-        raise DestinationRejected("destination DNS returned an invalid address") from exc
+        raise TargetPolicyError("target resolves to a non-public address")
-    if any(not address.is_global for address in parsed):
+    return ValidatedTarget(url=url, hostname=hostname, port=port, addresses=tuple(sorted(addresses)))
-        raise DestinationRejected("destination resolves to a non-public address")