118 lines
5.0 KiB
Python
118 lines
5.0 KiB
Python
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)
|