89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
"""Outbound HTTP transport to aissurance.eu.
|
|
|
|
All communication is Router-initiated (outbound only). This module performs the
|
|
three POSTs in the protocol — discovery ``/receive``, ``/health/{tenant_id}``,
|
|
and ``/config/report`` — and classifies the HTTP response into an action the
|
|
scheduler can act on (spec §5, §12.1). It owns a single long-lived
|
|
``httpx.AsyncClient``.
|
|
"""
|
|
import enum
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
from compliance.payload import ROUTER_VERSION
|
|
from compliance.settings import ComplianceSettings
|
|
|
|
|
|
class Outcome(enum.Enum):
|
|
OK = "ok" # 202 — stored
|
|
AUTH_FAILED = "auth" # 401/403 — stop, alert, clear buffer
|
|
RATE_LIMITED = "rate" # 429 — back off
|
|
TOO_LARGE = "too_large" # 413 — split + retry
|
|
RETRY = "retry" # network error / 5xx — buffer + backoff
|
|
|
|
|
|
@dataclass
|
|
class SendResult:
|
|
outcome: Outcome
|
|
status_code: Optional[int] = None
|
|
latency_ms: int = 0
|
|
body: Optional[dict] = None
|
|
|
|
|
|
def _classify(status: int) -> Outcome:
|
|
if status == 202 or status == 200:
|
|
return Outcome.OK
|
|
if status in (401, 403):
|
|
return Outcome.AUTH_FAILED
|
|
if status == 429:
|
|
return Outcome.RATE_LIMITED
|
|
if status == 413:
|
|
return Outcome.TOO_LARGE
|
|
return Outcome.RETRY
|
|
|
|
|
|
class Transport:
|
|
def __init__(self, cfg: ComplianceSettings):
|
|
self._cfg = cfg
|
|
self._client = httpx.AsyncClient(
|
|
timeout=httpx.Timeout(10.0, connect=5.0),
|
|
verify=cfg.verify_tls,
|
|
)
|
|
|
|
async def aclose(self) -> None:
|
|
await self._client.aclose()
|
|
|
|
def _headers(self) -> dict:
|
|
return {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self._cfg.api_key}",
|
|
"X-Router-Version": ROUTER_VERSION,
|
|
"X-Router-Id": self._cfg.router_id,
|
|
}
|
|
|
|
async def _post(self, url: str, body: dict) -> SendResult:
|
|
start = time.monotonic()
|
|
try:
|
|
resp = await self._client.post(url, json=body, headers=self._headers())
|
|
except httpx.HTTPError:
|
|
return SendResult(Outcome.RETRY, None, int((time.monotonic() - start) * 1000))
|
|
latency_ms = int((time.monotonic() - start) * 1000)
|
|
parsed: Optional[dict] = None
|
|
try:
|
|
parsed = resp.json()
|
|
except ValueError:
|
|
parsed = None
|
|
return SendResult(_classify(resp.status_code), resp.status_code, latency_ms, parsed)
|
|
|
|
async def send_payload(self, payload: dict) -> SendResult:
|
|
return await self._post(self._cfg.receive_url, payload)
|
|
|
|
async def send_health(self) -> SendResult:
|
|
# Health is a keepalive/config-sync probe; an empty body is sufficient.
|
|
return await self._post(self._cfg.health_url, {"router_id": self._cfg.router_id})
|
|
|
|
async def send_config_report(self, body: dict) -> SendResult:
|
|
return await self._post(self._cfg.config_report_url, body)
|