Release 1.7.0: Grafana catalog, ingress/IRM, tests

onguard24/grafana_sources.py

@@ -0,0 +1,84 @@
+"""Несколько инстансов Grafana: URL API, токен, секрет вебхука по slug (организация / стек)."""
+
+from __future__ import annotations
+
+import json
+import re
+from pydantic import BaseModel, Field, field_validator
+
+_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,62}$")
+
+
+class GrafanaSourceEntry(BaseModel):
+    """Один инстанс Grafana для API и/или приёма вебхуков."""
+
+    slug: str = Field(..., description="Идентификатор в URL: /ingress/grafana/{slug}")
+    api_url: str = Field(..., description="Базовый URL без завершающего слэша")
+    api_token: str = Field(default="", description="Service account token для HTTP API")
+    webhook_secret: str = Field(
+        default="",
+        description="Если задан — только этот секрет для вебхука этого slug; иначе см. GRAFANA_WEBHOOK_SECRET",
+    )
+
+    @field_validator("slug")
+    @classmethod
+    def slug_ok(cls, v: str) -> str:
+        s = v.strip().lower()
+        if not _SLUG_RE.match(s):
+            raise ValueError(
+                "slug: только a-z, цифры, - и _, длина 1–63, с буквы/цифры"
+            )
+        return s
+
+    @field_validator("api_url")
+    @classmethod
+    def strip_url(cls, v: str) -> str:
+        return v.rstrip("/")
+
+
+def parse_grafana_sources_json(raw: str) -> list[GrafanaSourceEntry]:
+    if not raw.strip():
+        return []
+    data = json.loads(raw)
+    if not isinstance(data, list):
+        raise ValueError("GRAFANA_SOURCES_JSON должен быть JSON-массивом объектов")
+    return [GrafanaSourceEntry.model_validate(x) for x in data]
+
+
+def iter_grafana_sources(settings: "Settings") -> list[GrafanaSourceEntry]:
+    """Список источников: из GRAFANA_SOURCES_JSON или один синтетический default из GRAFANA_URL."""
+    try:
+        parsed = parse_grafana_sources_json(settings.grafana_sources_json)
+    except (json.JSONDecodeError, ValueError):
+        parsed = []
+    if parsed:
+        return parsed
+    gu = settings.grafana_url.strip()
+    if not gu:
+        return []
+    return [
+        GrafanaSourceEntry(
+            slug="default",
+            api_url=gu.rstrip("/"),
+            api_token=settings.grafana_service_account_token.strip(),
+            webhook_secret="",
+        )
+    ]
+
+
+def sources_by_slug(settings: "Settings") -> dict[str, GrafanaSourceEntry]:
+    return {s.slug: s for s in iter_grafana_sources(settings)}
+
+
+def effective_webhook_secret(settings: "Settings", source: GrafanaSourceEntry | None) -> str:
+    """Пустая строка = проверка вебхука отключена (только для dev)."""
+    if source and source.webhook_secret.strip():
+        return source.webhook_secret.strip()
+    return settings.grafana_webhook_secret.strip()
+
+
+def webhook_authorized(settings: "Settings", source: GrafanaSourceEntry | None, header: str | None) -> bool:
+    need = effective_webhook_secret(settings, source)
+    if not need:
+        return True
+    return (header or "") == need