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
|