From 0bf9f75bda744796b9b516f4edc994d925287c8f Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:46:48 +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 | 148 +++++++++++++++++++++++++++---------------------- 1 file changed, 83 insertions(+), 65 deletions(-) diff --git a/app/checker.py b/app/checker.py index 6a10819..d9ca61b 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,89 +1,107 @@ import logging -from time import perf_counter +import time +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime from urllib.parse import urljoin import httpx -from app.models import CheckResult, CurrentStatus, Monitor, State, now_utc -from app.security import Resolver, UnsafeTargetError, redacted_url, resolve_addresses, validate_target +from app.config import Settings +from app.models import CurrentStatus, State +from app.security import ( + DestinationRejected, + Resolver, + redacted_url, + require_public_destination, + system_resolver, +) logger = logging.getLogger(__name__) -REDIRECTS = {301, 302, 303, 307, 308} +RedirectCodes = {301, 302, 303, 307, 308} class EndpointChecker: def __init__( self, - timeout_seconds: float, - max_redirects: int, - resolver: Resolver = resolve_addresses, + settings: Settings, + resolver: Resolver = system_resolver, transport: httpx.AsyncBaseTransport | None = None, ) -> None: - self.timeout_seconds = timeout_seconds - self.max_redirects = max_redirects + self.settings = settings self.resolver = resolver self.transport = transport - async def check(self, monitor: Monitor) -> CheckResult: - started = perf_counter() - url = str(monitor.url) - state = State.ERROR - observed: int | None = None - error: str | None = None + async def check(self, monitor_id: str, url: str) -> CurrentStatus: + started = time.monotonic() + safe_url = redacted_url(url) try: - async with httpx.AsyncClient( - transport=self.transport, - timeout=httpx.Timeout(self.timeout_seconds), - follow_redirects=False, - ) as client: - for hop in range(self.max_redirects + 1): - await validate_target(url, self.resolver) - async with client.stream( - "GET", url, headers={"user-agent": "endpoint-monitor/0.1"} - ) as response: - observed = response.status_code - if observed not in REDIRECTS: - state = ( - State.UP if observed == monitor.expected_status else State.DOWN - ) - break - location = response.headers.get("location") - if not location: - error = "redirect response omitted Location" - break - if hop == self.max_redirects: - error = "redirect limit exceeded" - break - url = urljoin(url, location) - except UnsafeTargetError: - state = State.BLOCKED - error = "target blocked by outbound request policy" - except httpx.TimeoutException: - error = "outbound request timed out" - except httpx.HTTPError: - error = "outbound HTTP request failed" - except OSError: - error = "outbound network operation failed" + result = await self._request(url, started) + except DestinationRejected: + self._log(monitor_id, safe_url, State.ERROR, started, None) + raise + except (httpx.TimeoutException, httpx.NetworkError, httpx.ProtocolError) as exc: + result = self._error(started, self._safe_error(exc)) + self._log(monitor_id, safe_url, result.state, started, result.status_code) + return result - latency = round((perf_counter() - started) * 1000, 3) - status = CurrentStatus( - state=state, - checked_at=now_utc(), - observed_status=observed, - latency_ms=latency, - error=error, + async def _request(self, initial_url: str, started: float) -> CurrentStatus: + timeout = httpx.Timeout( + self.settings.request_timeout_seconds, + connect=self.settings.connect_timeout_seconds, ) + current = initial_url + async with httpx.AsyncClient( + timeout=timeout, follow_redirects=False, transport=self.transport + ) as client: + for hop in range(self.settings.max_redirects + 1): + await require_public_destination(current, self.resolver) + async with client.stream("GET", current) as response: + if response.status_code not in RedirectCodes: + state = State.UP if 200 <= response.status_code < 400 else State.DOWN + return CurrentStatus( + state=state, + checked_at=datetime.now(UTC), + status_code=response.status_code, + latency_ms=self._elapsed(started), + ) + location = response.headers.get("location") + if not location: + return self._error(started, "redirect response omitted Location") + if hop == self.settings.max_redirects: + return self._error(started, "redirect limit exceeded") + current = urljoin(current, location) + return self._error(started, "redirect limit exceeded") # pragma: no cover + + @staticmethod + def _elapsed(started: float) -> float: + return round((time.monotonic() - started) * 1000, 3) + + def _error(self, started: float, message: str) -> CurrentStatus: + return CurrentStatus( + state=State.ERROR, + checked_at=datetime.now(UTC), + latency_ms=self._elapsed(started), + error=message[:200], + ) + + @staticmethod + def _safe_error(exc: Exception) -> str: + if isinstance(exc, httpx.TimeoutException): + return "outbound request timed out" + return f"outbound request failed: {type(exc).__name__}" + + @staticmethod + def _log( + monitor_id: str, url: str, state: State, started: float, status_code: int | None + ) -> None: logger.info( - "endpoint_check", + "endpoint check completed", extra={ - "event_data": { - "event": "endpoint_check", - "monitor_id": str(monitor.id), - "url": redacted_url(str(monitor.url)), - "state": state.value, - "observed_status": observed, - "latency_ms": latency, - } + "event": "endpoint_check", + "monitor_id": monitor_id, + "url": url, + "state": state, + "latency_ms": EndpointChecker._elapsed(started), + "status_code": status_code, }, ) - return CheckResult(monitor_id=monitor.id, **status.model_dump())