decomposer: generate deliverable files for Define the service contract and project architecture for the FastAPI endpoint monitoring service.; Implement the typed monitor CRUD API and concurrency-safe in-memory state according to the service design.; Implement secure on-demand endpoint checks with status updates, latency measurement, robust error handling, and redacted structured logs.; Add operational API endpoints and environment-driven runtime configuration to the monitoring service.; Create automated tests for the monitoring service.; Package the service with Docker and developer documentation.; Validate the complete project.
Some checks failed
ci / test (push) Has been cancelled
Some checks failed
ci / test (push) Has been cancelled
This commit is contained in:
148
app/checker.py
148
app/checker.py
@@ -1,76 +1,116 @@
|
||||
import asyncio
|
||||
import ipaddress
|
||||
import logging
|
||||
import socket
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
from urllib.parse import urljoin
|
||||
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.models import MonitorState, StatusSnapshot
|
||||
from app.security import Resolver, UnsafeTarget, redact_url, system_resolver, validate_target
|
||||
from app.logging import redact_url
|
||||
from app.models import CheckStatus
|
||||
|
||||
logger = logging.getLogger("monitor.checker")
|
||||
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, settings: Settings, resolver: Resolver = system_resolver,
|
||||
transport: httpx.AsyncBaseTransport | None = None,
|
||||
) -> None:
|
||||
def __init__(self, client: httpx.AsyncClient, settings: Settings,
|
||||
resolver: Resolver = system_resolver) -> None:
|
||||
self.client = client
|
||||
self.settings = settings
|
||||
self.resolver = resolver
|
||||
self.transport = transport
|
||||
|
||||
async def check(self, monitor_id: str, url: str) -> StatusSnapshot:
|
||||
started = time.perf_counter()
|
||||
async def check(self, url: str) -> CheckStatus:
|
||||
started = time.monotonic()
|
||||
current = url
|
||||
try:
|
||||
timeout = httpx.Timeout(
|
||||
self.settings.check_timeout_seconds,
|
||||
connect=self.settings.connect_timeout_seconds,
|
||||
)
|
||||
async with httpx.AsyncClient(
|
||||
timeout=timeout, transport=self.transport, follow_redirects=False,
|
||||
) as client:
|
||||
for hop in range(self.settings.max_redirects + 1):
|
||||
for redirect_count in range(self.settings.max_redirects + 1):
|
||||
await validate_target(current, self.resolver)
|
||||
async with client.stream("GET", current) as response:
|
||||
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(MonitorState.ERROR, started, error="bad_redirect")
|
||||
if hop == self.settings.max_redirects:
|
||||
return self._result(MonitorState.ERROR, started, error="too_many_redirects")
|
||||
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 = MonitorState.UP if 200 <= response.status_code < 400 else MonitorState.DOWN
|
||||
result = self._result(state, started, response.status_code)
|
||||
self._log(monitor_id, url, result)
|
||||
return result
|
||||
except UnsafeTarget as exc:
|
||||
result = self._result(MonitorState.BLOCKED, started, error=str(exc))
|
||||
except httpx.TimeoutException:
|
||||
result = self._result(MonitorState.ERROR, started, error="timeout")
|
||||
except httpx.HTTPError:
|
||||
result = self._result(MonitorState.ERROR, started, error="transport_error")
|
||||
self._log(monitor_id, url, result)
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _result(
|
||||
state: MonitorState, started: float, status: int | None = None,
|
||||
error: str | None = None,
|
||||
) -> StatusSnapshot:
|
||||
return StatusSnapshot(
|
||||
state=state, checked_at=datetime.now(UTC),
|
||||
latency_ms=round((time.perf_counter() - started) * 1000, 3),
|
||||
http_status=status, error=error,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _log(monitor_id: str, url: str, result: StatusSnapshot) -> None:
|
||||
logger.info("check_complete", extra={
|
||||
"monitor_id": monitor_id, "target": redact_url(url),
|
||||
"state": result.state, "latency_ms": result.latency_ms,
|
||||
"http_status": result.http_status, "error": result.error,
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user