Upload files to "backend/app"

backend/app/auth.py

@@ -0,0 +1,42 @@
+"""Collector (machine) authentication.
+
+Dashboard *users* are authenticated by nginx in front of the app (basic auth today,
+oauth2-proxy + Microsoft Entra ID later), so nothing here deals with people.
+"""
+
+import hashlib
+import secrets
+import sqlite3
+
+from fastapi import Depends, Header, HTTPException
+
+from .db import get_db
+from .timeutil import to_db, utcnow
+
+KEY_PREFIX = "bdk_"
+
+
+def generate_key() -> str:
+    return KEY_PREFIX + secrets.token_urlsafe(32)
+
+
+def hash_key(key: str) -> str:
+    # Keys are 256-bit random, so a fast unsalted hash is fine (no dictionary to attack).
+    return hashlib.sha256(key.encode()).hexdigest()
+
+
+def require_ingest_key(
+    authorization: str | None = Header(None),
+    conn: sqlite3.Connection = Depends(get_db),
+) -> str:
+    scheme, _, token = (authorization or "").partition(" ")
+    token = token.strip()
+    if scheme.lower() != "bearer" or not token:
+        raise HTTPException(401, "Missing bearer token", headers={"WWW-Authenticate": "Bearer"})
+    row = conn.execute(
+        "SELECT id, name FROM api_keys WHERE key_hash = ? AND revoked_at IS NULL", (hash_key(token),)
+    ).fetchone()
+    if row is None:
+        raise HTTPException(401, "Invalid or revoked API key", headers={"WWW-Authenticate": "Bearer"})
+    conn.execute("UPDATE api_keys SET last_used_at = ? WHERE id = ?", (to_db(utcnow()), row["id"]))
+    return row["name"]

backend/app/db.py

