diff --git a/app/checker.py b/app/checker.py index 28e8ac4..3dff7e7 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,117 +1,125 @@ import asyncio import ipaddress -import json -import logging import socket +from dataclasses import dataclass from time import perf_counter -from urllib.parse import urljoin, urlsplit, urlunsplit +from urllib.parse import urljoin, urlsplit import aiohttp from aiohttp.abc import AbstractResolver -from app.models import CurrentStatus, State, utcnow -from app.settings import Settings -logger = logging.getLogger("endpoint_monitor.checker") - - -class SecurityError(Exception): +class UnsafeTargetError(Exception): pass -class StaticResolver(AbstractResolver): - def __init__(self, addresses: list[str]) -> None: - self.addresses = addresses +class CheckTimeoutError(Exception): + def __init__(self, latency_ms: float) -> None: + self.latency_ms = latency_ms - async def resolve(self, host: str, port: int = 0, family: int = socket.AF_INET): - del family - return [{"hostname": host, "host": address, "port": port, - "family": socket.AF_INET6 if ":" in address else socket.AF_INET, - "proto": socket.IPPROTO_TCP, "flags": socket.AI_NUMERICHOST} - for address in self.addresses] + +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 - -def redact_url(raw: str) -> str: - try: - parts = urlsplit(raw) - host = parts.hostname or "invalid" - if ":" in host and not host.startswith("["): - host = f"[{host}]" - port = f":{parts.port}" if parts.port else "" - query = "REDACTED" if parts.query else "" - return urlunsplit((parts.scheme, host + port, parts.path, query, "")) - except (TypeError, ValueError): - return "invalid-url" - - -def event(name: str, **fields: object) -> None: - logger.info(json.dumps({"event": name, **fields}, separators=(",", ":"))) + 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) class EndpointChecker: - def __init__(self, settings: Settings) -> None: - self.settings = settings + def __init__(self, timeout_seconds: float, max_redirects: int) -> None: + self.timeout_seconds = timeout_seconds + self.max_redirects = max_redirects + self.resolver = SafeResolver() - async def _resolve(self, host: str, port: int) -> list[str]: - loop = asyncio.get_running_loop() - answers = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM, - proto=socket.IPPROTO_TCP) - addresses = sorted({answer[4][0] for answer in answers}) - if not addresses: - raise OSError("no addresses") - for raw in addresses: - address = ipaddress.ip_address(raw) - if address.ipv4_mapped is not None: - address = address.ipv4_mapped - if not address.is_global: - raise SecurityError("destination is not globally routable") - return addresses - - async def _request_once(self, url: str, addresses: list[str]) -> tuple[int, str | None]: - timeout = aiohttp.ClientTimeout(total=self.settings.request_timeout_seconds, - connect=self.settings.connect_timeout_seconds) - connector = aiohttp.TCPConnector(resolver=StaticResolver(addresses), use_dns_cache=False) - async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session: - async with session.get(url, allow_redirects=False) as response: - return response.status, response.headers.get("Location") - - async def check(self, raw_url: str) -> CurrentStatus: + async def check(self, url: str) -> HttpOutcome: started = perf_counter() - current = raw_url - event("check_started", url=redact_url(raw_url)) + timeout = aiohttp.ClientTimeout(total=self.timeout_seconds) + connector = aiohttp.TCPConnector( + resolver=self.resolver, ttl_dns_cache=0, force_close=True + ) try: - for redirects in range(self.settings.max_redirects + 1): - parsed = urlsplit(current) - if parsed.scheme not in {"http", "https"} or not parsed.hostname: - raise SecurityError("unsupported redirect destination") - if parsed.username is not None or parsed.password is not None: - raise SecurityError("URL credentials are not allowed") - port = parsed.port or (443 if parsed.scheme == "https" else 80) - addresses = await self._resolve(parsed.hostname, port) - code, location = await self._request_once(current, addresses) - if code in {301, 302, 303, 307, 308} and location is not None: - if redirects >= self.settings.max_redirects: - raise SecurityError("redirect limit exceeded") - current = urljoin(current, location) - continue - state = State.UP if 200 <= code < 400 else State.DOWN - result = CurrentStatus(state=state, checked_at=utcnow(), - latency_ms=(perf_counter() - started) * 1000, - http_status=code, final_url=redact_url(current)) - event("check_finished", url=redact_url(raw_url), state=state, - http_status=code) - return result - raise SecurityError("redirect limit exceeded") - except SecurityError: - category = "blocked_destination" - except asyncio.TimeoutError: - category = "timeout" - except (aiohttp.ClientError, OSError, ValueError): - category = "transport_error" - event("check_failed", url=redact_url(raw_url), error=category) - return CurrentStatus(state=State.ERROR, checked_at=utcnow(), - latency_ms=(perf_counter() - started) * 1000, - error=category) + async with aiohttp.ClientSession( + timeout=timeout, connector=connector, trust_env=False + ) as session: + current = url + 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 + ) + 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 + + @staticmethod + def _elapsed(started: float) -> float: + return round((perf_counter() - started) * 1000, 3)