diff --git a/app/checker.py b/app/checker.py index d9ca61b..1fefa0d 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,107 +1,132 @@ +import asyncio import logging +import socket import time -from collections.abc import Awaitable, Callable -from datetime import UTC, datetime +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from typing import Protocol from urllib.parse import urljoin -import httpx +import aiohttp +from aiohttp.abc import AbstractResolver from app.config import Settings -from app.models import CurrentStatus, State -from app.security import ( - DestinationRejected, - Resolver, - redacted_url, - require_public_destination, - system_resolver, -) +from app.models import CurrentStatus, HealthState, utc_now +from app.security import Resolver, TargetPolicyError, ValidatedTarget, redact_url, system_resolver, validate_target -logger = logging.getLogger(__name__) -RedirectCodes = {301, 302, 303, 307, 308} +logger = logging.getLogger("monitor.checker") + + +@dataclass(frozen=True) +class HopResponse: + status: int + location: str | None + + +class Fetcher(Protocol): + async def fetch(self, target: ValidatedTarget, settings: Settings, remaining: float) -> HopResponse: ... + + +class PinnedResolver(AbstractResolver): + def __init__(self, hostname: str, addresses: Sequence[str]) -> None: + self.hostname = hostname + self.addresses = addresses + + async def resolve(self, host: str, port: int = 0, family: int = socket.AF_UNSPEC) -> list[dict[str, object]]: + if host.rstrip(".").lower() != self.hostname: + raise OSError("unpinned hostname requested") + result: list[dict[str, object]] = [] + for address in self.addresses: + address_family = socket.AF_INET6 if ":" in address else socket.AF_INET + if family in (socket.AF_UNSPEC, address_family): + result.append({"hostname": host, "host": address, "port": port, "family": address_family, "proto": 0, "flags": 0}) + return result + + async def close(self) -> None: + return None + + +class AioHttpFetcher: + async def fetch(self, target: ValidatedTarget, settings: Settings, remaining: float) -> HopResponse: + resolver = PinnedResolver(target.hostname, target.addresses) + connector = aiohttp.TCPConnector(resolver=resolver, use_dns_cache=True, force_close=True) + timeout = aiohttp.ClientTimeout( + total=remaining, + connect=min(settings.connect_timeout_seconds, remaining), + sock_read=min(settings.read_timeout_seconds, remaining), + ) + headers = {"User-Agent": settings.user_agent, "Accept": "*/*"} + async with aiohttp.ClientSession(connector=connector, timeout=timeout, trust_env=False) as session: + async with session.get(target.url, headers=headers, allow_redirects=False) as response: + await response.content.read(1) + return HopResponse(response.status, response.headers.get("Location")) + + +_ERROR_MESSAGES = { + "blocked_target": "target rejected by network policy", + "timeout": "endpoint check timed out", + "dns_or_connect": "endpoint connection failed", + "tls": "endpoint TLS validation failed", + "protocol": "endpoint returned an invalid HTTP response", + "redirect_limit": "endpoint exceeded the redirect limit", +} class EndpointChecker: def __init__( self, settings: Settings, + fetcher: Fetcher | None = None, resolver: Resolver = system_resolver, - transport: httpx.AsyncBaseTransport | None = None, + clock: Callable[[], float] = time.monotonic, ) -> None: self.settings = settings + self.fetcher = fetcher or AioHttpFetcher() self.resolver = resolver - self.transport = transport + self.clock = clock - async def check(self, monitor_id: str, url: str) -> CurrentStatus: - started = time.monotonic() - safe_url = redacted_url(url) + async def check(self, url: str, monitor_id: str = "unknown") -> CurrentStatus: + started = self.clock() + safe_url = redact_url(url) + error_code: str | None = None + http_status: int | None = None try: - 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 + async with asyncio.timeout(self.settings.total_timeout_seconds): + current = url + for redirects in range(self.settings.max_redirects + 1): + target = await validate_target(current, self.resolver) + remaining = max(0.001, self.settings.total_timeout_seconds - (self.clock() - started)) + response = await self.fetcher.fetch(target, self.settings, remaining) + http_status = response.status + if response.status not in {301, 302, 303, 307, 308} or not response.location: + break + if redirects == self.settings.max_redirects: + error_code = "redirect_limit" + break + current = urljoin(current, response.location) + state = HealthState.up if error_code is None and http_status is not None and 200 <= http_status < 400 else HealthState.down + except TargetPolicyError: + error_code, state = "blocked_target", HealthState.down + except TimeoutError: + error_code, state = "timeout", HealthState.down + except aiohttp.ClientConnectorCertificateError: + error_code, state = "tls", HealthState.down + except (aiohttp.ClientConnectorError, socket.gaierror, OSError): + error_code, state = "dns_or_connect", HealthState.down + except aiohttp.ClientError: + error_code, state = "protocol", HealthState.down - async def _request(self, initial_url: str, started: float) -> CurrentStatus: - timeout = httpx.Timeout( - self.settings.request_timeout_seconds, - connect=self.settings.connect_timeout_seconds, + latency = max(0.0, (self.clock() - started) * 1000) + status = CurrentStatus( + state=state, + checked_at=utc_now(), + latency_ms=round(latency, 3), + http_status=http_status, + error_code=error_code, + error_message=_ERROR_MESSAGES.get(error_code) if error_code else None, ) - 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 completed", - extra={ - "event": "endpoint_check", - "monitor_id": monitor_id, - "url": url, - "state": state, - "latency_ms": EndpointChecker._elapsed(started), - "status_code": status_code, - }, + extra={"event": "check.completed", "monitor_id": monitor_id, "url": safe_url, "outcome": state, "error_code": error_code, "latency_ms": status.latency_ms}, ) + return status