108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
import logging
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
from datetime import UTC, datetime
|
|
from urllib.parse import urljoin
|
|
|
|
import httpx
|
|
|
|
from app.config import Settings
|
|
from app.models import CurrentStatus, State
|
|
from app.security import (
|
|
DestinationRejected,
|
|
Resolver,
|
|
redacted_url,
|
|
require_public_destination,
|
|
system_resolver,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
RedirectCodes = {301, 302, 303, 307, 308}
|
|
|
|
|
|
class EndpointChecker:
|
|
def __init__(
|
|
self,
|
|
settings: Settings,
|
|
resolver: Resolver = system_resolver,
|
|
transport: httpx.AsyncBaseTransport | None = None,
|
|
) -> None:
|
|
self.settings = settings
|
|
self.resolver = resolver
|
|
self.transport = transport
|
|
|
|
async def check(self, monitor_id: str, url: str) -> CurrentStatus:
|
|
started = time.monotonic()
|
|
safe_url = redacted_url(url)
|
|
try:
|
|
result = await self._request(url, started)
|
|
except DestinationRejected:
|
|
self._log(monitor_id, safe_url, State.ERROR, started, None)
|
|
raise
|
|
except (httpx.TimeoutException, httpx.NetworkError, httpx.ProtocolError) as exc:
|
|
result = self._error(started, self._safe_error(exc))
|
|
self._log(monitor_id, safe_url, result.state, started, result.status_code)
|
|
return result
|
|
|
|
async def _request(self, initial_url: str, started: float) -> CurrentStatus:
|
|
timeout = httpx.Timeout(
|
|
self.settings.request_timeout_seconds,
|
|
connect=self.settings.connect_timeout_seconds,
|
|
)
|
|
current = initial_url
|
|
async with httpx.AsyncClient(
|
|
timeout=timeout, follow_redirects=False, transport=self.transport
|
|
) as client:
|
|
for hop in range(self.settings.max_redirects + 1):
|
|
await require_public_destination(current, self.resolver)
|
|
async with client.stream("GET", current) as response:
|
|
if response.status_code not in RedirectCodes:
|
|
state = State.UP if 200 <= response.status_code < 400 else State.DOWN
|
|
return CurrentStatus(
|
|
state=state,
|
|
checked_at=datetime.now(UTC),
|
|
status_code=response.status_code,
|
|
latency_ms=self._elapsed(started),
|
|
)
|
|
location = response.headers.get("location")
|
|
if not location:
|
|
return self._error(started, "redirect response omitted Location")
|
|
if hop == self.settings.max_redirects:
|
|
return self._error(started, "redirect limit exceeded")
|
|
current = urljoin(current, location)
|
|
return self._error(started, "redirect limit exceeded") # pragma: no cover
|
|
|
|
@staticmethod
|
|
def _elapsed(started: float) -> float:
|
|
return round((time.monotonic() - started) * 1000, 3)
|
|
|
|
def _error(self, started: float, message: str) -> CurrentStatus:
|
|
return CurrentStatus(
|
|
state=State.ERROR,
|
|
checked_at=datetime.now(UTC),
|
|
latency_ms=self._elapsed(started),
|
|
error=message[:200],
|
|
)
|
|
|
|
@staticmethod
|
|
def _safe_error(exc: Exception) -> str:
|
|
if isinstance(exc, httpx.TimeoutException):
|
|
return "outbound request timed out"
|
|
return f"outbound request failed: {type(exc).__name__}"
|
|
|
|
@staticmethod
|
|
def _log(
|
|
monitor_id: str, url: str, state: State, started: float, status_code: int | None
|
|
) -> None:
|
|
logger.info(
|
|
"endpoint check completed",
|
|
extra={
|
|
"event": "endpoint_check",
|
|
"monitor_id": monitor_id,
|
|
"url": url,
|
|
"state": state,
|
|
"latency_ms": EndpointChecker._elapsed(started),
|
|
"status_code": status_code,
|
|
},
|
|
)
|