51 lines
2.4 KiB
Python
51 lines
2.4 KiB
Python
import logging
|
|
from time import monotonic
|
|
from urllib.parse import urljoin
|
|
|
|
import httpx
|
|
|
|
from .config import Settings
|
|
from .logging import log_event, redacted_url
|
|
from .models import CheckSnapshot, State, now
|
|
from .security import UnsafeDestination, validate_destination
|
|
|
|
logger = logging.getLogger("endpoint_monitor.checker")
|
|
|
|
|
|
async def check_url(url: str, timeout: float, settings: Settings,
|
|
transport: httpx.AsyncBaseTransport | None = None) -> CheckSnapshot:
|
|
started = monotonic()
|
|
current = url
|
|
try:
|
|
async with httpx.AsyncClient(transport=transport, follow_redirects=False,
|
|
timeout=httpx.Timeout(timeout)) as client:
|
|
for redirect_count in range(settings.max_redirects + 1):
|
|
await validate_destination(current)
|
|
response = await client.get(current)
|
|
if response.is_redirect:
|
|
location = response.headers.get("location")
|
|
if not location:
|
|
break
|
|
if redirect_count == settings.max_redirects:
|
|
raise httpx.TooManyRedirects("redirect limit exceeded")
|
|
current = urljoin(current, location)
|
|
continue
|
|
elapsed = round((monotonic() - started) * 1000, 3)
|
|
state = State.up if 200 <= response.status_code < 400 else State.down
|
|
result = CheckSnapshot(state=state, checked_at=now(), latency_ms=elapsed,
|
|
http_status=response.status_code)
|
|
log_event(logger, "check_complete", url=redacted_url(current), state=state,
|
|
latency_ms=elapsed, http_status=response.status_code)
|
|
return result
|
|
raise httpx.RemoteProtocolError("redirect response missing location")
|
|
except UnsafeDestination:
|
|
log_event(logger, "check_blocked", url=redacted_url(current))
|
|
raise
|
|
except (httpx.HTTPError, TimeoutError) as exc:
|
|
elapsed = round((monotonic() - started) * 1000, 3)
|
|
result = CheckSnapshot(state=State.error, checked_at=now(), latency_ms=elapsed,
|
|
error=type(exc).__name__)
|
|
log_event(logger, "check_error", url=redacted_url(current), state=State.error,
|
|
error=type(exc).__name__, latency_ms=elapsed)
|
|
return result
|