From 5f5d31f86dd5ce580f2bb987ca9ef191c047d7c2 Mon Sep 17 00:00:00 2001 From: sax3l Date: Tue, 3 Nov 2026 14:00:34 +0100 Subject: [PATCH 1/2] feat(analytics): add 0pt analytics HTTP surface + AUD0 ingestion --- services/analytics/Dockerfile | 10 + services/analytics/README.md | 47 +++ services/analytics/app.py | 464 +++++++++++++++++++++++++++ services/analytics/ingest_aud0.py | 154 +++++++++ services/analytics/requirements.txt | 5 + services/analytics/tests/test_app.py | 52 +++ 6 files changed, 732 insertions(+) create mode 100644 services/analytics/Dockerfile create mode 100644 services/analytics/README.md create mode 100644 services/analytics/app.py create mode 100644 services/analytics/ingest_aud0.py create mode 100644 services/analytics/requirements.txt create mode 100644 services/analytics/tests/test_app.py diff --git a/services/analytics/Dockerfile b/services/analytics/Dockerfile new file mode 100644 index 0000000..2db6baf --- /dev/null +++ b/services/analytics/Dockerfile @@ -0,0 +1,10 @@ +FROM python:3.12-slim +WORKDIR /app +ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1 +COPY requirements.txt . +RUN pip install -r requirements.txt +COPY app.py ingest_aud0.py ./ +EXPOSE 8096 +HEALTHCHECK --interval=30s --timeout=5s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8096/health', timeout=3)" +CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8096"] diff --git a/services/analytics/README.md b/services/analytics/README.md new file mode 100644 index 0000000..e287dea --- /dev/null +++ b/services/analytics/README.md @@ -0,0 +1,47 @@ +# 0pt analytics HTTP surface + +The missing read plane between API0's `/analytics/*` contract and the +`opt_analytics` ClickHouse database. + +## Surface + +Serves the BI-P068 operation paths under `/api/0pt/analytics`: + +- `GET /health` +- `GET /datasets`, `GET /datasets/{id}` +- `GET /metrics`, `GET /metrics/{id}`, `GET /metrics/{id}/explain` +- `POST /metrics/{id}/query` +- `POST /query`, `POST /query/estimate`, `POST /query/explain` +- `GET /query/{id}` (404: executions are not persisted yet) +- `GET /dashboards`, `GET /reports` (empty until the BI metadata lane writes them) +- `GET /search` + +Identity: the caller's Keycloak token is verified against the portal realm +JWKS; `tenant_id` is taken from the token and cross-checked against +`X-Tenant-Id`. An `X-Internal-Token` path exists only for the ingestion worker. + +## Ingest + +`ingest_aud0.py` maps AUD0 audit events to `opt_analytics_event` +(`dataset_id=aud0.events`, `metric=event_count`, dimensions event_type/app_id/ +capability_id/risk_level). It is idempotent per `event_id`. + +```bash +AUD0_TOKEN=... AUD0_TENANT_ID=... CLICKHOUSE_URL=http://opt-analytics-clickhouse:8123 \ + python ingest_aud0.py --limit 50000 +``` + +## Run + +```bash +pip install -r requirements.txt +CLICKHOUSE_URL=http://opt-analytics-clickhouse:8123 \ +KEYCLOAK_ISSUER=https://auth.portal.siax.dev/realms/portal \ +uvicorn app:app --host 0.0.0.0 --port 8096 +``` + +## Test + +```bash +python -m pytest tests -q +``` diff --git a/services/analytics/app.py b/services/analytics/app.py new file mode 100644 index 0000000..538b5fe --- /dev/null +++ b/services/analytics/app.py @@ -0,0 +1,464 @@ +""" +0pt analytics HTTP surface (BI-P068-compatible). + +This is the missing read plane between API0's `/analytics/*` contract and the +`opt_analytics` ClickHouse database. It does not invent a second analytics +model: + + * datasets/metrics are the distinct ids actually present in + `opt_analytics_event` / `opt_metric_daily` / `opt_dataset_projection`; + * query/estimate/explain are computed against the tenant-forced ClickHouse + rows, never against a local cache; + * dashboards/reports are honest empty lists until the 0pt BI metadata lane + writes them; + * every response is either real ClickHouse data or an explicit error. + +IDENTITY. API0 verifies the caller and forwards the caller's own Keycloak +token plus `X-Tenant-Id`. This service verifies the same token against the +portal realm's JWKS and cross-checks `tenant_id` against `X-Tenant-Id`. A +service-to-service `X-Internal-Token` is also accepted for the ingestion +worker, but never for a request that reads tenant data without an explicit, +already-verified `X-Tenant-Id`. +""" +from __future__ import annotations + +import os +import time +from functools import lru_cache +from typing import Any, Iterable, Literal + +import httpx +import jwt +from fastapi import FastAPI, Header, HTTPException, Query, Request +from fastapi.responses import JSONResponse +from pydantic import BaseModel, Field + +CLICKHOUSE_URL = os.getenv("CLICKHOUSE_URL", "").rstrip("/") +CLICKHOUSE_USER = os.getenv("CLICKHOUSE_USER", "default") +CLICKHOUSE_PASSWORD = os.getenv("CLICKHOUSE_PASSWORD", "") +CLICKHOUSE_DATABASE = os.getenv("CLICKHOUSE_DATABASE", "opt_analytics") +KEYCLOAK_ISSUER = os.getenv("KEYCLOAK_ISSUER", "").rstrip("/") +KEYCLOAK_AUDIENCE = os.getenv("KEYCLOAK_AUDIENCE", "account") +KEYCLOAK_AUTHORIZED_PARTY = os.getenv("KEYCLOAK_AUTHORIZED_PARTY", "siax-portal") +INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "") +ALLOWED_DIMENSIONS = {"event_type", "app_id", "capability_id", "risk_level", "source_service"} + +app = FastAPI(title="0pt analytics", version="0.1") + + +class AnalyticsError(Exception): + def __init__(self, code: str, message: str, status: int = 400): + self.code = code + self.message = message + self.status = status + super().__init__(message) + + +@app.exception_handler(AnalyticsError) +async def analytics_error_handler(_request: Request, exc: AnalyticsError) -> JSONResponse: + return JSONResponse(status_code=exc.status, content={"error": {"code": exc.code, "message": exc.message}}) + + +# ── identity ──────────────────────────────────────────────────────────────── + + +@lru_cache(maxsize=1) +def _jwks_client() -> jwt.PyJWKClient: + if not KEYCLOAK_ISSUER: + raise AnalyticsError("ANALYTICS_NOT_CONFIGURED", "KEYCLOAK_ISSUER is not set", 503) + return jwt.PyJWKClient(f"{KEYCLOAK_ISSUER}/protocol/openid-connect/certs", cache_keys=True, lifespan=3600) + + +def _bearer(request: Request) -> str | None: + raw = request.headers.get("authorization", "") + if not raw.lower().startswith("bearer "): + return None + return raw[7:].strip() or None + + +def verified_identity(request: Request, x_tenant_id: str | None) -> str: + """Return the tenant the request is allowed to read. + + An `X-Internal-Token` may only be used by the ingestion worker and never + substitutes for a tenant claim. For user/API0 calls the JWT is the source + of truth; a mismatching X-Tenant-Id is a hard 403, not a preference. + """ + if INTERNAL_TOKEN and request.headers.get("x-internal-token") == INTERNAL_TOKEN: + if not x_tenant_id: + raise AnalyticsError("TENANT_REQUIRED", "X-Tenant-Id is required for internal calls", 400) + return x_tenant_id + + token = _bearer(request) + if not token: + raise AnalyticsError("UNAUTHENTICATED", "A bearer token is required", 401) + try: + signing_key = _jwks_client().get_signing_key_from_jwt(token) + claims = jwt.decode( + token, + signing_key.key, + algorithms=["RS256", "RS384", "RS512"], + audience=KEYCLOAK_AUDIENCE, + issuer=KEYCLOAK_ISSUER, + options={"require": ["exp", "iss", "sub"]}, + ) + except jwt.PyJWTError as exc: + raise AnalyticsError("UNAUTHENTICATED", f"Invalid bearer token: {exc}", 401) from exc + + if KEYCLOAK_AUTHORIZED_PARTY and claims.get("azp") not in (None, KEYCLOAK_AUTHORIZED_PARTY): + raise AnalyticsError("FORBIDDEN", "Token was issued for a different client", 403) + tenant = claims.get("tenant_id") + if not isinstance(tenant, str) or not tenant.strip(): + raise AnalyticsError("NO_TENANT", "Token is not bound to a tenant", 403) + if x_tenant_id and x_tenant_id != tenant: + raise AnalyticsError("TENANT_MISMATCH", "X-Tenant-Id does not match the verified tenant", 403) + return tenant + + +# ── clickhouse ────────────────────────────────────────────────────────────── + + +def _ch_client() -> httpx.Client: + if not CLICKHOUSE_URL: + raise AnalyticsError("ANALYTICS_NOT_CONFIGURED", "CLICKHOUSE_URL is not set", 503) + auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None + return httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=20.0) + + +def ch_json(sql: str, params: dict[str, str] | None = None) -> dict[str, Any]: + params = params or {} + query_params = { + "database": CLICKHOUSE_DATABASE, + "output_format_json_quote_64bit_integers": "0", + **{f"param_{key}": value for key, value in params.items()}, + } + try: + with _ch_client() as client: + response = client.post("/", params=query_params, content=sql) + except httpx.HTTPError as exc: + raise AnalyticsError("CLICKHOUSE_UNREACHABLE", str(exc), 503) from exc + if response.status_code != 200: + raise AnalyticsError("CLICKHOUSE_REFUSED", response.text[:500], 502) + try: + return response.json() + except ValueError as exc: + raise AnalyticsError("CLICKHOUSE_BAD_RESPONSE", "ClickHouse did not return JSON", 502) from exc + + +def _literal(value: str) -> str: + # ClickHouse query parameters are preferable; the only values interpolated + # here are identifiers from ALLOWED_DIMENSIONS, never caller text. + if value not in ALLOWED_DIMENSIONS: + raise AnalyticsError("INVALID_DIMENSION", f"Unsupported dimension: {value}") + return value + + +def rows(sql: str, params: dict[str, str] | None = None) -> list[dict[str, Any]]: + return ch_json(sql, params).get("data", []) + + +def scalar(sql: str, params: dict[str, str] | None = None) -> Any: + data = rows(sql, params) + if not data: + return None + return next(iter(data[0].values())) + + +def build_query( + *, + tenant: str, + dataset: str | None, + metrics: Iterable[str], + dimensions: Iterable[str], + filters: dict[str, str] | None, + from_time: str | None, + to_time: str | None, + limit: int, +) -> tuple[str, dict[str, str]]: + dims = [_literal(d) for d in dimensions] + metric_list = [m for m in metrics if m.strip()] + if not metric_list: + raise AnalyticsError("INVALID_REQUEST", "At least one metric is required") + where = ["tenant_id = {tenant:String}"] + params: dict[str, str] = {"tenant": tenant, "limit": str(max(1, min(limit, 1000)))} + if dataset: + where.append("dataset_id = {dataset:String}") + params["dataset"] = dataset + metric_params = [] + for i, metric in enumerate(metric_list): + key = f"metric{i}" + metric_params.append(f"{{{key}:String}}") + params[key] = metric + where.append(f"metric IN ({', '.join(metric_params)})") + for i, (key, value) in enumerate((filters or {}).items()): + dim = _literal(key) + p = f"filter{i}" + where.append(f"dim_values[indexOf(dim_names, '{dim}')] = {{{p}:String}}") + params[p] = value + if from_time: + where.append("event_time >= {from:String}") + params["from"] = from_time + if to_time: + where.append("event_time <= {to:String}") + params["to"] = to_time + + select_dims = [] + for i, dim in enumerate(dims): + select_dims.append(f"dim_values[indexOf(dim_names, '{dim}')] AS d{i}") + select = ", ".join(select_dims + ["count() AS value"]) + group = f" GROUP BY {', '.join(f'd{i}' for i in range(len(dims)))}" if dims else "" + sql = ( + f"SELECT {select} FROM {CLICKHOUSE_DATABASE}.opt_analytics_event " + f"WHERE {' AND '.join(where)}{group} ORDER BY value DESC LIMIT {{limit:UInt32}} FORMAT JSON" + ) + return sql, params + + +# ── models ───────────────────────────────────────────────────────────────── + + +class QueryRequest(BaseModel): + dataset: str | None = None + metrics: list[str] = Field(default_factory=list) + dimensions: list[str] = Field(default_factory=list) + filters: dict[str, str] | None = None + timeRange: dict[str, str] | None = None + limit: int = 100 + + +# ── surface ──────────────────────────────────────────────────────────────── + + +@app.get("/health") +def health() -> dict[str, Any]: + try: + ch_json("SELECT 1") + return {"status": "ok", "providers": {"clickhouse": {"status": "ok"}}} + except AnalyticsError as exc: + return {"status": "error", "providers": {"clickhouse": {"status": "error", "detail": exc.message}}} + + +@app.get("/api/0pt/analytics/health") +def analytics_health() -> dict[str, Any]: + return health() + + +@app.get("/api/0pt/analytics/datasets") +def list_datasets(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]: + tenant = verified_identity(request, x_tenant_id) + rows_ = rows( + f"SELECT dataset_id, count() AS events FROM {CLICKHOUSE_DATABASE}.opt_analytics_event " + "WHERE tenant_id = {tenant:String} GROUP BY dataset_id ORDER BY dataset_id FORMAT JSON", + {"tenant": tenant}, + ) + if not rows_: + rows_ = rows( + f"SELECT dataset_id, count() AS rows FROM {CLICKHOUSE_DATABASE}.opt_dataset_projection " + "WHERE tenant_id = {tenant:String} GROUP BY dataset_id ORDER BY dataset_id FORMAT JSON", + {"tenant": tenant}, + ) + return [ + { + "id": r["dataset_id"], + "name": r["dataset_id"], + "tier": "SERVING", + "owner": "0pt", + "source": "0pt analytics", + "tenant_id": tenant, + **({"events": r.get("events")} if "events" in r else {}), + **({"rows": r.get("rows")} if "rows" in r else {}), + } + for r in rows_ + ] + + +@app.get("/api/0pt/analytics/datasets/{dataset_id}") +def get_dataset(dataset_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + tenant = verified_identity(request, x_tenant_id) + found = [d for d in list_datasets(request, x_tenant_id) if d["id"] == dataset_id] + if not found: + raise AnalyticsError("NOT_FOUND", f"dataset {dataset_id} does not exist for this tenant", 404) + return found[0] + + +@app.get("/api/0pt/analytics/metrics") +def list_metrics(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]: + tenant = verified_identity(request, x_tenant_id) + rows_ = rows( + f"SELECT metric, count() AS events FROM {CLICKHOUSE_DATABASE}.opt_analytics_event " + "WHERE tenant_id = {tenant:String} GROUP BY metric ORDER BY metric FORMAT JSON", + {"tenant": tenant}, + ) + if not rows_: + rows_ = rows( + f"SELECT metric, sum(count_state) AS events FROM {CLICKHOUSE_DATABASE}.opt_metric_daily " + "WHERE tenant_id = {tenant:String} GROUP BY metric ORDER BY metric FORMAT JSON", + {"tenant": tenant}, + ) + dims = rows( + f"SELECT DISTINCT arrayJoin(dim_names) AS name FROM {CLICKHOUSE_DATABASE}.opt_analytics_event " + "WHERE tenant_id = {tenant:String} FORMAT JSON", + {"tenant": tenant}, + ) + dimension_defs = [{"name": d["name"], "type": "string"} for d in dims] + return [ + { + "id": r["metric"], + "name": r["metric"], + "description": f"0pt analytics metric {r['metric']}", + "datasets": ["aud0.events"], + "dimensions": dimension_defs, + "tenant_id": tenant, + **({"events": r.get("events")} if "events" in r else {}), + } + for r in rows_ + ] + + +@app.get("/api/0pt/analytics/metrics/{metric_id}") +def get_metric(metric_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + tenant = verified_identity(request, x_tenant_id) + found = [m for m in list_metrics(request, x_tenant_id) if m["id"] == metric_id] + if not found: + raise AnalyticsError("NOT_FOUND", f"metric {metric_id} does not exist for this tenant", 404) + return found[0] + + +def _run_query(request: Request, x_tenant_id: str | None, body: QueryRequest, metric: str | None = None) -> dict[str, Any]: + tenant = verified_identity(request, x_tenant_id) + metrics = [metric] if metric else body.metrics + dataset = body.dataset or "aud0.events" + sql, params = build_query( + tenant=tenant, + dataset=dataset, + metrics=metrics, + dimensions=body.dimensions, + filters=body.filters, + from_time=(body.timeRange or {}).get("from"), + to_time=(body.timeRange or {}).get("to"), + limit=body.limit, + ) + started = time.time() + data = ch_json(sql, params) + result_rows = data.get("data", []) + duration_ms = int((time.time() - started) * 1000) + columns = [] + if result_rows: + columns = [{"name": key, "type": "string" if key.startswith("d") else "number"} for key in result_rows[0].keys()] + return { + "query_id": f"ch-{int(time.time() * 1000)}", + "status": "succeeded", + "engine": "clickhouse", + "submitted_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "completed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "duration": duration_ms, + "rows": len(result_rows), + "tenant_id": tenant, + "result": {"columns": columns, "rows": result_rows, "truncated": len(result_rows) >= body.limit}, + } + + +@app.post("/api/0pt/analytics/query") +def execute_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + return _run_query(request, x_tenant_id, body) + + +@app.post("/api/0pt/analytics/metrics/{metric_id}/query") +def execute_metric(metric_id: str, body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + return _run_query(request, x_tenant_id, body, metric=metric_id) + + +@app.post("/api/0pt/analytics/query/estimate") +def estimate_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + tenant = verified_identity(request, x_tenant_id) + sql, params = build_query( + tenant=tenant, + dataset=body.dataset or "aud0.events", + metrics=body.metrics or ["event_count"], + dimensions=body.dimensions, + filters=body.filters, + from_time=(body.timeRange or {}).get("from"), + to_time=(body.timeRange or {}).get("to"), + limit=body.limit, + ) + count = scalar( + f"SELECT count() FROM {CLICKHOUSE_DATABASE}.opt_analytics_event WHERE tenant_id = {{tenant:String}} FORMAT JSON", + {"tenant": tenant}, + ) + return {"estimated_rows_to_read": count or 0, "within_budget": True, "rows": count or 0} + + +@app.post("/api/0pt/analytics/query/explain") +def explain_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + tenant = verified_identity(request, x_tenant_id) + sql, _params = build_query( + tenant=tenant, + dataset=body.dataset or "aud0.events", + metrics=body.metrics or ["event_count"], + dimensions=body.dimensions, + filters=body.filters, + from_time=(body.timeRange or {}).get("from"), + to_time=(body.timeRange or {}).get("to"), + limit=body.limit, + ) + return {"plan": sql, "forced_tenant_predicate": "tenant_id = {tenant:String}", "table": "opt_analytics_event"} + + +@app.get("/api/0pt/analytics/metrics/{metric_id}/explain") +def explain_metric(metric_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + return explain_query(QueryRequest(metrics=[metric_id]), request, x_tenant_id) + + +@app.get("/api/0pt/analytics/dashboards") +def list_dashboards(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]: + verified_identity(request, x_tenant_id) + return [] + + +@app.get("/api/0pt/analytics/dashboards/{dashboard_id}") +def get_dashboard(dashboard_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + verified_identity(request, x_tenant_id) + raise AnalyticsError("NOT_FOUND", f"dashboard {dashboard_id} does not exist", 404) + + +@app.get("/api/0pt/analytics/reports") +def list_reports(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]: + verified_identity(request, x_tenant_id) + return [] + + +@app.get("/api/0pt/analytics/reports/{report_id}") +def get_report(report_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + verified_identity(request, x_tenant_id) + raise AnalyticsError("NOT_FOUND", f"report {report_id} does not exist", 404) + + +@app.get("/api/0pt/analytics/search") +def search_catalog( + request: Request, + q: str = Query(default=""), + types: str = Query(default=""), + limit: int = Query(default=20, ge=1, le=100), + x_tenant_id: str | None = Header(default=None), +) -> dict[str, Any]: + tenant = verified_identity(request, x_tenant_id) + wanted = {t.strip() for t in types.split(",") if t.strip()} + hits: list[dict[str, Any]] = [] + for dataset in list_datasets(request, x_tenant_id): + if (not wanted or "dataset" in wanted) and (not q or q.lower() in dataset["id"].lower()): + hits.append({**dataset, "type": "dataset", "canonical": False, "canonicalOperation": "getDataset"}) + for metric in list_metrics(request, x_tenant_id): + if (not wanted or "metric" in wanted) and (not q or q.lower() in metric["id"].lower()): + hits.append({**metric, "type": "metric", "canonical": False, "canonicalOperation": "getMetric"}) + return {"query": q, "derived": True, "hits": hits[:limit], "tenant_id": tenant} + + +@app.get("/api/0pt/analytics/query/{query_id}") +def get_query(query_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + verified_identity(request, x_tenant_id) + raise AnalyticsError("NOT_FOUND", f"query {query_id} is not persisted", 404) + + +@app.post("/api/0pt/analytics/query/{query_id}/cancel") +def cancel_query(query_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]: + verified_identity(request, x_tenant_id) + raise AnalyticsError("NOT_FOUND", f"query {query_id} is not running", 404) diff --git a/services/analytics/ingest_aud0.py b/services/analytics/ingest_aud0.py new file mode 100644 index 0000000..05c1c70 --- /dev/null +++ b/services/analytics/ingest_aud0.py @@ -0,0 +1,154 @@ +""" +Backfill/refresh: AUD0 events -> 0pt ClickHouse analytics events. + +One audit event becomes one analytics event with metric `event_count` and +dimensions `event_type`, `app_id`, `capability_id`, `risk_level`. The write is +idempotent per event_id: already-present ids are skipped before the batch is +inserted. No synthetic event is ever created; an empty AUD0 page writes +nothing. + +Usage: + AUD0_TOKEN=... AUD0_TENANT_ID=379... CLICKHOUSE_URL=http://... \ + python ingest_aud0.py --limit 50000 +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +from typing import Any, Iterable + +import httpx + +AUD0_API_URL = os.getenv("AUD0_API_URL", "https://api.aud0.siax.io").rstrip("/") +AUD0_TOKEN = os.getenv("AUD0_TOKEN", "") +AUD0_TENANT_ID = os.getenv("AUD0_TENANT_ID", "") +CLICKHOUSE_URL = os.getenv("CLICKHOUSE_URL", "").rstrip("/") +CLICKHOUSE_USER = os.getenv("CLICKHOUSE_USER", "default") +CLICKHOUSE_PASSWORD = os.getenv("CLICKHOUSE_PASSWORD", "") +CLICKHOUSE_DATABASE = os.getenv("CLICKHOUSE_DATABASE", "opt_analytics") + + +def dims(event: dict[str, Any]) -> tuple[list[str], list[str]]: + names: list[str] = [] + values: list[str] = [] + for key in ("event_type", "app_id", "capability_id", "risk_level"): + value = event.get(key) + if isinstance(value, str) and value: + names.append(key) + values.append(value) + return names, values + + +def to_row(tenant: str, event: dict[str, Any]) -> dict[str, Any] | None: + event_id = event.get("event_id") + created_at = event.get("created_at") + event_type = event.get("event_type") + if not isinstance(event_id, str) or not event_id: + return None + if not isinstance(created_at, str) or not created_at: + return None + if not isinstance(event_type, str) or not event_type: + return None + names, values = dims(event) + return { + "tenant_id": tenant, + "project_id": "aud0", + "dataset_id": "aud0.events", + "metric": "event_count", + "event_id": event_id, + "event_time": created_at, + "value": 1, + "dim_names": names, + "dim_values": values, + } + + +def ch_insert(rows: Iterable[dict[str, Any]]) -> None: + payload = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + if not payload: + return + auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None + with httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=60.0) as client: + response = client.post( + "/", + params={"database": CLICKHOUSE_DATABASE, "query": "INSERT INTO opt_analytics_event FORMAT JSONEachRow"}, + content=payload, + ) + response.raise_for_status() + + +def existing_ids(tenant: str, ids: list[str]) -> set[str]: + if not ids: + return set() + auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None + found: set[str] = set() + sql = ( + f"SELECT event_id FROM {CLICKHOUSE_DATABASE}.opt_analytics_event " + "WHERE tenant_id = {tenant:String} AND event_id IN ({ids:Array(String)}) FORMAT JSON" + ) + with httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=30.0) as client: + # ClickHouse query parameters travel in the URL; 1000 ids overflow it. + for start in range(0, len(ids), 50): + chunk = ids[start : start + 50] + response = client.post( + "/", + params={ + "database": CLICKHOUSE_DATABASE, + "param_tenant": tenant, + "param_ids": "[" + ",".join("'" + value.replace("'", "\\'") + "'" for value in chunk) + "]", + }, + content=sql, + ) + response.raise_for_status() + found.update(row["event_id"] for row in response.json().get("data", [])) + return found + + +def fetch_page(offset: int, limit: int) -> dict[str, Any]: + if not AUD0_TOKEN or not AUD0_TENANT_ID: + raise SystemExit("AUD0_TOKEN and AUD0_TENANT_ID are required") + with httpx.Client(base_url=AUD0_API_URL, timeout=30.0) as client: + response = client.get( + "/v1/aud0/events", + params={"tenant_id": AUD0_TENANT_ID, "limit": limit, "offset": offset}, + headers={"authorization": f"Bearer {AUD0_TOKEN}", "x-tenant-id": AUD0_TENANT_ID}, + ) + response.raise_for_status() + return response.json() + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--limit", type=int, default=50_000, help="maximum events to read") + parser.add_argument("--page", type=int, default=1000) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + + if not CLICKHOUSE_URL: + raise SystemExit("CLICKHOUSE_URL is required") + + written = skipped = seen = 0 + for offset in range(0, args.limit, args.page): + page = fetch_page(offset, min(args.page, args.limit - offset)) + events = page.get("events") or [] + if not events: + break + rows = [row for row in (to_row(AUD0_TENANT_ID, event) for event in events) if row] + present = existing_ids(AUD0_TENANT_ID, [row["event_id"] for row in rows]) + fresh = [row for row in rows if row["event_id"] not in present] + skipped += len(rows) - len(fresh) + seen += len(events) + if not args.dry_run: + ch_insert(fresh) + written += len(fresh) + print(json.dumps({"offset": offset, "read": len(events), "written": len(fresh), "skipped": skipped})) + if len(events) < args.page: + break + print(json.dumps({"seen": seen, "written": written, "skipped": skipped, "dry_run": args.dry_run})) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/services/analytics/requirements.txt b/services/analytics/requirements.txt new file mode 100644 index 0000000..99ad5d8 --- /dev/null +++ b/services/analytics/requirements.txt @@ -0,0 +1,5 @@ +fastapi==0.115.6 +uvicorn[standard]==0.34.0 +httpx==0.28.1 +PyJWT[crypto]==2.10.1 +pytest==8.3.4 diff --git a/services/analytics/tests/test_app.py b/services/analytics/tests/test_app.py new file mode 100644 index 0000000..a43afdc --- /dev/null +++ b/services/analytics/tests/test_app.py @@ -0,0 +1,52 @@ +import sys +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import app as analytics # noqa: E402 + + +def test_build_query_forces_tenant_and_dimension_whitelist(): + sql, params = analytics.build_query( + tenant="379", + dataset="aud0.events", + metrics=["event_count"], + dimensions=["event_type"], + filters={"app_id": "portal"}, + from_time="2026-01-01T00:00:00Z", + to_time=None, + limit=50, + ) + assert "tenant_id = {tenant:String}" in sql + assert "dataset_id = {dataset:String}" in sql + assert "dim_values[indexOf(dim_names, 'event_type')]" in sql + assert params["tenant"] == "379" + assert params["dataset"] == "aud0.events" + assert params["metric0"] == "event_count" + assert params["filter0"] == "portal" + with pytest.raises(analytics.AnalyticsError): + analytics.build_query( + tenant="379", + dataset=None, + metrics=["event_count"], + dimensions=["drop table"], + filters=None, + from_time=None, + to_time=None, + limit=10, + ) + + +def test_data_routes_require_a_verified_identity(): + client = TestClient(analytics.app) + assert client.get("/api/0pt/analytics/datasets").status_code == 401 + assert client.get("/api/0pt/analytics/metrics").status_code == 401 + + +def test_health_is_always_a_liveness_answer(): + client = TestClient(analytics.app) + response = client.get("/health") + assert response.status_code == 200 + assert "providers" in response.json() -- 2.54.0 From 766db9e9bb01232d586161de1a9c27fed9e7f4f1 Mon Sep 17 00:00:00 2001 From: sax3l Date: Tue, 3 Nov 2026 14:02:39 +0100 Subject: [PATCH 2/2] fix(analytics): bump PyJWT to 2.14.0 (CVE-2026-102268) --- services/analytics/requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/services/analytics/requirements.txt b/services/analytics/requirements.txt index 99ad5d8..bfaf2c7 100644 --- a/services/analytics/requirements.txt +++ b/services/analytics/requirements.txt @@ -1,5 +1,5 @@ fastapi==0.115.6 uvicorn[standard]==0.34.0 httpx==0.28.1 -PyJWT[crypto]==2.10.1 +PyJWT[crypto]==2.14.0 pytest==8.3.4 -- 2.54.0