diff --git a/src/endpoint_monitor/checker.py b/src/endpoint_monitor/checker.py new file mode 100644 index 0000000..b2d16c4 --- /dev/null +++ b/src/endpoint_monitor/checker.py @@ -0,0 +1,107 @@ +import asyncio +import ipaddress +import logging +import socket +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime +from time import perf_counter +from urllib.parse import urljoin, urlsplit + +import httpx + +from .logging import redact_url +from .models import CheckResult, MonitorStatus + +Resolver = Callable[[str, int], Awaitable[list[str]]] +logger = logging.getLogger(__name__) + + +class TargetBlockedError(ValueError): + pass + + +async def resolve_host(host: str, port: int) -> list[str]: + loop = asyncio.get_running_loop() + records = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM) + return sorted({record[4][0] for record in records}) + + +def _validate_ip(value: str) -> None: + try: + address = ipaddress.ip_address(value) + except ValueError as exc: + raise TargetBlockedError("DNS returned an invalid address") from exc + if not address.is_global: + raise TargetBlockedError("target resolves to a non-public address") + + +async def validate_target(url: str, resolver: Resolver) -> None: + parsed = urlsplit(url) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise TargetBlockedError("only absolute HTTP(S) URLs are allowed") + if parsed.username is not None or parsed.password is not None: + raise TargetBlockedError("URL user-info is not allowed") + port = parsed.port or (443 if parsed.scheme == "https" else 80) + try: + literal = ipaddress.ip_address(parsed.hostname) + except ValueError: + try: + answers = await resolver(parsed.hostname, port) + except (OSError, asyncio.TimeoutError) as exc: + raise TargetBlockedError("target DNS resolution failed") from exc + if not answers: + raise TargetBlockedError("target DNS returned no addresses") + for answer in answers: + _validate_ip(answer) + else: + _validate_ip(str(literal)) + + +class EndpointChecker: + def __init__(self, timeout: float, max_redirects: int, resolver: Resolver = resolve_host, transport: httpx.AsyncBaseTransport | None = None) -> None: + self.timeout = timeout + self.max_redirects = max_redirects + self.resolver = resolver + self.transport = transport + + async def check(self, url: str, monitor_id: str) -> CheckResult: + started = perf_counter() + checked_at = datetime.now(UTC) + current = url + try: + async with asyncio.timeout(self.timeout): + async with httpx.AsyncClient(follow_redirects=False, timeout=self.timeout, transport=self.transport) as client: + for hop in range(self.max_redirects + 1): + await validate_target(current, self.resolver) + response = await client.get(current, headers={"user-agent": "endpoint-monitor/1.0"}) + if response.is_redirect: + location = response.headers.get("location") + if not location: + raise TargetBlockedError("redirect has no location") + if hop >= self.max_redirects: + raise TargetBlockedError("redirect limit exceeded") + current = urljoin(current, location) + continue + status = MonitorStatus.UP if 200 <= response.status_code < 400 else MonitorStatus.DOWN + result = self._result(status, checked_at, started, response.status_code, current, None) + self._log(monitor_id, result) + return result + raise RuntimeError("unreachable") + except TargetBlockedError: + raise + except TimeoutError: + result = self._result(MonitorStatus.ERROR, checked_at, started, None, current, "request timed out") + except httpx.HTTPError as exc: + result = self._result(MonitorStatus.ERROR, checked_at, started, None, current, f"transport error: {type(exc).__name__}") + except (OSError, ValueError) as exc: + result = self._result(MonitorStatus.ERROR, checked_at, started, None, current, f"check error: {type(exc).__name__}") + self._log(monitor_id, result) + return result + + @staticmethod + def _result(status: MonitorStatus, checked_at: datetime, started: float, http_status: int | None, url: str, detail: str | None) -> CheckResult: + return CheckResult(status=status, checked_at=checked_at, latency_ms=round((perf_counter() - started) * 1000, 3), http_status=http_status, final_url=redact_url(url), detail=detail) + + @staticmethod + def _log(monitor_id: str, result: CheckResult) -> None: + logger.info("check_completed", extra={"monitor_id": monitor_id, "status": result.status.value, "latency_ms": result.latency_ms, "url": result.final_url})