117 lines
5.0 KiB
Python
117 lines
5.0 KiB
Python
import asyncio
|
|
import ipaddress
|
|
import logging
|
|
import socket
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from typing import Awaitable, Callable
|
|
from urllib.parse import urljoin, urlsplit
|
|
|
|
import httpx
|
|
|
|
from app.config import Settings
|
|
from app.logging import redact_url
|
|
from app.models import CheckStatus
|
|
|
|
logger = logging.getLogger(__name__)
|
|
Resolver = Callable[[str, int], Awaitable[list[str]]]
|
|
|
|
|
|
class UnsafeTarget(ValueError):
|
|
pass
|
|
|
|
|
|
async def system_resolver(host: str, port: int) -> list[str]:
|
|
loop = asyncio.get_running_loop()
|
|
answers = await loop.run_in_executor(
|
|
None, lambda: socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
|
)
|
|
return sorted({answer[4][0] for answer in answers})
|
|
|
|
|
|
async def validate_target(url: str, resolver: Resolver) -> None:
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
|
raise UnsafeTarget("only absolute HTTP(S) targets are allowed")
|
|
if parsed.username is not None or parsed.password is not None:
|
|
raise UnsafeTarget("target user information is not allowed")
|
|
try:
|
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
except ValueError as exc:
|
|
raise UnsafeTarget("invalid target port") from exc
|
|
try:
|
|
addresses = await resolver(parsed.hostname, port)
|
|
except (OSError, socket.gaierror) as exc:
|
|
raise UnsafeTarget("target DNS resolution failed") from exc
|
|
if not addresses:
|
|
raise UnsafeTarget("target DNS resolution returned no addresses")
|
|
try:
|
|
unsafe = [address for address in addresses if not ipaddress.ip_address(address).is_global]
|
|
except ValueError as exc:
|
|
raise UnsafeTarget("target DNS returned an invalid address") from exc
|
|
if unsafe:
|
|
raise UnsafeTarget("target resolves to a non-public address")
|
|
|
|
|
|
class EndpointChecker:
|
|
def __init__(self, client: httpx.AsyncClient, settings: Settings,
|
|
resolver: Resolver = system_resolver) -> None:
|
|
self.client = client
|
|
self.settings = settings
|
|
self.resolver = resolver
|
|
|
|
async def check(self, url: str) -> CheckStatus:
|
|
started = time.monotonic()
|
|
current = url
|
|
try:
|
|
for redirect_count in range(self.settings.max_redirects + 1):
|
|
await validate_target(current, self.resolver)
|
|
remaining = self.settings.check_timeout_seconds - (time.monotonic() - started)
|
|
if remaining <= 0:
|
|
raise httpx.TimeoutException("total check timeout exceeded")
|
|
async with self.client.stream(
|
|
"GET", current, follow_redirects=False, timeout=remaining,
|
|
headers={"User-Agent": "endpoint-monitor/1.0"},
|
|
) as response:
|
|
body_size = 0
|
|
async for chunk in response.aiter_bytes():
|
|
body_size += len(chunk)
|
|
if body_size > self.settings.max_response_bytes:
|
|
raise httpx.TooManyRedirects("response exceeded configured limit")
|
|
if response.is_redirect:
|
|
location = response.headers.get("location")
|
|
if not location:
|
|
return self._result(started, "error", error="redirect missing Location")
|
|
if redirect_count >= self.settings.max_redirects:
|
|
return self._result(started, "error", error="redirect limit exceeded")
|
|
current = urljoin(current, location)
|
|
continue
|
|
state = "up" if 200 <= response.status_code < 400 else "down"
|
|
result = self._result(started, state, status_code=response.status_code)
|
|
logger.info("endpoint_check_complete", extra={
|
|
"target": redact_url(current), "state": state,
|
|
"status_code": response.status_code,
|
|
"latency_ms": result.latency_ms,
|
|
})
|
|
return result
|
|
except UnsafeTarget:
|
|
logger.warning("endpoint_check_blocked", extra={"target": redact_url(current)})
|
|
raise
|
|
except httpx.TimeoutException:
|
|
result = self._result(started, "error", error="request timed out")
|
|
except httpx.HTTPError:
|
|
result = self._result(started, "error", error="HTTP transport error")
|
|
logger.info("endpoint_check_failed", extra={
|
|
"target": redact_url(current), "state": result.state,
|
|
"error": result.error, "latency_ms": result.latency_ms,
|
|
})
|
|
return result
|
|
|
|
@staticmethod
|
|
def _result(started: float, state: str, status_code: int | None = None,
|
|
error: str | None = None) -> CheckStatus:
|
|
return CheckStatus(state=state, checked_at=datetime.now(timezone.utc),
|
|
status_code=status_code,
|
|
latency_ms=round((time.monotonic() - started) * 1000, 3),
|
|
error=error)
|