From 7a2e91aa806d1c856118b65f22e83a4d85e6413f Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 16:03:47 +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/checker.py | 143 +++++++++++++++++-------------------------------- 1 file changed, 49 insertions(+), 94 deletions(-) diff --git a/app/checker.py b/app/checker.py index 34c8eaf..87ff0b8 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,116 +1,71 @@ -import asyncio -import ipaddress import logging -import socket import time -from datetime import datetime, timezone -from typing import Awaitable, Callable -from urllib.parse import urljoin, urlsplit +from datetime import datetime +from urllib.parse import urljoin import httpx -from app.config import Settings -from app.logging import redact_url -from app.models import CheckStatus +from .logging_config import redact_url +from .models import CheckResult, MonitorStatus, utcnow +from .security import Resolver, UnsafeTarget, system_resolver, validate_target -logger = logging.getLogger(__name__) -Resolver = Callable[[str, int], Awaitable[list[str]]] - - -class UnsafeTarget(ValueError): - pass - - -async def system_resolver(host: str, port: int) -> list[str]: - loop = asyncio.get_running_loop() - answers = await loop.run_in_executor( - None, lambda: socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) - ) - return sorted({answer[4][0] for answer in answers}) - - -async def validate_target(url: str, resolver: Resolver) -> None: - parsed = urlsplit(url) - if parsed.scheme not in {"http", "https"} or not parsed.hostname: - raise UnsafeTarget("only absolute HTTP(S) targets are allowed") - if parsed.username is not None or parsed.password is not None: - raise UnsafeTarget("target user information is not allowed") - try: - port = parsed.port or (443 if parsed.scheme == "https" else 80) - except ValueError as exc: - raise UnsafeTarget("invalid target port") from exc - try: - addresses = await resolver(parsed.hostname, port) - except (OSError, socket.gaierror) as exc: - raise UnsafeTarget("target DNS resolution failed") from exc - if not addresses: - raise UnsafeTarget("target DNS resolution returned no addresses") - try: - unsafe = [address for address in addresses if not ipaddress.ip_address(address).is_global] - except ValueError as exc: - raise UnsafeTarget("target DNS returned an invalid address") from exc - if unsafe: - raise UnsafeTarget("target resolves to a non-public address") +logger = logging.getLogger("monitor.checker") class EndpointChecker: - def __init__(self, client: httpx.AsyncClient, settings: Settings, - resolver: Resolver = system_resolver) -> None: + def __init__( + self, + client: httpx.AsyncClient, + timeout: float, + max_redirects: int, + max_response_bytes: int, + resolver: Resolver = system_resolver, + ) -> None: self.client = client - self.settings = settings + self.timeout = timeout + self.max_redirects = max_redirects + self.max_response_bytes = max_response_bytes self.resolver = resolver - async def check(self, url: str) -> CheckStatus: - started = time.monotonic() - current = url + async def check(self, monitor_id: str, initial_url: str) -> CheckResult: + started = time.perf_counter() + checked_at: datetime = utcnow() + current_url = initial_url try: - for redirect_count in range(self.settings.max_redirects + 1): - await validate_target(current, self.resolver) - remaining = self.settings.check_timeout_seconds - (time.monotonic() - started) - if remaining <= 0: - raise httpx.TimeoutException("total check timeout exceeded") + for redirect_count in range(self.max_redirects + 1): + await validate_target(current_url, self.resolver) async with self.client.stream( - "GET", current, follow_redirects=False, timeout=remaining, - headers={"User-Agent": "endpoint-monitor/1.0"}, + "GET", current_url, follow_redirects=False, timeout=self.timeout, + headers={"User-Agent": "endpoint-monitor/1"}, ) as response: - body_size = 0 + consumed = 0 async for chunk in response.aiter_bytes(): - body_size += len(chunk) - if body_size > self.settings.max_response_bytes: - raise httpx.TooManyRedirects("response exceeded configured limit") - if response.is_redirect: - location = response.headers.get("location") - if not location: - return self._result(started, "error", error="redirect missing Location") - if redirect_count >= self.settings.max_redirects: - return self._result(started, "error", error="redirect limit exceeded") - current = urljoin(current, location) + consumed += len(chunk) + if consumed >= self.max_response_bytes: + break + if response.is_redirect and response.headers.get("location"): + if redirect_count == self.max_redirects: + return self._result(started, checked_at, MonitorStatus.error, error="too_many_redirects") + current_url = urljoin(str(response.url), response.headers["location"]) continue - state = "up" if 200 <= response.status_code < 400 else "down" - result = self._result(started, state, status_code=response.status_code) - logger.info("endpoint_check_complete", extra={ - "target": redact_url(current), "state": state, - "status_code": response.status_code, - "latency_ms": result.latency_ms, - }) - return result - except UnsafeTarget: - logger.warning("endpoint_check_blocked", extra={"target": redact_url(current)}) - raise + status = MonitorStatus.up if 200 <= response.status_code < 400 else MonitorStatus.down + return self._result(started, checked_at, status, response.status_code) + except UnsafeTarget as exc: + result = self._result(started, checked_at, MonitorStatus.error, error=str(exc)) except httpx.TimeoutException: - result = self._result(started, "error", error="request timed out") + result = self._result(started, checked_at, MonitorStatus.error, error="timeout") except httpx.HTTPError: - result = self._result(started, "error", error="HTTP transport error") - logger.info("endpoint_check_failed", extra={ - "target": redact_url(current), "state": result.state, - "error": result.error, "latency_ms": result.latency_ms, - }) + result = self._result(started, checked_at, MonitorStatus.error, error="network_error") + except Exception: + logger.exception("check_unexpected_error", extra={"monitor_id": monitor_id, "url": redact_url(initial_url)}) + result = self._result(started, checked_at, MonitorStatus.error, error="internal_error") + self._log(monitor_id, initial_url, result) return result @staticmethod - def _result(started: float, state: str, status_code: int | None = None, - error: str | None = None) -> CheckStatus: - return CheckStatus(state=state, checked_at=datetime.now(timezone.utc), - status_code=status_code, - latency_ms=round((time.monotonic() - started) * 1000, 3), - error=error) + def _result(started: float, checked_at: datetime, status: MonitorStatus, http_status: int | None = None, error: str | None = None) -> CheckResult: + return CheckResult(status=status, checked_at=checked_at, latency_ms=max(0.0, (time.perf_counter() - started) * 1000), http_status=http_status, error=error) + + @staticmethod + def _log(monitor_id: str, url: str, result: CheckResult) -> None: + logger.info("check_completed", extra={"monitor_id": monitor_id, "url": url, **result.model_dump()})