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 / validate (push) Has been cancelled
Some checks failed
ci / validate (push) Has been cancelled
This commit is contained in:
180
app/checker.py
180
app/checker.py
@@ -1,125 +1,89 @@
|
||||
import asyncio
|
||||
import ipaddress
|
||||
import socket
|
||||
from dataclasses import dataclass
|
||||
import logging
|
||||
from time import perf_counter
|
||||
from urllib.parse import urljoin, urlsplit
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import aiohttp
|
||||
from aiohttp.abc import AbstractResolver
|
||||
import httpx
|
||||
|
||||
from app.models import CheckResult, CurrentStatus, Monitor, State, now_utc
|
||||
from app.security import Resolver, UnsafeTargetError, redacted_url, resolve_addresses, validate_target
|
||||
|
||||
class UnsafeTargetError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class CheckTimeoutError(Exception):
|
||||
def __init__(self, latency_ms: float) -> None:
|
||||
self.latency_ms = latency_ms
|
||||
|
||||
|
||||
class CheckNetworkError(Exception):
|
||||
def __init__(self, latency_ms: float) -> None:
|
||||
self.latency_ms = latency_ms
|
||||
|
||||
|
||||
class TooManyRedirectsError(Exception):
|
||||
def __init__(self, latency_ms: float) -> None:
|
||||
self.latency_ms = latency_ms
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HttpOutcome:
|
||||
status_code: int
|
||||
latency_ms: float
|
||||
final_url: str
|
||||
|
||||
|
||||
class SafeResolver(AbstractResolver):
|
||||
async def resolve(
|
||||
self, host: str, port: int = 0, family: int = socket.AF_INET
|
||||
) -> list[dict[str, object]]:
|
||||
try:
|
||||
infos = await asyncio.to_thread(
|
||||
socket.getaddrinfo, host, port, family, socket.SOCK_STREAM
|
||||
)
|
||||
except socket.gaierror as exc:
|
||||
raise OSError("DNS resolution failed") from exc
|
||||
results: list[dict[str, object]] = []
|
||||
seen: set[str] = set()
|
||||
for resolved_family, _, proto, _, sockaddr in infos:
|
||||
address = sockaddr[0]
|
||||
ip = ipaddress.ip_address(address)
|
||||
if not ip.is_global:
|
||||
raise UnsafeTargetError("target resolves to a non-public address")
|
||||
if address not in seen:
|
||||
seen.add(address)
|
||||
results.append({
|
||||
"hostname": host, "host": address, "port": port,
|
||||
"family": resolved_family, "proto": proto, "flags": 0,
|
||||
})
|
||||
if not results:
|
||||
raise OSError("DNS returned no addresses")
|
||||
return results
|
||||
|
||||
async def close(self) -> None:
|
||||
return None
|
||||
|
||||
async def validate_url(self, url: str) -> None:
|
||||
parsed = urlsplit(url)
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
||||
raise UnsafeTargetError("only absolute HTTP(S) URLs are allowed")
|
||||
if parsed.username is not None or parsed.password is not None:
|
||||
raise UnsafeTargetError("URL userinfo is not allowed")
|
||||
try:
|
||||
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
||||
except ValueError as exc:
|
||||
raise UnsafeTargetError("invalid target port") from exc
|
||||
await self.resolve(parsed.hostname, port, socket.AF_UNSPEC)
|
||||
logger = logging.getLogger(__name__)
|
||||
REDIRECTS = {301, 302, 303, 307, 308}
|
||||
|
||||
|
||||
class EndpointChecker:
|
||||
def __init__(self, timeout_seconds: float, max_redirects: int) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
timeout_seconds: float,
|
||||
max_redirects: int,
|
||||
resolver: Resolver = resolve_addresses,
|
||||
transport: httpx.AsyncBaseTransport | None = None,
|
||||
) -> None:
|
||||
self.timeout_seconds = timeout_seconds
|
||||
self.max_redirects = max_redirects
|
||||
self.resolver = SafeResolver()
|
||||
self.resolver = resolver
|
||||
self.transport = transport
|
||||
|
||||
async def check(self, url: str) -> HttpOutcome:
|
||||
async def check(self, monitor: Monitor) -> CheckResult:
|
||||
started = perf_counter()
|
||||
timeout = aiohttp.ClientTimeout(total=self.timeout_seconds)
|
||||
connector = aiohttp.TCPConnector(
|
||||
resolver=self.resolver, ttl_dns_cache=0, force_close=True
|
||||
)
|
||||
url = str(monitor.url)
|
||||
state = State.ERROR
|
||||
observed: int | None = None
|
||||
error: str | None = None
|
||||
try:
|
||||
async with aiohttp.ClientSession(
|
||||
timeout=timeout, connector=connector, trust_env=False
|
||||
) as session:
|
||||
current = url
|
||||
async with httpx.AsyncClient(
|
||||
transport=self.transport,
|
||||
timeout=httpx.Timeout(self.timeout_seconds),
|
||||
follow_redirects=False,
|
||||
) as client:
|
||||
for hop in range(self.max_redirects + 1):
|
||||
await self.resolver.validate_url(current)
|
||||
async with session.get(current, allow_redirects=False) as response:
|
||||
if response.status in {301, 302, 303, 307, 308}:
|
||||
location = response.headers.get("Location")
|
||||
if location is None:
|
||||
return HttpOutcome(
|
||||
response.status, self._elapsed(started), current
|
||||
await validate_target(url, self.resolver)
|
||||
async with client.stream(
|
||||
"GET", url, headers={"user-agent": "endpoint-monitor/0.1"}
|
||||
) as response:
|
||||
observed = response.status_code
|
||||
if observed not in REDIRECTS:
|
||||
state = (
|
||||
State.UP if observed == monitor.expected_status else State.DOWN
|
||||
)
|
||||
break
|
||||
location = response.headers.get("location")
|
||||
if not location:
|
||||
error = "redirect response omitted Location"
|
||||
break
|
||||
if hop == self.max_redirects:
|
||||
raise TooManyRedirectsError(self._elapsed(started))
|
||||
current = urljoin(current, location)
|
||||
continue
|
||||
return HttpOutcome(
|
||||
response.status, self._elapsed(started), current
|
||||
)
|
||||
error = "redirect limit exceeded"
|
||||
break
|
||||
url = urljoin(url, location)
|
||||
except UnsafeTargetError:
|
||||
raise
|
||||
except TooManyRedirectsError:
|
||||
raise
|
||||
except (asyncio.TimeoutError, TimeoutError) as exc:
|
||||
raise CheckTimeoutError(self._elapsed(started)) from exc
|
||||
except aiohttp.ClientError as exc:
|
||||
raise CheckNetworkError(self._elapsed(started)) from exc
|
||||
state = State.BLOCKED
|
||||
error = "target blocked by outbound request policy"
|
||||
except httpx.TimeoutException:
|
||||
error = "outbound request timed out"
|
||||
except httpx.HTTPError:
|
||||
error = "outbound HTTP request failed"
|
||||
except OSError:
|
||||
error = "outbound network operation failed"
|
||||
|
||||
@staticmethod
|
||||
def _elapsed(started: float) -> float:
|
||||
return round((perf_counter() - started) * 1000, 3)
|
||||
latency = round((perf_counter() - started) * 1000, 3)
|
||||
status = CurrentStatus(
|
||||
state=state,
|
||||
checked_at=now_utc(),
|
||||
observed_status=observed,
|
||||
latency_ms=latency,
|
||||
error=error,
|
||||
)
|
||||
logger.info(
|
||||
"endpoint_check",
|
||||
extra={
|
||||
"event_data": {
|
||||
"event": "endpoint_check",
|
||||
"monitor_id": str(monitor.id),
|
||||
"url": redacted_url(str(monitor.url)),
|
||||
"state": state.value,
|
||||
"observed_status": observed,
|
||||
"latency_ms": latency,
|
||||
}
|
||||
},
|
||||
)
|
||||
return CheckResult(monitor_id=monitor.id, **status.model_dump())
|
||||
|
||||
Reference in New Issue
Block a user