From dfebce19acb8ff4a3031a4f0f66ce2a6cfe80ffe Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 16:00:56 +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 | 152 +++++++++++++++++++++++++++++++------------------ 1 file changed, 96 insertions(+), 56 deletions(-) diff --git a/app/checker.py b/app/checker.py index 0632d1d..34c8eaf 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,76 +1,116 @@ +import asyncio +import ipaddress import logging +import socket import time -from datetime import UTC, datetime -from urllib.parse import urljoin +from datetime import datetime, timezone +from typing import Awaitable, Callable +from urllib.parse import urljoin, urlsplit import httpx from app.config import Settings -from app.models import MonitorState, StatusSnapshot -from app.security import Resolver, UnsafeTarget, redact_url, system_resolver, validate_target +from app.logging import redact_url +from app.models import CheckStatus -logger = logging.getLogger("monitor.checker") +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") class EndpointChecker: - def __init__( - self, settings: Settings, resolver: Resolver = system_resolver, - transport: httpx.AsyncBaseTransport | None = None, - ) -> None: + def __init__(self, client: httpx.AsyncClient, settings: Settings, + resolver: Resolver = system_resolver) -> None: + self.client = client self.settings = settings self.resolver = resolver - self.transport = transport - async def check(self, monitor_id: str, url: str) -> StatusSnapshot: - started = time.perf_counter() + async def check(self, url: str) -> CheckStatus: + started = time.monotonic() current = url try: - timeout = httpx.Timeout( - self.settings.check_timeout_seconds, - connect=self.settings.connect_timeout_seconds, - ) - async with httpx.AsyncClient( - timeout=timeout, transport=self.transport, follow_redirects=False, - ) as client: - for hop in range(self.settings.max_redirects + 1): - await validate_target(current, self.resolver) - async with client.stream("GET", current) as response: - if response.is_redirect: - location = response.headers.get("location") - if not location: - return self._result(MonitorState.ERROR, started, error="bad_redirect") - if hop == self.settings.max_redirects: - return self._result(MonitorState.ERROR, started, error="too_many_redirects") - current = urljoin(current, location) - continue - state = MonitorState.UP if 200 <= response.status_code < 400 else MonitorState.DOWN - result = self._result(state, started, response.status_code) - self._log(monitor_id, url, result) - return result - except UnsafeTarget as exc: - result = self._result(MonitorState.BLOCKED, started, error=str(exc)) + 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") + async with self.client.stream( + "GET", current, follow_redirects=False, timeout=remaining, + headers={"User-Agent": "endpoint-monitor/1.0"}, + ) as response: + body_size = 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) + 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 except httpx.TimeoutException: - result = self._result(MonitorState.ERROR, started, error="timeout") + result = self._result(started, "error", error="request timed out") except httpx.HTTPError: - result = self._result(MonitorState.ERROR, started, error="transport_error") - self._log(monitor_id, url, result) + 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, + }) return result @staticmethod - def _result( - state: MonitorState, started: float, status: int | None = None, - error: str | None = None, - ) -> StatusSnapshot: - return StatusSnapshot( - state=state, checked_at=datetime.now(UTC), - latency_ms=round((time.perf_counter() - started) * 1000, 3), - http_status=status, error=error, - ) - - @staticmethod - def _log(monitor_id: str, url: str, result: StatusSnapshot) -> None: - logger.info("check_complete", extra={ - "monitor_id": monitor_id, "target": redact_url(url), - "state": result.state, "latency_ms": result.latency_ms, - "http_status": result.http_status, "error": result.error, - }) + 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)