diff --git a/app/checker.py b/app/checker.py index 1fefa0d..345dc6f 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,132 +1,50 @@ -import asyncio import logging -import socket -import time -from collections.abc import Callable, Sequence -from dataclasses import dataclass -from typing import Protocol +from time import monotonic from urllib.parse import urljoin -import aiohttp -from aiohttp.abc import AbstractResolver +import httpx -from app.config import Settings -from app.models import CurrentStatus, HealthState, utc_now -from app.security import Resolver, TargetPolicyError, ValidatedTarget, redact_url, system_resolver, validate_target +from .config import Settings +from .logging import log_event, redacted_url +from .models import CheckSnapshot, State, now +from .security import UnsafeDestination, validate_destination -logger = logging.getLogger("monitor.checker") +logger = logging.getLogger("endpoint_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}) +async def check_url(url: str, timeout: float, settings: Settings, + transport: httpx.AsyncBaseTransport | None = None) -> CheckSnapshot: + started = monotonic() + current = url + try: + async with httpx.AsyncClient(transport=transport, follow_redirects=False, + timeout=httpx.Timeout(timeout)) as client: + for redirect_count in range(settings.max_redirects + 1): + await validate_destination(current) + response = await client.get(current) + if response.is_redirect: + location = response.headers.get("location") + if not location: + break + if redirect_count == settings.max_redirects: + raise httpx.TooManyRedirects("redirect limit exceeded") + current = urljoin(current, location) + continue + elapsed = round((monotonic() - started) * 1000, 3) + state = State.up if 200 <= response.status_code < 400 else State.down + result = CheckSnapshot(state=state, checked_at=now(), latency_ms=elapsed, + http_status=response.status_code) + log_event(logger, "check_complete", url=redacted_url(current), state=state, + latency_ms=elapsed, http_status=response.status_code) + return result + raise httpx.RemoteProtocolError("redirect response missing location") + except UnsafeDestination: + log_event(logger, "check_blocked", url=redacted_url(current)) + raise + except (httpx.HTTPError, TimeoutError) as exc: + elapsed = round((monotonic() - started) * 1000, 3) + result = CheckSnapshot(state=State.error, checked_at=now(), latency_ms=elapsed, + error=type(exc).__name__) + log_event(logger, "check_error", url=redacted_url(current), state=State.error, + error=type(exc).__name__, latency_ms=elapsed) 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, - clock: Callable[[], float] = time.monotonic, - ) -> None: - self.settings = settings - self.fetcher = fetcher or AioHttpFetcher() - self.resolver = resolver - self.clock = clock - - 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: - 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 - - 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, - ) - logger.info( - "endpoint check completed", - extra={"event": "check.completed", "monitor_id": monitor_id, "url": safe_url, "outcome": state, "error_code": error_code, "latency_ms": status.latency_ms}, - ) - return status