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.
This commit is contained in:
143
app/checker.py
143
app/checker.py
@@ -1,116 +1,71 @@
|
||||
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
|
||||
from datetime import datetime
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import httpx
|
||||
|
||||
from app.config import Settings
|
||||
from app.logging import redact_url
|
||||
from app.models import CheckStatus
|
||||
from .logging_config import redact_url
|
||||
from .models import CheckResult, MonitorStatus, utcnow
|
||||
from .security import Resolver, UnsafeTarget, system_resolver, validate_target
|
||||
|
||||
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")
|
||||
logger = logging.getLogger("monitor.checker")
|
||||
|
||||
|
||||
class EndpointChecker:
|
||||
def __init__(self, client: httpx.AsyncClient, settings: Settings,
|
||||
resolver: Resolver = system_resolver) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
client: httpx.AsyncClient,
|
||||
timeout: float,
|
||||
max_redirects: int,
|
||||
max_response_bytes: int,
|
||||
resolver: Resolver = system_resolver,
|
||||
) -> None:
|
||||
self.client = client
|
||||
self.settings = settings
|
||||
self.timeout = timeout
|
||||
self.max_redirects = max_redirects
|
||||
self.max_response_bytes = max_response_bytes
|
||||
self.resolver = resolver
|
||||
|
||||
async def check(self, url: str) -> CheckStatus:
|
||||
started = time.monotonic()
|
||||
current = url
|
||||
async def check(self, monitor_id: str, initial_url: str) -> CheckResult:
|
||||
started = time.perf_counter()
|
||||
checked_at: datetime = utcnow()
|
||||
current_url = initial_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")
|
||||
for redirect_count in range(self.max_redirects + 1):
|
||||
await validate_target(current_url, self.resolver)
|
||||
async with self.client.stream(
|
||||
"GET", current, follow_redirects=False, timeout=remaining,
|
||||
headers={"User-Agent": "endpoint-monitor/1.0"},
|
||||
"GET", current_url, follow_redirects=False, timeout=self.timeout,
|
||||
headers={"User-Agent": "endpoint-monitor/1"},
|
||||
) as response:
|
||||
body_size = 0
|
||||
consumed = 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)
|
||||
consumed += len(chunk)
|
||||
if consumed >= self.max_response_bytes:
|
||||
break
|
||||
if response.is_redirect and response.headers.get("location"):
|
||||
if redirect_count == self.max_redirects:
|
||||
return self._result(started, checked_at, MonitorStatus.error, error="too_many_redirects")
|
||||
current_url = urljoin(str(response.url), response.headers["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
|
||||
status = MonitorStatus.up if 200 <= response.status_code < 400 else MonitorStatus.down
|
||||
return self._result(started, checked_at, status, response.status_code)
|
||||
except UnsafeTarget as exc:
|
||||
result = self._result(started, checked_at, MonitorStatus.error, error=str(exc))
|
||||
except httpx.TimeoutException:
|
||||
result = self._result(started, "error", error="request timed out")
|
||||
result = self._result(started, checked_at, MonitorStatus.error, error="timeout")
|
||||
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,
|
||||
})
|
||||
result = self._result(started, checked_at, MonitorStatus.error, error="network_error")
|
||||
except Exception:
|
||||
logger.exception("check_unexpected_error", extra={"monitor_id": monitor_id, "url": redact_url(initial_url)})
|
||||
result = self._result(started, checked_at, MonitorStatus.error, error="internal_error")
|
||||
self._log(monitor_id, initial_url, result)
|
||||
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)
|
||||
def _result(started: float, checked_at: datetime, status: MonitorStatus, http_status: int | None = None, error: str | None = None) -> CheckResult:
|
||||
return CheckResult(status=status, checked_at=checked_at, latency_ms=max(0.0, (time.perf_counter() - started) * 1000), http_status=http_status, error=error)
|
||||
|
||||
@staticmethod
|
||||
def _log(monitor_id: str, url: str, result: CheckResult) -> None:
|
||||
logger.info("check_completed", extra={"monitor_id": monitor_id, "url": url, **result.model_dump()})
|
||||
|
||||
Reference in New Issue
Block a user