From 392034f83973fad2e55dbf5459fc4db779b7a16c Mon Sep 17 00:00:00 2001 From: alpha nerd Date: Wed, 10 Jun 2026 13:22:44 +0200 Subject: [PATCH] feat: aissurance plugin initial commit --- .gitignore | 3 + README.md | 18 ++ compliance/__init__.py | 22 ++ compliance/buffer.py | 94 +++++++++ compliance/collector.py | 219 ++++++++++++++++++++ compliance/crypto.py | 45 +++++ compliance/log.py | 15 ++ compliance/payload.py | 81 ++++++++ compliance/plugin.py | 129 ++++++++++++ compliance/scheduler.py | 132 ++++++++++++ compliance/settings.py | 98 +++++++++ compliance/transport.py | 89 +++++++++ config.py | 5 + config.yaml | 23 ++- db.py | 28 +++ doc/README.md | 6 + doc/compliance.md | 342 ++++++++++++++++++++++++++++++++ doc/configuration.md | 22 ++ doc/examples/sample-config.yaml | 22 +- requirements.txt | 1 + router.py | 23 +++ test/test_compliance.py | 230 +++++++++++++++++++++ 22 files changed, 1645 insertions(+), 2 deletions(-) create mode 100644 compliance/__init__.py create mode 100644 compliance/buffer.py create mode 100644 compliance/collector.py create mode 100644 compliance/crypto.py create mode 100644 compliance/log.py create mode 100644 compliance/payload.py create mode 100644 compliance/plugin.py create mode 100644 compliance/scheduler.py create mode 100644 compliance/settings.py create mode 100644 compliance/transport.py create mode 100644 doc/compliance.md create mode 100644 test/test_compliance.py diff --git a/.gitignore b/.gitignore index 7cd8431..8cfc932 100644 --- a/.gitignore +++ b/.gitignore @@ -66,4 +66,7 @@ config.yaml # SQLite *.db* +# aissurance compliance plugin — encrypted offline retry buffer +compliance-buffer/ + *settings.json \ No newline at end of file diff --git a/README.md b/README.md index fb60988..ac277dd 100644 --- a/README.md +++ b/README.md @@ -181,6 +181,24 @@ curl http://localhost:12434/api/cache/stats # hit rate, counters, config curl -X POST http://localhost:12434/api/cache/invalidate # clear all entries ``` +## Compliance Plugin (aissurance.eu) + +Optional in-process plugin that periodically sends signed AI-infrastructure +evidence — model/endpoint inventory and aggregate telemetry, **never** prompt or +completion content — to [aissurance.eu](https://www.aissurance.eu) for EU AI Act +compliance. All traffic is outbound; the Router keeps proxying if the upstream is +down (evidence is encrypted and buffered for later). + +```yaml +compliance: + enabled: true + server_url: "https://www.aissurance.eu/api/v1/discovery/receive" + api_key: "${AISSURANCE_KEY}" # tenant secret, separate from the router API key + polling_interval: 300 +``` + +See the **[Compliance Guide](doc/compliance.md)** for the full reference. + ## Supplying the router API key If you set `nomyo-router-api-key` in `config.yaml` (or `NOMYO_ROUTER_API_KEY` env), every request to NOMYO Router must include the key: diff --git a/compliance/__init__.py b/compliance/__init__.py new file mode 100644 index 0000000..5a03b05 --- /dev/null +++ b/compliance/__init__.py @@ -0,0 +1,22 @@ +"""aissurance — the aissurance.eu compliance plugin. + +An in-process module that periodically snapshots the Router's AI-infrastructure +landscape and sends structured, HMAC-signed evidence to aissurance.eu (the EU +AI Act compliance platform). It only reads infrastructure metadata the Router +already maintains for routing — never prompt/completion content. + +Public entry points: + * ``AissurancePlugin`` — owns the scheduler task; ``start()`` / ``stop()`` are + called from router.py's startup/shutdown events. +""" +__all__ = ["AissurancePlugin"] + + +def __getattr__(name): + # Lazy export so that ``config.py`` can import ``compliance.settings`` without + # eagerly loading the plugin chain (plugin → scheduler → collector → config), + # which would form an import cycle. router.py triggers this after startup. + if name == "AissurancePlugin": + from compliance.plugin import AissurancePlugin + return AissurancePlugin + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/compliance/buffer.py b/compliance/buffer.py new file mode 100644 index 0000000..20ac8be --- /dev/null +++ b/compliance/buffer.py @@ -0,0 +1,94 @@ +"""Persistent, encrypted retry buffer for offline periods. + +When the upstream is unreachable, signed payloads are written to ``buffer_dir`` +as individual ``{timestamp}-{router_id}.json.enc`` files, encrypted with a key +derived from the tenant API key (see compliance.crypto). On reconnection they +are replayed oldest-first; each is deleted only after a ``202 Accepted``. + +The buffer is bounded (``max_buffer_payloads``); the oldest file is dropped +when the limit is exceeded. +""" +import json +import re +from pathlib import Path +from typing import Optional + +from compliance import crypto +from compliance.settings import ComplianceSettings + +_SAFE = re.compile(r"[^A-Za-z0-9_.-]") + + +def _safe(s: str) -> str: + return _SAFE.sub("_", s) + + +class BufferStore: + def __init__(self, cfg: ComplianceSettings): + self._cfg = cfg + self._dir = Path(cfg.buffer_dir) + + def _ensure_dir(self) -> None: + self._dir.mkdir(parents=True, exist_ok=True) + + def _files(self) -> list[Path]: + if not self._dir.exists(): + return [] + # Filenames are ``{epoch_ms}-{router_id}.json.enc``; lexical sort on the + # zero-padded epoch prefix is chronological. + return sorted(self._dir.glob("*.json.enc")) + + def count(self) -> int: + return len(self._files()) + + def save(self, payload: dict, timestamp_ms: int) -> Optional[Path]: + """Encrypt and persist ``payload``. Returns the path, or None on failure.""" + if not self._cfg.api_key: + return None + self._ensure_dir() + self._enforce_limit(reserve=1) + name = f"{timestamp_ms:020d}-{_safe(self._cfg.router_id)}.json.enc" + path = self._dir / name + token = crypto.encrypt(json.dumps(payload).encode("utf-8"), self._cfg.api_key) + path.write_bytes(token) + return path + + def load(self, path: Path) -> Optional[dict]: + """Decrypt and parse a buffer file. Returns None (and deletes) on corruption.""" + if not self._cfg.api_key: + return None + try: + token = path.read_bytes() + except OSError: + return None + plaintext = crypto.decrypt(token, self._cfg.api_key) + if plaintext is None: + # Wrong key or corrupt file — unrecoverable, drop it. + self.delete(path) + return None + try: + return json.loads(plaintext) + except json.JSONDecodeError: + self.delete(path) + return None + + def pending(self) -> list[Path]: + """Buffered files, oldest first.""" + return self._files() + + def delete(self, path: Path) -> None: + try: + path.unlink() + except OSError: + pass + + def clear(self) -> None: + for p in self._files(): + self.delete(p) + + def _enforce_limit(self, reserve: int = 0) -> None: + """Drop oldest files until under (max - reserve).""" + limit = max(self._cfg.max_buffer_payloads - reserve, 0) + files = self._files() + while len(files) > limit: + self.delete(files.pop(0)) diff --git a/compliance/collector.py b/compliance/collector.py new file mode 100644 index 0000000..305fd0d --- /dev/null +++ b/compliance/collector.py @@ -0,0 +1,219 @@ +"""Snapshot collector — turns live Router state into the discovery data model. + +Reads only structures the Router already maintains for routing: + * ``state.usage_counts`` — live per-endpoint/model active connections + * ``config.endpoints`` / ``llama_server_endpoints`` / ``api_keys`` / limits + * ``backends.probe`` model discovery + endpoint health + * ``cache.LLMCache.stats()`` — global cache hit rate + * the token time-series DB — 24h token totals and per-model ``last_used`` + +Fields the Router does not currently instrument (per-model latency percentiles, +request/error counts, per-model cache hit rate, token averages) are emitted as +``None`` rather than fabricated. See the plugin docs / spec §4 for the rationale. +""" +import time +from datetime import datetime, timezone +from typing import Optional + +from config import Config +from state import usage_counts +from cache import get_llm_cache +from db import get_db +from routing import get_max_connections +from backends.probe import fetch, _endpoint_health +from backends.normalize import ( + is_ext_openai_endpoint, + _normalize_llama_model_name, + _extract_llama_quant, +) + +_DAY_SECONDS = 86400 + + +def _iso_now() -> str: + return datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _iso_from_unix(ts: Optional[int]) -> Optional[str]: + if not ts: + return None + return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _endpoint_type(cfg: Config, ep: str) -> str: + """Best-effort classification from config membership + URL heuristics.""" + low = ep.lower() + if "googleapis" in low or "gemini" in low: + return "gemini" + if "anthropic" in low: + return "anthropic" + if "cohere" in low: + return "cohere" + if "cerebras" in low: + return "cerebras" + if "inceptionlabs" in low or "inception" in low: + return "inception labs" + if ep in cfg.llama_server_endpoints: + # llama_server_endpoints covers both llama.cpp and vLLM (both OpenAI- + # compatible). vLLM commonly serves on :8000; treat the rest as llama.cpp. + return "vllm" if ":8000" in ep else "llama_cpp" + if is_ext_openai_endpoint(ep): + return "openai" + return "ollama" + + +def _auth_method(cfg: Config, ep: str) -> str: + return "bearer" if cfg.api_keys.get(ep) else "none" + + +def _status_label(health: dict) -> str: + # _endpoint_health only distinguishes ok/error; map onto the spec's enum. + return "healthy" if health.get("status") == "ok" else "unhealthy" + + +async def _collect_endpoints(cfg: Config) -> tuple[list[dict], int]: + """Build the endpoint registry and the global active-connection count.""" + seen: list[str] = [] + for ep in list(cfg.endpoints) + list(cfg.llama_server_endpoints): + if ep not in seen: + seen.append(ep) + + endpoints: list[dict] = [] + total_active = 0 + for ep in seen: + concurrent = sum(usage_counts.get(ep, {}).values()) + total_active += concurrent + try: + health = await _endpoint_health(ep, timeout=5) + except Exception: + health = {"status": "error"} + endpoints.append({ + "url": ep, + "type": _endpoint_type(cfg, ep), + "status": _status_label(health), + "concurrent_connections": concurrent, + "max_concurrent_connections": get_max_connections(ep), + "tls_enabled": ep.lower().startswith("https://"), + "auth_method": _auth_method(cfg, ep), + }) + return endpoints, total_active + + +def _ollama_detail_index(details: list[dict]) -> dict: + """Index Ollama /api/tags entries by model name for size/quant lookup.""" + index: dict[str, dict] = {} + for item in details or []: + name = item.get("name") or item.get("model") + if name: + index[name] = item + return index + + +async def _collect_models(cfg: Config, last_used: dict) -> list[dict]: + models: list[dict] = [] + for ep in list(cfg.endpoints) + list(cfg.llama_server_endpoints): + is_llama = ep in cfg.llama_server_endpoints + try: + available = await fetch.available_models(ep, cfg.api_keys.get(ep)) + except Exception: + available = set() + if not available: + continue + + # Ollama exposes weight size + quantization via /api/tags details. + detail_index: dict = {} + if not is_llama and not is_ext_openai_endpoint(ep): + try: + details = await fetch.endpoint_details(ep, "/api/tags", "models") + detail_index = _ollama_detail_index(details) + except Exception: + detail_index = {} + + for raw_name in sorted(available): + if is_llama: + name = _normalize_llama_model_name(raw_name) + quant = _extract_llama_quant(raw_name) or "none" + version = raw_name.split(":", 1)[1] if ":" in raw_name else "latest" + size_gb = None + else: + name = raw_name + version = raw_name.split(":", 1)[1] if ":" in raw_name else "latest" + meta = detail_index.get(raw_name, {}) + det = meta.get("details", {}) if isinstance(meta, dict) else {} + quant = det.get("quantization_level") or "none" + size_bytes = meta.get("size") if isinstance(meta, dict) else None + size_gb = round(size_bytes / 1e9, 3) if size_bytes else None + + models.append({ + "name": name, + "version": version, + "quantization": quant, + "size_gb": size_gb, + "endpoint": ep, + "last_used": _iso_from_unix(last_used.get((ep, raw_name)) or last_used.get((ep, name))), + # Un-instrumented today — emitted as null rather than fabricated. + "request_count_24h": None, + "avg_latency_ms": None, + "p99_latency_ms": None, + "error_count_24h": None, + "cache_hit_rate": None, + "avg_input_tokens": None, + "avg_output_tokens": None, + }) + return models + + +async def _collect_telemetry(cfg: Config, active_connections: int, started_at: float) -> dict: + cache = get_llm_cache() + if cache is not None: + cstats = cache.stats() + total_cache_hits = cstats.get("hits", 0) + cache_hit_rate = cstats.get("hit_rate", 0.0) + else: + total_cache_hits = 0 + cache_hit_rate = 0.0 + + cutoff = int(time.time()) - _DAY_SECONDS + try: + tokens_in, tokens_out = await get_db().get_token_totals_since(cutoff) + except Exception: + tokens_in, tokens_out = 0, 0 + + uptime_hours = round((time.monotonic() - started_at) / 3600.0, 3) + + return { + # Request/error totals are un-instrumented today (null, not fabricated). + "total_requests_24h": None, + "total_cache_hits": total_cache_hits, + "cache_hit_rate": cache_hit_rate, + "error_rate": None, + # Router tracks tokens, not raw bytes; surfaced under the byte fields as + # the closest available signal (24h windowed token counts). + "total_bytes_in_24h": tokens_in, + "total_bytes_out_24h": tokens_out, + "active_connections": active_connections, + "uptime_hours": uptime_hours, + } + + +async def collect_snapshot(cfg: Config, started_at: float) -> dict: + """Assemble the full discovery snapshot body (without signature). + + ``started_at`` is a ``time.monotonic()`` reading taken at plugin start, + used to derive ``uptime_hours``. + """ + try: + last_used = await get_db().get_last_used_map() + except Exception: + last_used = {} + + endpoints, active_connections = await _collect_endpoints(cfg) + models = await _collect_models(cfg, last_used) + telemetry = await _collect_telemetry(cfg, active_connections, started_at) + + return { + "timestamp": _iso_now(), + "models": models, + "endpoints": endpoints, + "telemetry": telemetry, + } diff --git a/compliance/crypto.py b/compliance/crypto.py new file mode 100644 index 0000000..84cfb79 --- /dev/null +++ b/compliance/crypto.py @@ -0,0 +1,45 @@ +"""Buffer encryption for offline payloads. + +Buffered discovery payloads contain only infrastructure metadata (model names, +endpoint URLs, latency/error aggregates) — never prompt or completion content. +They are still encrypted at rest with a key derived from the tenant's +``AISSURANCE_KEY`` via HKDF-SHA256, so a leaked buffer file is useless without +the tenant secret. + +The on-disk format is Fernet (AES-128-CBC + HMAC-SHA256), which authenticates +the ciphertext and embeds a timestamp. +""" +import base64 +from typing import Optional + +from cryptography.fernet import Fernet, InvalidToken +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +# Domain-separation label so this derived key can never collide with a key +# derived from the same secret for a different purpose. +_HKDF_INFO = b"aissurance-buffer-v1" + + +def _fernet(api_key: str) -> Fernet: + """Derive a Fernet instance from the tenant API key.""" + derived = HKDF( + algorithm=hashes.SHA256(), + length=32, + salt=None, + info=_HKDF_INFO, + ).derive(api_key.encode("utf-8")) + return Fernet(base64.urlsafe_b64encode(derived)) + + +def encrypt(plaintext: bytes, api_key: str) -> bytes: + """Encrypt ``plaintext`` for at-rest buffer storage.""" + return _fernet(api_key).encrypt(plaintext) + + +def decrypt(token: bytes, api_key: str) -> Optional[bytes]: + """Decrypt a buffer file. Returns None if the key is wrong or data is corrupt.""" + try: + return _fernet(api_key).decrypt(token) + except (InvalidToken, ValueError): + return None diff --git a/compliance/log.py b/compliance/log.py new file mode 100644 index 0000000..5a5fc35 --- /dev/null +++ b/compliance/log.py @@ -0,0 +1,15 @@ +"""Structured JSON logging to stderr (spec §12.2).""" +import json +import sys +from datetime import datetime, timezone + + +def log(event: str, level: str = "info", **fields) -> None: + record = { + "timestamp": datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "level": level, + "module": "aissurance", + "event": event, + **fields, + } + print(json.dumps(record), file=sys.stderr, flush=True) diff --git a/compliance/payload.py b/compliance/payload.py new file mode 100644 index 0000000..267a468 --- /dev/null +++ b/compliance/payload.py @@ -0,0 +1,81 @@ +"""Payload assembly and HMAC-SHA256 signing. + +The signature covers every payload field *except* ``hmac_signature`` itself, +over a canonical ``json.dumps(..., sort_keys=True)`` serialization (spec §6.2), +so any tampering with the body invalidates it. +""" +import hashlib +import hmac +import json +from typing import Optional + +from compliance.settings import ComplianceSettings + +# Router release this plugin ships with. Sent as ``router_version`` and in the +# ``X-Router-Version`` header. +ROUTER_VERSION = "0.9" + + +def _signing_key(cfg: ComplianceSettings) -> str: + """HMAC key: the per-tenant override if set, else the API key (spec §6.2).""" + return cfg.hmac_secret or cfg.api_key or "" + + +def canonical_message(body: dict) -> str: + """Canonical serialization signed by both client and server.""" + fields = {k: v for k, v in body.items() if k != "hmac_signature"} + return json.dumps(fields, sort_keys=True, separators=(",", ":")) + + +def sign(body: dict, cfg: ComplianceSettings) -> str: + """Return the ``sha256=`` signature for ``body``.""" + digest = hmac.new( + _signing_key(cfg).encode("utf-8"), + canonical_message(body).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return f"sha256={digest}" + + +def build_payload(cfg: ComplianceSettings, snapshot: dict) -> dict: + """Combine config identity + collector snapshot into a signed payload.""" + body = { + "tenant_id": cfg.tenant_id, + "router_id": cfg.router_id, + "router_version": ROUTER_VERSION, + "timestamp": snapshot["timestamp"], + "models": snapshot["models"], + "endpoints": snapshot["endpoints"], + "telemetry": snapshot["telemetry"], + } + body["hmac_signature"] = sign(body, cfg) + return body + + +def build_config_report(cfg: ComplianceSettings) -> dict: + """The body posted to ``/config/report`` (spec §5.3).""" + return { + "polling_interval": cfg.polling_interval, + "batch_size": cfg.batch_size, + "server_url": cfg.server_url, + "router_version": ROUTER_VERSION, + } + + +def split_into_batches(payload: dict, batch_size: int, cfg: ComplianceSettings) -> list[dict]: + """Split a payload whose ``models`` list exceeds ``batch_size`` into several + signed payloads sharing the same timestamp/identity (used on HTTP 413). + + Each batch is re-signed because its ``models`` slice differs. + """ + models = payload.get("models", []) + if batch_size <= 0 or len(models) <= batch_size: + return [payload] + + batches: list[dict] = [] + for i in range(0, len(models), batch_size): + chunk = {k: v for k, v in payload.items() if k != "hmac_signature"} + chunk["models"] = models[i : i + batch_size] + chunk["hmac_signature"] = sign(chunk, cfg) + batches.append(chunk) + return batches diff --git a/compliance/plugin.py b/compliance/plugin.py new file mode 100644 index 0000000..2ca0bf9 --- /dev/null +++ b/compliance/plugin.py @@ -0,0 +1,129 @@ +"""Plugin lifecycle: owns the periodic tasks and shared transport/buffer state. + +Instantiated and driven by router.py's startup/shutdown events. The plugin is +non-critical to the proxy: any failure here is logged and contained, never +propagated into request routing. +""" +import asyncio +import time +from typing import Optional + +from compliance import scheduler +from compliance.buffer import BufferStore +from compliance.log import log +from compliance.settings import ComplianceSettings +from compliance.transport import Transport + + +class AissurancePlugin: + def __init__(self, settings: ComplianceSettings): + self._settings = settings + self.started_at = time.monotonic() + self.stopping = False + self.sent_total = 0 + + self.transport: Optional[Transport] = None + self.buffer = BufferStore(settings) + + self._stop_event = asyncio.Event() + self._tasks: list[asyncio.Task] = [] + self._auth_failed = False + + # -- accessors used by the scheduler ------------------------------------- + @property + def cfg(self) -> ComplianceSettings: + return self._settings + + @property + def cfg_obj(self): + """Live Router Config, so the collector sees current endpoints.""" + from config import get_config + return get_config() + + async def sleep(self, seconds: float) -> None: + """Sleep up to ``seconds``, returning early if the plugin is stopping.""" + try: + await asyncio.wait_for(self._stop_event.wait(), timeout=seconds) + except asyncio.TimeoutError: + pass + + async def handle_auth_failure(self) -> None: + """401/403: stop retrying, clear the buffer, alert, and go dormant.""" + if self._auth_failed: + return + self._auth_failed = True + self.buffer.clear() + log("auth_failed", level="error", router_id=self._settings.router_id, + message="AISSURANCE_KEY rejected — compliance reporting halted") + self.stopping = True + self._stop_event.set() + + # -- lifecycle ----------------------------------------------------------- + async def start(self) -> None: + errors = self._settings.validation_errors() + if errors: + log("plugin_disabled", level="error", reasons=errors) + return + + self.transport = Transport(self._settings) + self.started_at = time.monotonic() + + if not self._settings.tenant_id: + log("tenant_id_missing", level="warning", + message="No AISSURANCE_TENANT_ID / compliance.tenant_id; payload tenant_id will be null") + + # Startup handshake: connectivity check + config report (best effort). + try: + await scheduler.run_health_cycle(self) + await scheduler.run_config_report(self) + except Exception as exc: # never let handshake failure block startup + log("handshake_failed", level="warning", error=str(exc)) + + self._tasks = [ + asyncio.create_task(self._poll_loop()), + asyncio.create_task(self._health_loop()), + ] + log("plugin_started", router_id=self._settings.router_id, + polling_interval=self._settings.polling_interval, + buffered=self.buffer.count()) + + async def _poll_loop(self) -> None: + # Emit an initial snapshot immediately so evidence appears without + # waiting a full interval, then settle into the configured cadence. + while not self.stopping: + try: + await scheduler.run_discovery_cycle(self) + except Exception as exc: # contain — plugin is non-critical + log("cycle_error", level="error", error=str(exc)) + if self.stopping: + break + await self.sleep(self._settings.polling_interval) + + async def _health_loop(self) -> None: + while not self.stopping: + await self.sleep(self._settings.health_interval) + if self.stopping: + break + try: + await scheduler.run_health_cycle(self) + except Exception as exc: + log("health_error", level="warning", error=str(exc)) + + async def stop(self) -> None: + self.stopping = True + self._stop_event.set() + for t in self._tasks: + t.cancel() + try: + await t + except (asyncio.CancelledError, Exception): + pass + self._tasks = [] + if self.transport is not None: + try: + await self.transport.aclose() + except Exception: + pass + self.transport = None + log("plugin_stopped", router_id=self._settings.router_id, + buffered=self.buffer.count(), successful=self.sent_total) diff --git a/compliance/scheduler.py b/compliance/scheduler.py new file mode 100644 index 0000000..a17ca35 --- /dev/null +++ b/compliance/scheduler.py @@ -0,0 +1,132 @@ +"""Snapshot lifecycle: collect → sign → send → buffer-on-failure, plus replay. + +These coroutines operate on an ``AissurancePlugin`` instance (duck-typed to +avoid an import cycle) and contain the per-cycle decision logic. The periodic +timing loops live in plugin.py. +""" +import asyncio +import time +from typing import TYPE_CHECKING + +from compliance import payload as payload_mod +from compliance.collector import collect_snapshot +from compliance.log import log +from compliance.transport import Outcome, SendResult + +if TYPE_CHECKING: # pragma: no cover + from compliance.plugin import AissurancePlugin + +# Snapshot collection budget (spec §9.2). Exceeding it skips the cycle. +_SNAPSHOT_TIMEOUT = 10.0 + + +async def _send_once(plugin: "AissurancePlugin", body: dict) -> SendResult: + result = await plugin.transport.send_payload(body) + if result.outcome is Outcome.TOO_LARGE: + # 413 — split by batch_size and send each chunk; worst chunk outcome wins. + batches = payload_mod.split_into_batches(body, plugin.cfg.batch_size, plugin.cfg) + if len(batches) > 1: + worst = SendResult(Outcome.OK, 202) + for chunk in batches: + r = await plugin.transport.send_payload(chunk) + if r.outcome is not Outcome.OK: + worst = r + return worst + return result + + +async def send_with_retry(plugin: "AissurancePlugin", body: dict) -> Outcome: + """Send one payload, retrying transient failures with exponential backoff. + + Returns the final outcome. The caller decides whether to buffer. + """ + attempts = plugin.cfg.max_retry_attempts + base = plugin.cfg.retry_backoff_base + for attempt in range(max(attempts, 1)): + result = await _send_once(plugin, body) + if result.outcome is Outcome.OK: + plugin.sent_total += 1 + log("payload_sent", tenant_id=plugin.cfg.tenant_id, router_id=plugin.cfg.router_id, + models_count=len(body.get("models", [])), response_code=result.status_code, + latency_ms=result.latency_ms) + return Outcome.OK + if result.outcome in (Outcome.AUTH_FAILED, Outcome.RATE_LIMITED): + # Non-retryable within this cycle. + return result.outcome + # RETRY (network/5xx): back off unless this was the last attempt. + if attempt < attempts - 1 and not plugin.stopping: + log("payload_retry", level="warning", attempt=attempt + 1, + response_code=result.status_code) + await plugin.sleep(min(base ** attempt, 60)) + return Outcome.RETRY + + +async def replay_buffer(plugin: "AissurancePlugin") -> None: + """Replay buffered payloads oldest-first. Stops at the first non-OK send.""" + for path in plugin.buffer.pending(): + if plugin.stopping: + return + body = plugin.buffer.load(path) + if body is None: + continue # corrupt file already dropped by load() + result = await plugin.transport.send_payload(body) + if result.outcome is Outcome.OK: + plugin.buffer.delete(path) + plugin.sent_total += 1 + log("payload_sent", router_id=plugin.cfg.router_id, response_code=result.status_code, + latency_ms=result.latency_ms, replayed=True) + elif result.outcome is Outcome.AUTH_FAILED: + await plugin.handle_auth_failure() + return + else: + # Still offline / rate limited — keep the rest buffered for next time. + return + + +async def run_discovery_cycle(plugin: "AissurancePlugin") -> None: + """One scheduled discovery snapshot.""" + try: + snapshot = await asyncio.wait_for( + collect_snapshot(plugin.cfg_obj, plugin.started_at), timeout=_SNAPSHOT_TIMEOUT + ) + except asyncio.TimeoutError: + log("snapshot_too_slow", level="warning", timeout_s=_SNAPSHOT_TIMEOUT) + return + except Exception as exc: # serialization / state read error — skip this cycle + log("snapshot_failed", level="error", error=str(exc)) + return + + body = payload_mod.build_payload(plugin.cfg, snapshot) + outcome = await send_with_retry(plugin, body) + + if outcome is Outcome.OK: + await replay_buffer(plugin) + elif outcome is Outcome.AUTH_FAILED: + await plugin.handle_auth_failure() + elif outcome is Outcome.RATE_LIMITED: + # Server already rate-limited us; drop this snapshot (next is as good). + log("payload_rate_limited", level="warning") + else: # RETRY exhausted — persist for later. + path = plugin.buffer.save(body, int(time.time() * 1000)) + log("payload_buffered", level="warning", buffered=plugin.buffer.count(), + path=str(path) if path else None) + + +async def run_config_report(plugin: "AissurancePlugin") -> None: + """POST current config to /config/report (handshake / on change).""" + result = await plugin.transport.send_config_report( + payload_mod.build_config_report(plugin.cfg) + ) + if result.outcome is Outcome.AUTH_FAILED: + await plugin.handle_auth_failure() + return + suggested = (result.body or {}).get("server_suggested_polling_interval") + log("config_reported", response_code=result.status_code, + server_suggested_polling_interval=suggested) + + +async def run_health_cycle(plugin: "AissurancePlugin") -> None: + """Periodic health/keepalive + config-sync probe.""" + result = await plugin.transport.send_health() + if result.outcome is Outcome.AUTH_FAILED: + await plugin.handle_auth_failure() diff --git a/compliance/settings.py b/compliance/settings.py new file mode 100644 index 0000000..812aee3 --- /dev/null +++ b/compliance/settings.py @@ -0,0 +1,98 @@ +"""Configuration model for the aissurance compliance plugin. + +Populated from the ``compliance`` block of the Router's ``config.yaml`` (see +config.Config). ``${VAR}`` references in the YAML are already expanded by +``Config._expand_env_refs`` before this model is constructed; on top of that, +the aissurance-specific environment variables (``AISSURANCE_KEY`` etc.) are +honoured as defaults so the plugin works with env-only configuration too. + +This module imports nothing from the Router, so config.py can import it +without creating a cycle. +""" +import os +import socket +from typing import Optional + +from pydantic import BaseModel, Field, field_validator + + +def _env_or_none(name: str) -> Optional[str]: + val = os.getenv(name) + return val if val else None + + +class ComplianceSettings(BaseModel): + """The ``compliance:`` config block.""" + + enabled: bool = False + + # Full URL of the discovery receive endpoint. Health and config-report URLs + # are derived from this by replacing the trailing ``/receive`` path. + server_url: str = "https://www.aissurance.eu/api/v1/discovery/receive" + + # Tenant-provisioned secret (AISSURANCE_KEY). Used both as the Bearer token + # and as the HMAC signing key unless ``hmac_secret`` overrides the latter. + api_key: Optional[str] = Field(default_factory=lambda: _env_or_none("AISSURANCE_KEY")) + # Optional per-tenant HMAC secret override (AISSURANCE_HMAC_SECRET). + hmac_secret: Optional[str] = Field(default_factory=lambda: _env_or_none("AISSURANCE_HMAC_SECRET")) + # Tenant UUID, if not embedded in the key (AISSURANCE_TENANT_ID). + tenant_id: Optional[str] = Field(default_factory=lambda: _env_or_none("AISSURANCE_TENANT_ID")) + + # Unique identifier for this Router instance. Multiple instances per tenant + # are allowed, so this defaults to the hostname rather than the tenant id. + router_id: str = Field(default_factory=socket.gethostname) + + polling_interval: int = 300 # seconds between discovery snapshots + health_interval: int = 3600 # seconds between health/config-sync calls + batch_size: int = 50 # max models per payload + max_retry_attempts: int = 10 # in-cycle send retries before buffering + retry_backoff_base: int = 2 # exponential backoff base (seconds) + + buffer_dir: str = "./compliance-buffer" # persistent encrypted retry storage + max_buffer_payloads: int = 100 # oldest dropped first when full + + # TLS verification. AISSURANCE_VERIFY_TLS=0 disables it (dev only). + verify_tls: bool = Field( + default_factory=lambda: os.getenv("AISSURANCE_VERIFY_TLS", "1") != "0" + ) + + @field_validator("api_key", "hmac_secret", "tenant_id", mode="before") + @classmethod + def _empty_to_none(cls, v): + # config.yaml ``${VAR}`` expansion yields "" for unset vars; treat as None + # so the env-var default_factory can take over. + if v == "": + return None + return v + + @property + def discovery_base_url(self) -> str: + """Base discovery URL, derived by stripping a trailing ``/receive``.""" + url = self.server_url.rstrip("/") + if url.endswith("/receive"): + url = url[: -len("/receive")] + return url + + @property + def receive_url(self) -> str: + return self.server_url + + @property + def health_url(self) -> str: + tid = self.tenant_id or "unknown" + return f"{self.discovery_base_url}/health/{tid}" + + @property + def config_report_url(self) -> str: + return f"{self.discovery_base_url}/config/report" + + def validation_errors(self) -> list[str]: + """Return human-readable reasons the plugin cannot run, or [] if OK.""" + errors: list[str] = [] + if not self.api_key: + errors.append("AISSURANCE_KEY / compliance.api_key is not set") + if not self.server_url.lower().startswith("https://"): + errors.append(f"server_url must be HTTPS (got {self.server_url!r})") + if self.polling_interval < 60: + errors.append("polling_interval must be >= 60 seconds") + return errors diff --git a/compliance/transport.py b/compliance/transport.py new file mode 100644 index 0000000..bb2d39a --- /dev/null +++ b/compliance/transport.py @@ -0,0 +1,89 @@ +"""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) diff --git a/config.py b/config.py index 143a2f9..7db9c11 100644 --- a/config.py +++ b/config.py @@ -13,6 +13,8 @@ import yaml from pydantic import Field from pydantic_settings import BaseSettings +from compliance.settings import ComplianceSettings + class Config(BaseSettings): # List of Ollama endpoints @@ -62,6 +64,9 @@ class Config(BaseSettings): # 0.3 = 30% history context signal, 70% question signal cache_history_weight: float = Field(default=0.3) + # aissurance.eu compliance plugin (`compliance:` block). See compliance/. + compliance: ComplianceSettings = Field(default_factory=ComplianceSettings) + class Config: # YAML loading is handled manually via Config.from_yaml(); env vars use this prefix. env_prefix = "NOMYO_ROUTER_" diff --git a/config.yaml b/config.yaml index 2107a3c..998810f 100644 --- a/config.yaml +++ b/config.yaml @@ -91,4 +91,25 @@ api_keys: # Weight of the BM25-weighted chat-history embedding vs last-user-message embedding. # 0.3 = 30% history context signal, 70% question signal. # Only relevant when cache_similarity < 1.0. -# cache_history_weight: 0.3 \ No newline at end of file +# cache_history_weight: 0.3 + +# ------------------------------------------------------------- +# aissurance.eu compliance plugin (optional — disabled by default) +# Periodically snapshots the AI-infrastructure landscape (model/endpoint +# inventory + aggregate telemetry — never prompt/completion content) and +# sends HMAC-signed evidence to aissurance.eu. All traffic is outbound. +# See compliance/ for the implementation. +# ------------------------------------------------------------- +# compliance: +# enabled: true +# server_url: "https://www.aissurance.eu/api/v1/discovery/receive" +# api_key: "${AISSURANCE_KEY}" # tenant secret (distinct from nomyo-router-api-key) +# tenant_id: "${AISSURANCE_TENANT_ID}" # optional if embedded in the key +# router_id: "router-prod-001" # defaults to the machine hostname +# polling_interval: 300 # seconds between discovery snapshots +# health_interval: 3600 # seconds between health/keepalive calls +# batch_size: 50 # max models per payload (413 → auto-split) +# max_retry_attempts: 10 # in-cycle send retries before buffering +# retry_backoff_base: 2 # exponential backoff base (seconds) +# buffer_dir: "./compliance-buffer" # encrypted offline retry storage +# max_buffer_payloads: 100 # oldest dropped first when full \ No newline at end of file diff --git a/db.py b/db.py index 3621144..051fed0 100644 --- a/db.py +++ b/db.py @@ -255,6 +255,34 @@ class TokenDatabase: } return None + async def get_last_used_map(self) -> dict: + """Return {(endpoint, model): latest_unix_timestamp} from the time series. + + Read-only aggregate used by the compliance collector to derive each + model's ``last_used``. Computed in SQL in a single pass. + """ + db = await self._get_connection() + async with self._operation_lock: + async with db.execute(''' + SELECT endpoint, model, MAX(timestamp) + FROM token_time_series + GROUP BY endpoint, model + ''') as cursor: + rows = await cursor.fetchall() + return {(row[0], row[1]): row[2] for row in rows} + + async def get_token_totals_since(self, cutoff_ts: int) -> tuple[int, int]: + """Return (input_tokens, output_tokens) summed over entries at/after cutoff_ts.""" + db = await self._get_connection() + async with self._operation_lock: + async with db.execute(''' + SELECT COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0) + FROM token_time_series + WHERE timestamp >= ? + ''', (cutoff_ts,)) as cursor: + row = await cursor.fetchone() + return (int(row[0]), int(row[1])) if row else (0, 0) + async def aggregate_time_series_older_than(self, days: int, trim_old: bool = False) -> int: """ Aggregate time_series entries older than 'days' days into daily aggregates by diff --git a/doc/README.md b/doc/README.md index 3909089..9fde189 100644 --- a/doc/README.md +++ b/doc/README.md @@ -11,6 +11,7 @@ doc/ ├── usage.md # API usage examples ├── deployment.md # Deployment scenarios ├── monitoring.md # Monitoring and troubleshooting +├── compliance.md # aissurance.eu compliance plugin └── examples/ # Example configurations and scripts ├── docker-compose.yml ├── sample-config.yaml @@ -59,6 +60,7 @@ uvicorn router:app --host 0.0.0.0 --port 12434 - **Real-time Monitoring**: Server-Sent Events for live usage updates - **OpenAI Compatibility**: Full OpenAI API compatibility layer - **MOE System**: Multiple Opinions Ensemble for improved responses with smaller models +- **Compliance Plugin**: Optional EU AI Act evidence reporting to aissurance.eu ## Documentation Guides @@ -82,6 +84,10 @@ Step-by-step deployment guides for bare metal, Docker, Kubernetes, and productio Monitoring endpoints, troubleshooting guides, performance tuning, and best practices for maintaining your router. +### [Compliance](compliance.md) + +The optional aissurance.eu compliance plugin: what it collects (and what it never does), configuration, payload structure, security model, and troubleshooting. + ## Examples The [examples](examples/) directory contains ready-to-use configuration files: diff --git a/doc/compliance.md b/doc/compliance.md new file mode 100644 index 0000000..3cd5770 --- /dev/null +++ b/doc/compliance.md @@ -0,0 +1,342 @@ +# Compliance Plugin (aissurance.eu) + +The compliance plugin turns the NOMYO Router from a pure traffic proxy into an +**AI compliance agent**. When enabled, it periodically snapshots your +AI-infrastructure landscape — model inventory, endpoint registry, and aggregate +telemetry — and sends signed, structured evidence to +[aissurance.eu](https://www.aissurance.eu), the EU AI Actf compliance platform. + +The plugin runs **in-process** inside the Router; it is not a separate service. +It only reads infrastructure metadata the Router already maintains for routing. +**No prompt or completion content is ever read, stored, or transmitted.** + +> **Non-critical by design.** Any failure inside the plugin is logged and +> contained — it never affects request routing. If the upstream is unreachable, +> the Router keeps proxying and buffers evidence for later. + +--- + +## How it works + +``` + Router state (models, endpoints, usage, cache, token DB) + │ read-only snapshot, every polling_interval + ▼ + Snapshot Collector → Payload Builder → HMAC-SHA256 Signer + │ + ▼ + HTTPS POST (outbound only) ──────────────► aissurance.eu + │ 202 Accepted → done /api/v1/discovery/receive + │ network error / 5xx → encrypt + buffer to disk, retry next cycle + ▼ + Encrypted offline buffer (replayed oldest-first on reconnect) +``` + +**All communication is outbound from the Router.** aissurance.eu never +initiates contact, so the Router can run behind NAT, firewalls, or strict egress +policies. + +--- + +## What is and isn't collected + +This is the most important part to understand before enabling the plugin. + +### Collected (infrastructure metadata only) + +- **Model inventory** — name, version, quantization, weight size, serving endpoint +- **Endpoint registry** — URL, type, health status, concurrency, TLS, auth method +- **Aggregate telemetry** — active connections, uptime, global cache hit rate, + 24h token totals + +### Never collected + +- Individual request payloads (prompt / completion content) +- User identity or auth tokens from proxied requests +- Conversation content passing through the proxy +- End-user client IP addresses + +### Emitted as `null` (not yet instrumented) + +The Router does not currently measure the following, so the plugin emits `null` +rather than fabricating values. They are candidates for a future instrumentation +phase: + +| Field | Where | +|---|---| +| `request_count_24h`, `error_count_24h` | per model | +| `avg_latency_ms`, `p99_latency_ms` | per model | +| `cache_hit_rate`, `avg_input_tokens`, `avg_output_tokens` | per model | +| `total_requests_24h`, `error_rate` | telemetry | + +> **Note on byte fields.** The Router tracks **tokens**, not raw bytes. +> `total_bytes_in_24h` / `total_bytes_out_24h` therefore carry 24h **token** +> counts as the closest available signal. + +--- + +## Quick start + +1. Obtain your `AISSURANCE_KEY` (and optionally `AISSURANCE_TENANT_ID`) from + aissurance.eu during tenant signup. This is **distinct** from the Router's + own `nomyo-router-api-key` — see [Dual-key model](#dual-key-model). + +2. Export the secret (never commit it): + + ```bash + export AISSURANCE_KEY="aiss_live_…" + export AISSURANCE_TENANT_ID="550e8400-e29b-41d4-a716-446655440000" # optional + ``` + +3. Add the `compliance` block to `config.yaml`: + + ```yaml + compliance: + enabled: true + server_url: "https://www.aissurance.eu/api/v1/discovery/receive" + api_key: "${AISSURANCE_KEY}" + tenant_id: "${AISSURANCE_TENANT_ID}" + polling_interval: 300 + ``` + +4. Restart the Router. On startup you should see structured logs: + + ```json + {"event":"plugin_started","router_id":"router-prod-001","polling_interval":300,"buffered":0} + ``` + +--- + +## Configuration options + +All options live under the `compliance:` block in `config.yaml`. + +| Option | Type | Default | Description | +|---|---|---|---| +| `enabled` | `bool` | `false` | Master switch. All other options are ignored when `false`. | +| `server_url` | `str` | `https://www.aissurance.eu/api/v1/discovery/receive` | Discovery receive URL. Must be HTTPS. Health and config-report URLs are derived from it. | +| `api_key` | `str` | `${AISSURANCE_KEY}` | Tenant secret. Used as the Bearer token and as the HMAC signing key. | +| `tenant_id` | `str` | `${AISSURANCE_TENANT_ID}` | Tenant UUID. Optional if embedded in the key. | +| `router_id` | `str` | machine hostname | Unique identifier for this Router instance. Multiple instances per tenant are allowed. | +| `polling_interval` | `int` | `300` | Seconds between discovery snapshots. Minimum `60`. | +| `health_interval` | `int` | `3600` | Seconds between health / keepalive calls. | +| `batch_size` | `int` | `50` | Max models per payload. On `413` the payload is auto-split into batches. | +| `max_retry_attempts` | `int` | `10` | In-cycle send retries before the payload is buffered to disk. | +| `retry_backoff_base` | `int` | `2` | Exponential backoff base in seconds (`base ** attempt`, capped at 60s). | +| `buffer_dir` | `str` | `./compliance-buffer` | Directory for encrypted offline retry files. | +| `max_buffer_payloads` | `int` | `100` | Buffer cap. Oldest file is dropped first when exceeded. | + +### `polling_interval` + +How often a snapshot is collected and sent. The server enforces a maximum of one +payload per `polling_interval` per tenant; sending faster yields `429`. Values +below `60` are rejected at startup. + +### `batch_size` + +Caps how many models a single payload carries. If the upstream replies `413 +Payload Too Large`, the plugin splits the model list into `batch_size`-sized +chunks, re-signs each chunk, and sends them individually. + +### `buffer_dir` and `max_buffer_payloads` + +When the upstream is unreachable, each signed payload is encrypted and written to +`buffer_dir`. The directory is bounded by `max_buffer_payloads`; once full, the +oldest file is dropped before a new one is written. Buffered payloads are +replayed oldest-first on the next successful cycle. Add `buffer_dir` to your +backup exclusions — files are encrypted but transient. + +--- + +## Environment variables + +| Variable | Required | Description | +|---|---|---| +| `AISSURANCE_KEY` | Yes | Tenant secret issued at signup. Referenced by `api_key: "${AISSURANCE_KEY}"`. | +| `AISSURANCE_HMAC_SECRET` | No | Per-tenant HMAC secret override. When set, it signs payloads instead of `AISSURANCE_KEY`. | +| `AISSURANCE_TENANT_ID` | No | Tenant UUID, if not embedded in the key. | +| `AISSURANCE_VERIFY_TLS` | No | Set to `"0"` to allow self-signed certs (development only — logged as a warning). | + +Values referenced as `${VAR}` in `config.yaml` are expanded at load time. If a +referenced variable is unset, the corresponding environment-variable default +above takes over. + +--- + +## Outbound endpoints + +All three are Router-initiated POSTs. URLs are derived from `server_url` by +replacing the trailing `/receive` path. + +| Endpoint | When called | Purpose | +|---|---|---| +| `POST …/discovery/receive` | every `polling_interval` | Send the signed discovery payload. | +| `POST …/discovery/health/{tenant_id}` | on startup + every `health_interval` | Connectivity check / keepalive / config sync. | +| `POST …/discovery/config/report` | on startup handshake | Report the active compliance config for acknowledgment. | + +Every request carries: + +``` +Authorization: Bearer ${AISSURANCE_KEY} +X-Router-Version: +X-Router-Id: +Content-Type: application/json +``` + +### Response handling + +| Status | Meaning | Plugin action | +|---|---|---| +| `202` / `200` | Accepted | Mark sent; replay any buffered payloads. | +| `401` / `403` | Invalid key / blacklisted | **Stop reporting**, clear the buffer, log `auth_failed`. | +| `429` | Rate limited | Drop this snapshot (the next one is just as fresh). | +| `413` | Too large | Split by `batch_size`, re-sign, resend. | +| `5xx` / network error | Server / connectivity failure | Retry with backoff, then encrypt + buffer to disk. | + +--- + +## Payload structure + +```json +{ + "tenant_id": "550e8400-e29b-41d4-a716-446655440000", + "router_id": "router-prod-001", + "router_version": "0.9", + "timestamp": "2026-06-10T10:30:00Z", + "models": [ + { + "name": "mistral-7b-instruct", + "version": "latest", + "quantization": "Q4_K_M", + "size_gb": 4.1, + "endpoint": "http://192.168.0.50:11434", + "last_used": "2026-06-10T10:28:14Z", + "request_count_24h": null, + "avg_latency_ms": null, + "p99_latency_ms": null, + "error_count_24h": null, + "cache_hit_rate": null, + "avg_input_tokens": null, + "avg_output_tokens": null + } + ], + "endpoints": [ + { + "url": "http://192.168.0.50:11434", + "type": "ollama", + "status": "healthy", + "concurrent_connections": 2, + "max_concurrent_connections": 4, + "tls_enabled": false, + "auth_method": "none" + } + ], + "telemetry": { + "total_requests_24h": null, + "total_cache_hits": 1547, + "cache_hit_rate": 0.634, + "error_rate": null, + "total_bytes_in_24h": 9120344, + "total_bytes_out_24h": 14882910, + "active_connections": 2, + "uptime_hours": 72.4 + }, + "hmac_signature": "sha256=abcdef…" +} +``` + +`endpoint.type` is best-effort: `ollama`, `llama_cpp`, `vllm`, `openai`, +`gemini`, `anthropic`, `cohere`, `cerebras`, `inception labs`, or `other`. +`endpoint.status` maps the Router's health probe onto `healthy` / `unhealthy`. + +--- + +## Security + +### Dual-key model + +| Credential | Protects | Managed by | +|---|---|---| +| `nomyo-router-api-key` | The Router's own API / dashboard | Router operator | +| `AISSURANCE_KEY` | The upstream reporting channel | aissurance.eu (issued at signup) | + +These are independent. See [Router API key usage](configuration.md#using-the-router-api-key) +for the former. + +### Payload signing + +Every payload is HMAC-SHA256 signed. The signature covers all fields **except** +`hmac_signature`, over a canonical `sort_keys` serialization, so any tampering +with the body invalidates it. The signing key is `AISSURANCE_HMAC_SECRET` if set, +otherwise `AISSURANCE_KEY`. + +### Buffer encryption + +Offline buffer files are encrypted at rest with a key derived from your +`AISSURANCE_KEY` via HKDF-SHA256 (Fernet / AES-128-CBC + HMAC). A leaked buffer +file is useless without the tenant secret. Corrupt or wrong-key files are dropped +on read. + +### Transport + +TLS is required (`server_url` must be HTTPS). `AISSURANCE_VERIFY_TLS=0` disables +certificate verification for development only and is logged as a warning. + +--- + +## Observability + +The plugin writes structured JSON logs to stderr (one object per line): + +```json +{"timestamp":"2026-06-10T10:30:00Z","level":"info","module":"aissurance","event":"payload_sent","tenant_id":"abc123","router_id":"router-prod-001","models_count":12,"response_code":202,"latency_ms":245} +``` + +Event types: + +| Event | Meaning | +|---|---| +| `plugin_started` / `plugin_stopped` | Lifecycle. `plugin_stopped` reports buffered + successful counts. | +| `plugin_disabled` | Startup validation failed (e.g. missing key, non-HTTPS URL). Reasons included. | +| `payload_sent` | A discovery payload (or a replayed buffer file) was accepted. | +| `payload_buffered` | Upstream unreachable; payload encrypted to disk. | +| `payload_retry` | A transient failure is being retried with backoff. | +| `payload_rate_limited` | Server returned `429`; snapshot dropped. | +| `auth_failed` | Key rejected (`401`/`403`); reporting halted, buffer cleared. **Operator action required.** | +| `config_reported` | Config handshake completed; any server suggestion included. | +| `snapshot_too_slow` | Collection exceeded its 10s budget; cycle skipped. | +| `tenant_id_missing` | No tenant id resolved; payloads will carry `tenant_id: null`. | + +--- + +## Lifecycle + +**Startup** — validate config (HTTPS URL, key present, `polling_interval ≥ 60`); +on failure the plugin logs `plugin_disabled` and stays dormant without affecting +the Router. On success it performs a health + config-report handshake, then +starts the snapshot and health loops. + +**Runtime** — the first snapshot is emitted immediately so evidence appears +without waiting a full interval, then the plugin settles into `polling_interval`. +Snapshot collection runs under a 10s budget and reads from shared Router state +without blocking request routing. + +**Shutdown** — the loops stop, the HTTP client closes, and a `plugin_stopped` +log records how many payloads were buffered vs. sent. The Router continues and +completes its own shutdown normally. + +> **Config changes require a restart.** Like the rest of the Router config, the +> `compliance` block is read at startup. Hot-reload of the compliance settings is +> not yet supported. + +--- + +## Troubleshooting + +| Symptom | Likely cause | Fix | +|---|---|---| +| `plugin_disabled` at startup | Missing key, non-HTTPS `server_url`, or `polling_interval < 60` | Check the `reasons` array in the log line. | +| `auth_failed`, reporting stops | `AISSURANCE_KEY` invalid/expired/blacklisted | Re-provision the key on aissurance.eu and restart. | +| Buffer files accumulating | Upstream unreachable | Confirm egress to aissurance.eu over HTTPS; check firewall/NAT. | +| `tenant_id` is `null` in payloads | No `AISSURANCE_TENANT_ID` / `compliance.tenant_id` | Set it, unless your key embeds the tenant. | +| `snapshot_too_slow` warnings | Slow/unhealthy backends delaying health probes | Investigate endpoint health; the cycle self-recovers next tick. | diff --git a/doc/configuration.md b/doc/configuration.md index 1addd66..ef9ebfd 100644 --- a/doc/configuration.md +++ b/doc/configuration.md @@ -532,6 +532,28 @@ docker build -t nomyo-router . docker build --build-arg SEMANTIC_CACHE=true -t nomyo-router:semantic . ``` +## Compliance Plugin (aissurance.eu) + +The optional `compliance` block enables the aissurance.eu compliance plugin, +which periodically sends signed AI-infrastructure evidence (model/endpoint +inventory + aggregate telemetry — never prompt or completion content) to +aissurance.eu for EU AI Act compliance. + +```yaml +compliance: + enabled: true + server_url: "https://www.aissurance.eu/api/v1/discovery/receive" + api_key: "${AISSURANCE_KEY}" # tenant secret (distinct from nomyo-router-api-key) + tenant_id: "${AISSURANCE_TENANT_ID}" # optional if embedded in the key + polling_interval: 300 # seconds between discovery snapshots +``` + +The `AISSURANCE_KEY` is separate from `router_api_key` — one secures the proxy, +the other secures the upstream reporting channel. All traffic is outbound. + +See the **[Compliance Guide](compliance.md)** for the full option reference, +payload structure, security model, and troubleshooting. + ## Configuration Validation The router validates the configuration at startup: diff --git a/doc/examples/sample-config.yaml b/doc/examples/sample-config.yaml index a1081af..f252ec7 100644 --- a/doc/examples/sample-config.yaml +++ b/doc/examples/sample-config.yaml @@ -63,4 +63,24 @@ api_keys: # Weight of the BM25-weighted chat-history embedding vs last-user-message embedding. # 0.3 = 30% history context signal, 70% question signal. # Only relevant when cache_similarity < 1.0. -# cache_history_weight: 0.3 \ No newline at end of file +# cache_history_weight: 0.3 + +# ------------------------------------------------------------- +# aissurance.eu compliance plugin (optional — disabled by default) +# Sends signed AI-infrastructure evidence (model/endpoint inventory + +# aggregate telemetry — never prompt/completion content) to aissurance.eu. +# All traffic is outbound. See doc/compliance.md for the full reference. +# ------------------------------------------------------------- +# compliance: +# enabled: true +# server_url: "https://www.aissurance.eu/api/v1/discovery/receive" +# api_key: "${AISSURANCE_KEY}" # tenant secret (separate from nomyo-router-api-key) +# tenant_id: "${AISSURANCE_TENANT_ID}" # optional if embedded in the key +# router_id: "router-prod-001" # defaults to the machine hostname +# polling_interval: 300 # seconds between discovery snapshots +# health_interval: 3600 # seconds between health/keepalive calls +# batch_size: 50 # max models per payload (413 → auto-split) +# max_retry_attempts: 10 # in-cycle send retries before buffering +# retry_backoff_base: 2 # exponential backoff base (seconds) +# buffer_dir: "./compliance-buffer" # encrypted offline retry storage +# max_buffer_payloads: 100 # oldest dropped first when full \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index ba20023..0d61e91 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,6 +7,7 @@ async-timeout==5.0.1 attrs==26.1.0 certifi==2026.4.22 click==8.4.0 +cryptography==48.0.1 distro==1.9.0 exceptiongroup==1.3.1 fastapi==0.136.1 diff --git a/router.py b/router.py index 676e42b..bbd6b15 100644 --- a/router.py +++ b/router.py @@ -68,6 +68,10 @@ from state import ( token_worker_task: asyncio.Task | None = None flush_task: asyncio.Task | None = None +# aissurance compliance plugin instance (created on startup if enabled). +from compliance import AissurancePlugin +compliance_plugin: "AissurancePlugin | None" = None + from config import Config, _config_path_from_env from ollama._types import TokenLogprob, Logprob @@ -399,10 +403,29 @@ async def startup_event() -> None: flush_task = asyncio.create_task(flush_buffer()) await init_llm_cache(config) + # Start the aissurance compliance plugin (non-critical; guarded by config). + global compliance_plugin + if config.compliance.enabled: + try: + compliance_plugin = AissurancePlugin(config.compliance) + await compliance_plugin.start() + except Exception as e: + print(f"[startup] aissurance compliance plugin failed to start: {e}") + compliance_plugin = None + @app.on_event("shutdown") async def shutdown_event() -> None: await close_all_sse_queues() + # Stop the compliance plugin first (best-effort buffer flush on its own). + global compliance_plugin + if compliance_plugin is not None: + try: + await compliance_plugin.stop() + except Exception as e: + print(f"[shutdown] Error stopping aissurance compliance plugin: {e}") + compliance_plugin = None + # Stop background tasks first so they stop touching the DB before we close it. for t in (token_worker_task, flush_task): if t is not None: diff --git a/test/test_compliance.py b/test/test_compliance.py new file mode 100644 index 0000000..754a345 --- /dev/null +++ b/test/test_compliance.py @@ -0,0 +1,230 @@ +"""Unit tests for the aissurance compliance plugin. + +Covers the pieces that don't require a live upstream or backend: + * config/settings resolution and URL derivation + * HMAC payload signing (determinism + tamper detection) + * buffer encrypt/decrypt round-trip + bound enforcement + * transport status-code → outcome classification + * collector field mapping against a mocked Router state +""" +import json +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from compliance.settings import ComplianceSettings +from compliance import payload as payload_mod +from compliance import crypto +from compliance.buffer import BufferStore +from compliance.transport import Outcome, _classify + + +# ── settings ──────────────────────────────────────────────────────────────── + +def test_url_derivation(): + s = ComplianceSettings( + server_url="https://www.aissurance.eu/api/v1/discovery/receive", + tenant_id="tid-123", + ) + assert s.receive_url.endswith("/discovery/receive") + assert s.health_url == "https://www.aissurance.eu/api/v1/discovery/health/tid-123" + assert s.config_report_url == "https://www.aissurance.eu/api/v1/discovery/config/report" + + +def test_health_url_without_tenant(): + s = ComplianceSettings(tenant_id=None) + assert s.health_url.endswith("/health/unknown") + + +def test_validation_errors(): + assert ComplianceSettings(api_key=None).validation_errors() # missing key + assert ComplianceSettings(api_key="k", server_url="http://x").validation_errors() # not https + assert ComplianceSettings(api_key="k", polling_interval=5).validation_errors() # too fast + assert ComplianceSettings( + api_key="k", server_url="https://x/receive", polling_interval=300 + ).validation_errors() == [] + + +def test_empty_string_env_becomes_none(): + # config.yaml ${VAR} expansion yields "" for unset vars. + s = ComplianceSettings(api_key="", tenant_id="") + assert s.api_key is None + assert s.tenant_id is None + + +# ── payload signing ───────────────────────────────────────────────────────── + +def _cfg(**kw): + base = dict(api_key="secret-key", tenant_id="t1", router_id="r1", + server_url="https://x/api/v1/discovery/receive") + base.update(kw) + return ComplianceSettings(**base) + + +def test_signature_deterministic_and_key_order_independent(): + cfg = _cfg() + snap = {"timestamp": "2026-01-01T00:00:00Z", "models": [{"b": 1, "a": 2}], + "endpoints": [], "telemetry": {"x": 1}} + p1 = payload_mod.build_payload(cfg, snap) + p2 = payload_mod.build_payload(cfg, snap) + assert p1["hmac_signature"] == p2["hmac_signature"] + assert p1["hmac_signature"].startswith("sha256=") + + +def test_signature_detects_tampering(): + cfg = _cfg() + snap = {"timestamp": "t", "models": [], "endpoints": [], "telemetry": {}} + p = payload_mod.build_payload(cfg, snap) + sig = p.pop("hmac_signature") + assert payload_mod.sign(p, cfg) == sig # untouched verifies + p["router_id"] = "evil" + assert payload_mod.sign(p, cfg) != sig # tampered does not + + +def test_hmac_secret_overrides_api_key(): + snap = {"timestamp": "t", "models": [], "endpoints": [], "telemetry": {}} + a = payload_mod.build_payload(_cfg(), snap) + b = payload_mod.build_payload(_cfg(hmac_secret="other"), snap) + assert a["hmac_signature"] != b["hmac_signature"] + + +def test_split_into_batches_resigns(): + cfg = _cfg() + snap = {"timestamp": "t", "models": [{"name": f"m{i}"} for i in range(5)], + "endpoints": [], "telemetry": {}} + p = payload_mod.build_payload(cfg, snap) + batches = payload_mod.split_into_batches(p, batch_size=2, cfg=cfg) + assert len(batches) == 3 + assert [len(b["models"]) for b in batches] == [2, 2, 1] + for b in batches: + sig = b.pop("hmac_signature") + assert payload_mod.sign(b, cfg) == sig + + +# ── buffer crypto + store ─────────────────────────────────────────────────── + +def test_crypto_roundtrip(): + ct = crypto.encrypt(b"hello", "key-1") + assert ct != b"hello" + assert crypto.decrypt(ct, "key-1") == b"hello" + assert crypto.decrypt(ct, "wrong-key") is None # wrong key -> None, no raise + + +def test_buffer_save_load_roundtrip(tmp_path): + cfg = _cfg(buffer_dir=str(tmp_path / "buf")) + store = BufferStore(cfg) + body = {"router_id": "r1", "models": [{"name": "x"}]} + path = store.save(body, timestamp_ms=1000) + assert path is not None and path.exists() + # raw file must not contain plaintext + assert b"router_id" not in path.read_bytes() + assert store.count() == 1 + assert store.load(path) == body + + +def test_buffer_replay_order_and_bound(tmp_path): + cfg = _cfg(buffer_dir=str(tmp_path / "buf"), max_buffer_payloads=3) + store = BufferStore(cfg) + for ts in (5000, 1000, 3000): + store.save({"ts": ts}, timestamp_ms=ts) + pending = store.pending() + loaded_ts = [store.load(p)["ts"] for p in pending] + assert loaded_ts == [1000, 3000, 5000] # oldest-first + + # exceeding the bound drops the oldest + for ts in (7000, 9000): + store.save({"ts": ts}, timestamp_ms=ts) + remaining = [store.load(p)["ts"] for p in store.pending()] + assert 1000 not in remaining + assert len(remaining) == 3 + + +def test_buffer_drops_corrupt_file(tmp_path): + cfg = _cfg(buffer_dir=str(tmp_path / "buf")) + store = BufferStore(cfg) + p = store.save({"a": 1}, timestamp_ms=1) + p.write_bytes(b"not-a-valid-fernet-token") + assert store.load(p) is None + assert not p.exists() # corrupt file removed + + +# ── transport classification ──────────────────────────────────────────────── + +@pytest.mark.parametrize("status,expected", [ + (202, Outcome.OK), + (200, Outcome.OK), + (401, Outcome.AUTH_FAILED), + (403, Outcome.AUTH_FAILED), + (429, Outcome.RATE_LIMITED), + (413, Outcome.TOO_LARGE), + (500, Outcome.RETRY), + (502, Outcome.RETRY), +]) +def test_status_classification(status, expected): + assert _classify(status) == expected + + +# ── collector field mapping ───────────────────────────────────────────────── + +async def test_collector_maps_endpoints_and_models(): + from compliance import collector + + cfg = SimpleNamespace( + endpoints=["http://ollama:11434"], + llama_server_endpoints=["https://llama:8000/v1"], + api_keys={"https://llama:8000/v1": "k"}, + endpoint_config={}, + max_concurrent_connections=4, + ) + + async def fake_health(ep, timeout=5): + return {"status": "ok"} + + async def fake_available(ep, api_key=None): + if "ollama" in ep: + return {"mistral:7b"} + return {"org/gpt-oss-20b:Q4_K_M"} + + async def fake_details(ep, route, detail): + return [{"name": "mistral:7b", "size": 4_100_000_000, + "details": {"quantization_level": "Q4_0"}}] + + fake_db = SimpleNamespace( + get_last_used_map=lambda: _async_return({}), + get_token_totals_since=lambda ts: _async_return((10, 20)), + ) + + with patch.object(collector, "_endpoint_health", fake_health), \ + patch.object(collector.fetch, "available_models", fake_available), \ + patch.object(collector.fetch, "endpoint_details", fake_details), \ + patch.object(collector, "get_db", lambda: fake_db), \ + patch.object(collector, "get_llm_cache", lambda: None), \ + patch.dict(collector.usage_counts, {"http://ollama:11434": {"mistral:7b": 2}}, clear=True), \ + patch.object(collector, "get_max_connections", lambda ep: 4): + snap = await collector.collect_snapshot(cfg, started_at=0.0) + + eps = {e["url"]: e for e in snap["endpoints"]} + assert eps["http://ollama:11434"]["type"] == "ollama" + assert eps["http://ollama:11434"]["tls_enabled"] is False + assert eps["http://ollama:11434"]["concurrent_connections"] == 2 + assert eps["https://llama:8000/v1"]["type"] == "vllm" + assert eps["https://llama:8000/v1"]["tls_enabled"] is True + assert eps["https://llama:8000/v1"]["auth_method"] == "bearer" + + models = {m["name"]: m for m in snap["models"]} + assert models["mistral:7b"]["quantization"] == "Q4_0" + assert models["mistral:7b"]["size_gb"] == pytest.approx(4.1, rel=1e-3) + assert models["gpt-oss-20b"]["quantization"] == "Q4_K_M" + # un-instrumented fields are null, not fabricated + assert models["mistral:7b"]["avg_latency_ms"] is None + assert models["mistral:7b"]["request_count_24h"] is None + + tel = snap["telemetry"] + assert tel["active_connections"] == 2 + assert tel["total_bytes_in_24h"] == 10 + assert tel["total_requests_24h"] is None + + +async def _async_return(value): + return value