97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import tomllib
|
|
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
KNOWN_NOTIFIERS = ("fluxer",)
|
|
|
|
SOURCE_DEFAULTS: dict[str, dict[str, object]] = {
|
|
"pending_reports": {"enabled": True, "stale_after_hours": 48},
|
|
"pending_registrations": {"enabled": True},
|
|
"discovery_applications": {"enabled": True},
|
|
"enforcement_review": {"enabled": True},
|
|
}
|
|
|
|
|
|
class ConfigError(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SourceSettings:
|
|
enabled: bool
|
|
stale_after_hours: float | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Settings:
|
|
api_url: str
|
|
api_key: str
|
|
admin_ui_url: str
|
|
request_timeout: float
|
|
realert: timedelta | None
|
|
notify_cleared: bool
|
|
state_file: Path
|
|
notifiers: tuple[str, ...]
|
|
fluxer_webhook_url: str | None
|
|
sources: dict[str, SourceSettings]
|
|
|
|
|
|
def _required(name: str) -> str:
|
|
value = os.environ.get(name, "").strip()
|
|
if not value:
|
|
raise ConfigError(f"{name} is not set. Add it to .env (see .env.example).")
|
|
return value
|
|
|
|
|
|
def load_settings(config_path: Path) -> Settings:
|
|
load_dotenv(PROJECT_ROOT / ".env")
|
|
|
|
if not config_path.is_file():
|
|
raise ConfigError(f"config file not found: {config_path}. Copy config.example.toml to config.toml.")
|
|
with config_path.open("rb") as fh:
|
|
raw = tomllib.load(fh)
|
|
|
|
general = raw.get("general", {})
|
|
raw_sources = raw.get("sources", {})
|
|
|
|
notifiers = tuple(general.get("notifiers", ["fluxer"]))
|
|
for name in notifiers:
|
|
if name not in KNOWN_NOTIFIERS:
|
|
raise ConfigError(f"unknown notifier in config: {name!r}. Known: {', '.join(KNOWN_NOTIFIERS)}")
|
|
|
|
fluxer_webhook_url = None
|
|
if "fluxer" in notifiers:
|
|
fluxer_webhook_url = _required("FLUXER_WEBHOOK_URL")
|
|
|
|
realert_hours = float(general.get("realert_hours", 12))
|
|
|
|
sources: dict[str, SourceSettings] = {}
|
|
for name, defaults in SOURCE_DEFAULTS.items():
|
|
table = {**defaults, **raw_sources.get(name, {})}
|
|
sources[name] = SourceSettings(
|
|
enabled=bool(table["enabled"]),
|
|
stale_after_hours=(
|
|
float(table["stale_after_hours"]) if table.get("stale_after_hours") is not None else None
|
|
),
|
|
)
|
|
|
|
return Settings(
|
|
api_url=_required("FLUXER_API_URL").rstrip("/"),
|
|
api_key=_required("FLUXER_ADMIN_API_KEY"),
|
|
admin_ui_url=_required("ADMIN_UI_URL").rstrip("/"),
|
|
request_timeout=float(general.get("request_timeout", 15)),
|
|
realert=timedelta(hours=realert_hours) if realert_hours > 0 else None,
|
|
notify_cleared=bool(general.get("notify_cleared", True)),
|
|
state_file=PROJECT_ROOT / str(general.get("state_file", "state.db")),
|
|
notifiers=notifiers,
|
|
fluxer_webhook_url=fluxer_webhook_url,
|
|
sources=sources,
|
|
)
|