New independent `admin` service (admin.main:app) on the same image, reusing app.db/app.models read-only. Never runs migrations or the scheduler; issues SELECTs only. - Password-gated (ADMIN_CONSOLE_PASSWORD) with a 12h signed cookie; closed by default when the password is empty. - Bound to 127.0.0.1:8091 (SSH-tunnel access); off the intranet/NPM network. - Pages: overview stats, user list + search, per-user history/payment detail, DB usage (information_schema size + row estimates). - Compose: base `admin` service (+prod DB-host override, test mount); Dockerfile bakes admin/ into runtime + test stages. - Tests: tests/test_admin_console.py (auth, queries, page wiring) — 12 passing. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
"""Password gate for the superadmin console.
|
|
|
|
A single shared password (``ADMIN_CONSOLE_PASSWORD``) is exchanged at
|
|
``/login`` for a signed, time-limited session cookie. There is no user
|
|
identity here — the console has exactly one principal, "the operator".
|
|
|
|
The cookie is signed with itsdangerous using ``ADMIN_CONSOLE_SESSION_SECRET``
|
|
(falling back to the main app's session secret / token), mirroring the scheme
|
|
in ``app.auth`` but under a distinct salt so an app session cookie can never
|
|
be replayed against the console and vice-versa.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
|
|
|
|
from app.config import get_settings
|
|
|
|
SESSION_COOKIE_NAME = "admin_console_session"
|
|
SESSION_TTL_SECONDS = 12 * 60 * 60 # 12 hours — re-auth daily-ish.
|
|
_SALT = "admin-console-session-v1"
|
|
|
|
|
|
def _serializer() -> URLSafeTimedSerializer:
|
|
s = get_settings()
|
|
secret = (
|
|
s.ADMIN_CONSOLE_SESSION_SECRET
|
|
or s.CASSANDRA_SESSION_SECRET
|
|
or s.CASSANDRA_TOKEN
|
|
or "dev-insecure-secret"
|
|
)
|
|
return URLSafeTimedSerializer(secret, salt=_SALT)
|
|
|
|
|
|
def check_password(provided: str) -> bool:
|
|
"""Constant-time compare against ``ADMIN_CONSOLE_PASSWORD``. Returns
|
|
False when the password is unset — the console is closed by default so a
|
|
fresh deploy can't be logged into without an explicit password."""
|
|
expected = get_settings().ADMIN_CONSOLE_PASSWORD
|
|
if not expected:
|
|
return False
|
|
return secrets.compare_digest(provided.encode(), expected.encode())
|
|
|
|
|
|
def sign_session() -> str:
|
|
"""Signed value proving a successful password login. Carries no identity
|
|
beyond a version marker — the console has a single principal."""
|
|
return _serializer().dumps({"v": 1})
|
|
|
|
|
|
def verify_session(cookie: str) -> bool:
|
|
try:
|
|
data = _serializer().loads(cookie, max_age=SESSION_TTL_SECONDS)
|
|
return data.get("v") == 1
|
|
except (BadSignature, SignatureExpired, KeyError, TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
def require_admin(request: Request) -> None:
|
|
"""FastAPI dependency guarding every console page. Valid cookie → passes;
|
|
otherwise 303 → /login (browser) so the operator lands on the form."""
|
|
cookie = request.cookies.get(SESSION_COOKIE_NAME)
|
|
if cookie and verify_session(cookie):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_303_SEE_OTHER,
|
|
detail="Login required",
|
|
headers={"Location": "/login"},
|
|
)
|