feat: Add API key authentication for backup tool

- Add get_api_key() and require_api_key() to auth.py
- Add generic key-value settings storage to SettingsStore
- Add test_api_key_auth to verify the implementation
- Uses secrets.compare_digest for timing-safe comparison
- Auto-generates API key on first use and stores in settings DB
This commit is contained in:
2026-05-11 21:25:46 +02:00
parent f912623677
commit 836f733a2e
31 changed files with 2014 additions and 1 deletions
+24 -1
View File
@@ -3,21 +3,25 @@
from __future__ import annotations
import logging
import secrets
from functools import lru_cache
from typing import Any
from urllib.parse import urljoin
import jwt
import requests
from fastapi import Request
from fastapi import Header, HTTPException, Request
from fastapi.responses import JSONResponse
from jwt import PyJWKClient
from jwt.exceptions import InvalidTokenError
from media_library_viewer_api.config import Settings, get_settings
from media_library_viewer_api.dependencies import get_settings_store
logger = logging.getLogger(__name__)
_API_KEY: str | None = None
EXEMPT_PATHS = {
"/api/health",
"/docs",
@@ -115,3 +119,22 @@ async def require_jwt_auth(request: Request, call_next):
request.state.jwt_claims = claims
request.state.jwt_subject = claims.get("sub") if isinstance(claims, dict) else None
return await call_next(request)
def get_api_key() -> str:
global _API_KEY
if _API_KEY is None:
store = get_settings_store()
settings = store.get_settings()
_API_KEY = settings.get("backup_api_key")
if not _API_KEY:
_API_KEY = secrets.token_urlsafe(32)
store.update_setting("backup_api_key", _API_KEY)
return _API_KEY
def require_api_key(authorization: str = Header("", alias="Authorization")) -> str:
expected = f"Bearer {get_api_key()}"
if not authorization or not secrets.compare_digest(authorization, expected):
raise HTTPException(status_code=401, detail="Invalid or missing API key")
return authorization
@@ -1205,6 +1205,69 @@ class SettingsStore:
)
return int(cur.rowcount or 0)
# ------------------------------------------------------------------
# Generic key-value settings
# ------------------------------------------------------------------
def get_settings(self) -> dict[str, Any]:
"""Return all generic settings as a dict."""
self.init_schema()
with self.connect() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS app_settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at INTEGER NOT NULL
)
"""
)
rows = conn.execute("SELECT key, value FROM app_settings").fetchall()
return {row["key"]: row["value"] for row in rows}
def get_setting(self, key: str, default: Any = None) -> Any:
"""Return a single setting value or default."""
self.init_schema()
with self.connect() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS app_settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at INTEGER NOT NULL
)
"""
)
row = conn.execute(
"SELECT value FROM app_settings WHERE key = ?", (key,)
).fetchone()
return row["value"] if row else default
def update_setting(self, key: str, value: str) -> None:
"""Set a generic key-value setting."""
self.init_schema()
now = int(time.time())
with self.connect() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS app_settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at INTEGER NOT NULL
)
"""
)
conn.execute(
"""
INSERT INTO app_settings (key, value, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(key) DO UPDATE SET
value = excluded.value,
updated_at = excluded.updated_at
""",
(key, value, now),
)
_store: SettingsStore | None = None