alexei.dolgolyov
7cbb02b1ef
Take a consistent, atomic copy of the DB at lifespan startup BEFORE migrations run, so a botched future upgrade is recoverable by restoring a single file instead of a data-loss incident. Uses SQLite's VACUUM INTO — safe under WAL, cannot tear against concurrent writes. Best-effort: failures are logged, never raised — the main DB remains the source of truth. Configurable via NOTIFY_BRIDGE_PRE_MIGRATE_SNAPSHOT_KEEP (default 5; 0 disables). Snapshots land in ``data_dir/backups/pre-migrate-<ts>.db`` and the N oldest are pruned each boot.
123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
"""Server configuration from environment variables."""
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
from pydantic_settings import BaseSettings
|
|
|
|
# Secret keys we will actively refuse. These cover the default template value
|
|
# and dev-only literals that have appeared in scripts or documentation.
|
|
_FORBIDDEN_SECRETS: frozenset[str] = frozenset(
|
|
{
|
|
"change-me-in-production",
|
|
"test-secret-key-minimum-32-chars",
|
|
"dev-secret-key-not-for-production",
|
|
}
|
|
)
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""Application settings loaded from environment variables."""
|
|
|
|
data_dir: Path = Path("/data")
|
|
database_url: str = ""
|
|
|
|
secret_key: str = "change-me-in-production"
|
|
|
|
access_token_expire_minutes: int = 15
|
|
refresh_token_expire_days: int = 30
|
|
|
|
jwt_issuer: str = "notify-bridge"
|
|
jwt_audience: str = "notify-bridge-api"
|
|
|
|
host: str = "0.0.0.0"
|
|
port: int = 8420
|
|
debug: bool = False
|
|
|
|
# Comma-separated list of trusted proxy IPs uvicorn will honor for
|
|
# X-Forwarded-For / X-Forwarded-Proto. Use "*" ONLY when you trust the
|
|
# network (never directly on the internet). Default matches uvicorn.
|
|
forwarded_allow_ips: str = "127.0.0.1"
|
|
|
|
# How long to wait for in-flight requests / scheduler jobs before force
|
|
# killing on SIGTERM.
|
|
graceful_shutdown_seconds: int = 60
|
|
|
|
anthropic_api_key: str = ""
|
|
ai_model: str = "claude-sonnet-4-20250514"
|
|
ai_max_tokens: int = 1024
|
|
|
|
telegram_webhook_secret: str = ""
|
|
|
|
cors_allowed_origins: str = "http://localhost:5175"
|
|
"""Comma-separated allowed origins for CORS (e.g. 'http://localhost:5173,https://myapp.com')."""
|
|
|
|
static_dir: str = ""
|
|
"""Path to frontend static files. Set to serve SvelteKit build via FastAPI (e.g. /app/static in Docker)."""
|
|
|
|
# --- Logging ---
|
|
log_level: str = "INFO"
|
|
"""Root log level for the app loggers (``DEBUG``/``INFO``/``WARNING``/``ERROR``)."""
|
|
|
|
log_format: str = "text"
|
|
"""Log output format: ``text`` (human-readable) or ``json`` (one object per line)."""
|
|
|
|
log_levels: str = ""
|
|
"""Comma-separated per-module overrides, e.g. ``notify_bridge_core.notifications.telegram.client=DEBUG,sqlalchemy.engine=INFO``."""
|
|
|
|
# --- Retention ---
|
|
event_log_retention_days: int = 30
|
|
"""Days of event_log history to retain. 0 disables the retention job."""
|
|
|
|
pre_migrate_snapshot_keep: int = 5
|
|
"""Number of pre-migration DB snapshots to keep in ``data_dir/backups/``.
|
|
0 disables snapshotting entirely. Each snapshot is produced at boot
|
|
before migrations run using SQLite's ``VACUUM INTO`` (atomic, consistent).
|
|
"""
|
|
|
|
model_config = {"env_prefix": "NOTIFY_BRIDGE_"}
|
|
|
|
def model_post_init(self, __context: Any) -> None:
|
|
if self.secret_key in _FORBIDDEN_SECRETS:
|
|
raise ValueError(
|
|
"SECURITY: Refusing to start with a known/default secret_key. "
|
|
"Set NOTIFY_BRIDGE_SECRET_KEY to a random value (>=32 bytes) "
|
|
"before starting the server."
|
|
)
|
|
if len(self.secret_key) < 32:
|
|
raise ValueError(
|
|
"SECURITY: NOTIFY_BRIDGE_SECRET_KEY must be at least 32 characters."
|
|
)
|
|
origins = [o.strip() for o in self.cors_allowed_origins.split(",") if o.strip()]
|
|
if "*" in origins:
|
|
raise ValueError(
|
|
"SECURITY: wildcard '*' is not allowed in CORS origins when credentials are enabled."
|
|
)
|
|
for origin in origins:
|
|
parsed = urlparse(origin)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise ValueError(
|
|
f"CORS origin {origin!r} is invalid — must include scheme (http/https) and host."
|
|
)
|
|
if self.access_token_expire_minutes <= 0:
|
|
raise ValueError("access_token_expire_minutes must be > 0")
|
|
if self.refresh_token_expire_days <= 0:
|
|
raise ValueError("refresh_token_expire_days must be > 0")
|
|
if not (1 <= self.port <= 65535):
|
|
raise ValueError("port must be in range 1..65535")
|
|
if self.event_log_retention_days < 0:
|
|
raise ValueError("event_log_retention_days must be >= 0")
|
|
if self.pre_migrate_snapshot_keep < 0:
|
|
raise ValueError("pre_migrate_snapshot_keep must be >= 0")
|
|
|
|
@property
|
|
def effective_database_url(self) -> str:
|
|
if self.database_url:
|
|
return self.database_url
|
|
db_path = self.data_dir / "notify_bridge.db"
|
|
return f"sqlite+aiosqlite:///{db_path}"
|
|
|
|
|
|
settings = Settings()
|