@@ -0,0 +1,99 @@
+import sqlite3
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+
+from fastapi import Request
+
+SCHEMA_VERSION = 1
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS tenants (
+    id   INTEGER PRIMARY KEY,
+    name TEXT NOT NULL UNIQUE COLLATE NOCASE
+);
+
+-- A job is identified by (tenant, source system, the source's own job id), so two
+-- products can't collide and renaming a job in Veeam doesn't create a duplicate.
+CREATE TABLE IF NOT EXISTS jobs (
+    id            INTEGER PRIMARY KEY,
+    tenant_id     INTEGER NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
+    source_system TEXT NOT NULL,
+    external_id   TEXT NOT NULL,
+    name          TEXT NOT NULL,
+    next_run_at   TEXT,
+    created_at    TEXT NOT NULL,
+    updated_at    TEXT NOT NULL,
+    UNIQUE (tenant_id, source_system, external_id)
+);
+
+CREATE TABLE IF NOT EXISTS runs (
+    id          INTEGER PRIMARY KEY,
+    job_id      INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+    external_id TEXT NOT NULL,
+    status      TEXT NOT NULL CHECK (status IN ('running', 'success', 'warning', 'failed')),
+    started_at  TEXT NOT NULL,
+    finished_at TEXT,
+    message     TEXT NOT NULL DEFAULT '',
+    size_bytes  INTEGER,
+    details     TEXT NOT NULL DEFAULT '{}',
+    received_at TEXT NOT NULL,
+    CHECK ((status = 'running') = (finished_at IS NULL)),
+    UNIQUE (job_id, external_id)
+);
+CREATE INDEX IF NOT EXISTS runs_job_started  ON runs (job_id, started_at);
+CREATE INDEX IF NOT EXISTS runs_job_finished ON runs (job_id, finished_at);
+
+-- Keys used by collectors to push results. Only the SHA-256 is stored.
+CREATE TABLE IF NOT EXISTS api_keys (
+    id           INTEGER PRIMARY KEY,
+    name         TEXT NOT NULL UNIQUE,
+    key_hash     TEXT NOT NULL UNIQUE,
+    created_at   TEXT NOT NULL,
+    last_used_at TEXT,
+    revoked_at   TEXT
+);
+"""
+
+
+def connect(path: Path | str) -> sqlite3.Connection:
+    # Autocommit mode; writes that must be atomic go through transaction().
+    # check_same_thread=False: FastAPI may run a dependency and its endpoint on different
+    # threadpool threads, but a connection is only ever used by one request at a time.
+    conn = sqlite3.connect(path, timeout=10, isolation_level=None, check_same_thread=False)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA foreign_keys = ON")
+    conn.execute("PRAGMA busy_timeout = 10000")
+    return conn
+
+
+def init_db(path: Path | str) -> None:
+    conn = connect(path)
+    try:
+        conn.execute("PRAGMA journal_mode = WAL")
+        version = conn.execute("PRAGMA user_version").fetchone()[0]
+        if version > SCHEMA_VERSION:
+            raise RuntimeError(f"Database schema v{version} is newer than this app (v{SCHEMA_VERSION})")
+        conn.executescript(SCHEMA)
+        conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
+    finally:
+        conn.close()
+
+
+@contextmanager
+def transaction(conn: sqlite3.Connection) -> Iterator[None]:
+    conn.execute("BEGIN IMMEDIATE")
+    try:
+        yield
+    except BaseException:
+        conn.execute("ROLLBACK")
+        raise
+    conn.execute("COMMIT")
+
+
+def get_db(request: Request) -> Iterator[sqlite3.Connection]:
+    conn = connect(request.app.state.settings.db_path)
+    try:
+        yield conn
+    finally:
+        conn.close()

backend/app/main.py

@@ -0,0 +1,118 @@
+import asyncio
+import logging
+import sqlite3
+from contextlib import asynccontextmanager
+from datetime import timedelta
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, Request
+
+from . import db, repo
+from .auth import require_ingest_key
+from .models import IngestIn
+from .settings import HISTORY_DAYS, Settings, load_settings
+from .timeutil import to_db, utcnow
+
+log = logging.getLogger("backup_dashboard")
+PURGE_INTERVAL_SECONDS = 6 * 3600
+
+router = APIRouter(prefix="/api/v1")
+
+
+@router.get("/health")
+def health(conn: sqlite3.Connection = Depends(db.get_db)):
+    conn.execute("SELECT 1")
+    return {"status": "ok"}
+
+
+@router.get("/jobs")
+def list_jobs(conn: sqlite3.Connection = Depends(db.get_db)):
+    return {"generated_at": to_db(utcnow()), "jobs": repo.list_jobs(conn)}
+
+
+@router.get("/jobs/{job_id}")
+def get_job(job_id: int, conn: sqlite3.Connection = Depends(db.get_db)):
+    job = repo.get_job(conn, job_id)
+    if job is None:
+        raise HTTPException(404, "Job not found")
+    return job
+
+
+@router.get("/jobs/{job_id}/runs")
+def list_runs(
+    job_id: int,
+    days: int = Query(HISTORY_DAYS, ge=1, le=HISTORY_DAYS),
+    conn: sqlite3.Connection = Depends(db.get_db),
+):
+    if conn.execute("SELECT 1 FROM jobs WHERE id = ?", (job_id,)).fetchone() is None:
+        raise HTTPException(404, "Job not found")
+    since = utcnow() - timedelta(days=days)
+    return {"job_id": job_id, "since": to_db(since), "runs": repo.list_runs(conn, job_id, since)}
+
+
+@router.get("/runs/{run_id}")
+def get_run(run_id: int, conn: sqlite3.Connection = Depends(db.get_db)):
+    run = repo.get_run(conn, run_id)
+    if run is None:
+        raise HTTPException(404, "Run not found")
+    return run
+
+
+@router.post("/ingest")
+def ingest(
+    item: IngestIn,
+    key_name: str = Depends(require_ingest_key),
+    conn: sqlite3.Connection = Depends(db.get_db),
+):
+    job_id, run_id = repo.ingest(conn, item)
+    log.info("ingest key=%s tenant=%r job=%r run=%s", key_name, item.tenant, item.job.name, run_id)
+    return {"job_id": job_id, "run_id": run_id}
+
+
+async def _purge_loop(settings: Settings) -> None:
+    def purge() -> int:
+        conn = db.connect(settings.db_path)
+        try:
+            return repo.purge_old_runs(conn, settings.retention_days)
+        finally:
+            conn.close()
+
+    while True:
+        try:
+            if deleted := await asyncio.to_thread(purge):
+                log.info("purged %d runs older than %d days", deleted, settings.retention_days)
+        except Exception:
+            log.exception("run purge failed")
+        await asyncio.sleep(PURGE_INTERVAL_SECONDS)
+
+
+def create_app(settings: Settings | None = None) -> FastAPI:
+    settings = settings or load_settings()
+
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        db.init_db(settings.db_path)
+        task = asyncio.create_task(_purge_loop(settings))
+        try:
+            yield
+        finally:
+            task.cancel()
+
+    # Interactive docs are off: nginx's strict CSP would block their CDN assets anyway.
+    app = FastAPI(
+        title="Backup Dashboard API",
+        lifespan=lifespan,
+        docs_url=None,
+        redoc_url=None,
+        openapi_url="/api/v1/openapi.json",
+    )
+    app.state.settings = settings
+    app.include_router(router)
+
+    if settings.serve_static:
+        from fastapi.staticfiles import StaticFiles
+
+        public = Path(__file__).resolve().parents[2] / "public"
+        app.mount("/", StaticFiles(directory=public, html=True), name="static")
+
+    return app