RECORDGOD/app/auth.py
type-two 5e2a9e7ab1 feat(auth): staff accounts + roles — operators run the store, admins keep the keys
Token resolver now checks the env owner-token (admin) then a staff table (name/token/
role). New require_admin gates /settings/* (API keys/connections) + staff CRUD; everything
operational (search/scan/inventory/labels/intake) stays on require_token so staff can do it.
admin.html: /admin/me drives ROLE; data-admin nav (Staff + Connections) hidden for staff +
go() blocks the views; new Staff tab to add operators, set role, reset/show token once,
deactivate. Tokens shown once on create.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-23 15:33:19 +10:00

35 lines
1.5 KiB
Python

import os
from fastapi import Header, HTTPException, Depends
from sqlalchemy import text
from .db import get_db
# Single-tenant v1: one store. The env token is the OWNER (always admin); staff get their own
# tokens from the `staff` table with a role (admin | staff). Role gates the sensitive endpoints.
STORE = os.getenv("RECORDGOD_STORE", "Monster Robot Party")
PLAN = os.getenv("RECORDGOD_PLAN", "enterprise")
TOKEN = os.getenv("RECORDGOD_TOKEN") # no default — unset means "deny everything", not "open"
async def require_token(authorization: str | None = Header(None), db=Depends(get_db)):
# Bearer per WOWPLATTER_BRIEF. Owner env-token first (no DB hit); then the staff table.
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(401, "missing bearer token")
tok = authorization[7:].strip()
base = {"store": STORE, "plan": PLAN, "store_id": 1}
if TOKEN and tok == TOKEN:
return {**base, "role": "admin", "name": "Owner", "staff_id": None}
row = (await db.execute(
text("SELECT id, name, role FROM staff WHERE token = :t AND active"), {"t": tok}
)).mappings().first()
if row:
return {**base, "role": row["role"], "name": row["name"], "staff_id": row["id"]}
raise HTTPException(401, "invalid token")
async def require_admin(ident=Depends(require_token)):
"""Gate the sensitive surface (API keys / connections / staff admin) to admins only."""
if ident.get("role") != "admin":
raise HTTPException(403, "admin only")
return ident