diff --git a/app/checker.py b/app/checker.py index 3dff7e7..6a10819 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,125 +1,89 @@ -import asyncio -import ipaddress -import socket -from dataclasses import dataclass +import logging from time import perf_counter -from urllib.parse import urljoin, urlsplit +from urllib.parse import urljoin -import aiohttp -from aiohttp.abc import AbstractResolver +import httpx +from app.models import CheckResult, CurrentStatus, Monitor, State, now_utc +from app.security import Resolver, UnsafeTargetError, redacted_url, resolve_addresses, validate_target -class UnsafeTargetError(Exception): - pass - - -class CheckTimeoutError(Exception): - def __init__(self, latency_ms: float) -> None: - self.latency_ms = latency_ms - - -class CheckNetworkError(Exception): - def __init__(self, latency_ms: float) -> None: - self.latency_ms = latency_ms - - -class TooManyRedirectsError(Exception): - def __init__(self, latency_ms: float) -> None: - self.latency_ms = latency_ms - - -@dataclass(frozen=True) -class HttpOutcome: - status_code: int - latency_ms: float - final_url: str - - -class SafeResolver(AbstractResolver): - async def resolve( - self, host: str, port: int = 0, family: int = socket.AF_INET - ) -> list[dict[str, object]]: - try: - infos = await asyncio.to_thread( - socket.getaddrinfo, host, port, family, socket.SOCK_STREAM - ) - except socket.gaierror as exc: - raise OSError("DNS resolution failed") from exc - results: list[dict[str, object]] = [] - seen: set[str] = set() - for resolved_family, _, proto, _, sockaddr in infos: - address = sockaddr[0] - ip = ipaddress.ip_address(address) - if not ip.is_global: - raise UnsafeTargetError("target resolves to a non-public address") - if address not in seen: - seen.add(address) - results.append({ - "hostname": host, "host": address, "port": port, - "family": resolved_family, "proto": proto, "flags": 0, - }) - if not results: - raise OSError("DNS returned no addresses") - return results - - async def close(self) -> None: - return None - - async def validate_url(self, url: str) -> None: - parsed = urlsplit(url) - if parsed.scheme not in {"http", "https"} or not parsed.hostname: - raise UnsafeTargetError("only absolute HTTP(S) URLs are allowed") - if parsed.username is not None or parsed.password is not None: - raise UnsafeTargetError("URL userinfo is not allowed") - try: - port = parsed.port or (443 if parsed.scheme == "https" else 80) - except ValueError as exc: - raise UnsafeTargetError("invalid target port") from exc - await self.resolve(parsed.hostname, port, socket.AF_UNSPEC) +logger = logging.getLogger(__name__) +REDIRECTS = {301, 302, 303, 307, 308} class EndpointChecker: - def __init__(self, timeout_seconds: float, max_redirects: int) -> None: + def __init__( + self, + timeout_seconds: float, + max_redirects: int, + resolver: Resolver = resolve_addresses, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: self.timeout_seconds = timeout_seconds self.max_redirects = max_redirects - self.resolver = SafeResolver() + self.resolver = resolver + self.transport = transport - async def check(self, url: str) -> HttpOutcome: + async def check(self, monitor: Monitor) -> CheckResult: started = perf_counter() - timeout = aiohttp.ClientTimeout(total=self.timeout_seconds) - connector = aiohttp.TCPConnector( - resolver=self.resolver, ttl_dns_cache=0, force_close=True - ) + url = str(monitor.url) + state = State.ERROR + observed: int | None = None + error: str | None = None try: - async with aiohttp.ClientSession( - timeout=timeout, connector=connector, trust_env=False - ) as session: - current = url + 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 self.resolver.validate_url(current) - async with session.get(current, allow_redirects=False) as response: - if response.status in {301, 302, 303, 307, 308}: - location = response.headers.get("Location") - if location is None: - return HttpOutcome( - response.status, self._elapsed(started), current - ) - if hop == self.max_redirects: - raise TooManyRedirectsError(self._elapsed(started)) - current = urljoin(current, location) - continue - return HttpOutcome( - response.status, self._elapsed(started), current - ) + 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: - raise - except TooManyRedirectsError: - raise - except (asyncio.TimeoutError, TimeoutError) as exc: - raise CheckTimeoutError(self._elapsed(started)) from exc - except aiohttp.ClientError as exc: - raise CheckNetworkError(self._elapsed(started)) from exc + 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" - @staticmethod - def _elapsed(started: float) -> float: - return round((perf_counter() - started) * 1000, 3) + latency = round((perf_counter() - started) * 1000, 3) + status = CurrentStatus( + state=state, + checked_at=now_utc(), + observed_status=observed, + latency_ms=latency, + error=error, + ) + logger.info( + "endpoint_check", + 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, + } + }, + ) + return CheckResult(monitor_id=monitor.id, **status.model_dump())