85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
|
"""Несколько инстансов 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
|