Add an admin page for managing collector API keys
Admins (Microsoft 365 sign-in names in BACKUP_DASH_ADMINS) get an Admin link to /admin.html, where they can create, list and revoke the API keys collectors use, instead of running manage.py on the server. Backend - /api/v1/admin/keys (list, create, revoke), admins only (403 otherwise); the user comes from the headers nginx sets after sign-in - Changes need an X-Requested-By header, which another site's page can't add, so a signed-in admin's browser can't be used to forge them - A new key is returned once; only its hash is stored. Names are unique for good, so revoked keys' history stays unambiguous - Keys record who created/revoked them and the company they last reported for (schema v2; databases migrate automatically on start) - /api/v1/me reports is_admin; manage.py shares the key code and records itself as creator/revoker Frontend - admin.html / admin.js: name validation shown after interaction (:user-invalid, aria-invalid kept in sync), the one-time key with a copy button (falls back to selecting it if clipboard access is refused), a key table flagging keys quiet for over 24 hours, and revoking confirmed in a <dialog>; results announced in a status region - Shared el() and signed-in helpers moved to dom.js / signed-in.js - Mock mode (?mock=1) includes sample keys Tested: 15 new backend tests (access, forged requests, names, duplicates, revoking, last-used company, v1 -> v2 upgrade); in the browser, every admin flow in mock mode and at phone width; end to end through the nginx config with the real API (a key created on the API works for ingest; revoking it makes ingest 401; non-admins see an explanation and no Admin link). Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
"""Collector API keys: create, list, revoke. Used by the admin API and manage.py."""
|
||||
|
||||
import re
|
||||
import sqlite3
|
||||
from typing import Any
|
||||
|
||||
from .auth import generate_key, hash_key
|
||||
from .timeutil import to_db, utcnow
|
||||
|
||||
# Letters, digits, '.', '_', '-'; starts with a letter or digit. E.g. "veeam-acme", "mars.fs01"
|
||||
NAME_PATTERN = r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}"
|
||||
|
||||
_FIELDS = "id, name, created_at, created_by, last_used_at, last_used_tenant, revoked_at, revoked_by"
|
||||
|
||||
|
||||
class KeyNameTaken(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def valid_name(name: str) -> bool:
|
||||
return re.fullmatch(NAME_PATTERN, name) is not None
|
||||
|
||||
|
||||
def list_keys(conn: sqlite3.Connection) -> list[dict[str, Any]]:
|
||||
# Active keys first, then by name
|
||||
rows = conn.execute(f"SELECT {_FIELDS} FROM api_keys ORDER BY revoked_at IS NOT NULL, name").fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def get_key(conn: sqlite3.Connection, key_id: int) -> dict[str, Any] | None:
|
||||
row = conn.execute(f"SELECT {_FIELDS} FROM api_keys WHERE id = ?", (key_id,)).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def create_key(conn: sqlite3.Connection, name: str, created_by: str) -> tuple[dict[str, Any], str]:
|
||||
"""Returns the stored key's record and the key itself, which isn't stored and can't be shown again."""
|
||||
secret = generate_key()
|
||||
try:
|
||||
key_id = conn.execute(
|
||||
"INSERT INTO api_keys (name, key_hash, created_at, created_by) VALUES (?, ?, ?, ?) RETURNING id",
|
||||
(name, hash_key(secret), to_db(utcnow()), created_by),
|
||||
).fetchone()[0]
|
||||
except sqlite3.IntegrityError:
|
||||
# Names stay taken after revocation, so the audit trail stays unambiguous
|
||||
raise KeyNameTaken(name) from None
|
||||
return get_key(conn, key_id), secret
|
||||
|
||||
|
||||
def revoke_key(conn: sqlite3.Connection, key_id: int, revoked_by: str) -> dict[str, Any] | None:
|
||||
"""Revokes the key (no-op if already revoked). Returns its record, or None if there's no such key."""
|
||||
conn.execute(
|
||||
"UPDATE api_keys SET revoked_at = ?, revoked_by = ? WHERE id = ? AND revoked_at IS NULL",
|
||||
(to_db(utcnow()), revoked_by, key_id),
|
||||
)
|
||||
return get_key(conn, key_id)
|
||||
Reference in New Issue
Block a user