diff --git a/app/checker.py b/app/checker.py new file mode 100644 index 0000000..28e8ac4 --- /dev/null +++ b/app/checker.py @@ -0,0 +1,117 @@ +import asyncio +import ipaddress +import json +import logging +import socket +from time import perf_counter +from urllib.parse import urljoin, urlsplit, urlunsplit + +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): + pass + + +class StaticResolver(AbstractResolver): + def __init__(self, addresses: list[str]) -> None: + self.addresses = addresses + + 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] + + 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=(",", ":"))) + + +class EndpointChecker: + def __init__(self, settings: Settings) -> None: + self.settings = settings + + 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: + started = perf_counter() + current = raw_url + event("check_started", url=redact_url(raw_url)) + 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)