From 164cbfe8a951368f6d7a5576307d0dde1c0774c2 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:49:39 +0000 Subject: [PATCH] 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 | 102 ++++++++++++++++++++++++++++++++---------------- 1 file changed, 69 insertions(+), 33 deletions(-) diff --git a/app/security.py b/app/security.py index b8bcaba..f8a9830 100644 --- a/app/security.py +++ b/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 - -Resolver = Callable[[str, int], Awaitable[list[str]]] +from dataclasses import dataclass +from urllib.parse import SplitResult, urlsplit, urlunsplit -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 sorted({record[4][0] for record in records}) + return await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM, proto=socket.IPPROTO_TCP) -def redacted_url(url: str) -> str: - parts = urlsplit(url) - host = parts.hostname or "" - authority = f"{host}:{parts.port}" if parts.port else host - query = "" 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") +async def validate_target(url: str, resolver: Resolver = system_resolver) -> ValidatedTarget: 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) - except (OSError, UnicodeError) as exc: - raise DestinationRejected("destination DNS resolution failed") from exc + literal = ipaddress.ip_address(hostname.split("%", 1)[0]) + addresses.add(str(literal)) + 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") - try: - parsed = [ipaddress.ip_address(address) for address in addresses] - except ValueError as exc: - raise DestinationRejected("destination DNS returned an invalid address") from exc - if any(not address.is_global for address in parsed): - raise DestinationRejected("destination resolves to a non-public address") + 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)))