refactor(auth): centralize token hashing, email validation & admin gating

- app/utils.py: shared valid_email() + now_utc() (one email regex, was duplicated
  in auth.py and admin.py with a looser "@"-in check elsewhere).
- security.hash_token(): one SHA-256 token hasher (was duplicated in auth.py + state.py).
- auth.admin_user dependency + count_admins() helper, replacing the inline role
  checks and the last-admin count query repeated across admin.py/me.py/tags.py.
This commit is contained in:
npeter83 2026-06-26 03:15:36 +02:00
parent 8ef748c7b8
commit 63701f5344
7 changed files with 84 additions and 65 deletions

View file

@ -1,6 +1,5 @@
"""Admin-only onboarding: review access requests (Invites), approve/deny, manual add.
Also the demo-account admin surface: the demo email whitelist + a manual state reset."""
import re
from datetime import datetime, timezone
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
@ -8,7 +7,13 @@ from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session
from app import email as email_mod
from app.auth import current_user, get_or_create_demo_user, operator_contact, purge_user
from app.auth import (
admin_user,
count_admins,
get_or_create_demo_user,
operator_contact,
purge_user,
)
from app.db import get_db
from app.models import (
DemoWhitelist,
@ -20,17 +25,10 @@ from app.models import (
Video,
VideoState,
)
from app.utils import valid_email
router = APIRouter(prefix="/api/admin", tags=["admin"])
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
def admin_user(user: User = Depends(current_user)) -> User:
if user.role != "admin":
raise HTTPException(status_code=403, detail="Admin only")
return user
def _serialize(inv: Invite) -> dict:
return {
@ -110,7 +108,7 @@ def add_invite(
) -> dict:
"""Manually whitelist an email (approved straight away). Convenience for the admin."""
email = (payload.get("email") or "").strip().lower()
if "@" not in email:
if not valid_email(email):
raise HTTPException(status_code=400, detail="Enter a valid email address.")
inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
if inv is None:
@ -168,10 +166,8 @@ def set_user_role(
raise HTTPException(status_code=400, detail="The demo account's role can't be changed.")
if target.id == admin.id:
raise HTTPException(status_code=400, detail="You can't change your own role.")
if target.role == "admin" and role == "user":
admin_count = db.scalar(select(func.count()).select_from(User).where(User.role == "admin"))
if (admin_count or 0) <= 1:
raise HTTPException(status_code=400, detail="Can't remove the last admin.")
if target.role == "admin" and role == "user" and count_admins(db) <= 1:
raise HTTPException(status_code=400, detail="Can't remove the last admin.")
changed = target.role != role
target.role = role
db.commit()
@ -201,14 +197,13 @@ def set_user_suspended(
raise HTTPException(status_code=400, detail="The demo account can't be suspended.")
if target.id == admin.id:
raise HTTPException(status_code=400, detail="You can't suspend your own account.")
if suspended and target.role == "admin" and not target.is_suspended:
active_admins = db.scalar(
select(func.count())
.select_from(User)
.where(User.role == "admin", User.is_suspended.is_(False))
)
if (active_admins or 0) <= 1:
raise HTTPException(status_code=400, detail="Can't suspend the last active admin.")
if (
suspended
and target.role == "admin"
and not target.is_suspended
and count_admins(db, active_only=True) <= 1
):
raise HTTPException(status_code=400, detail="Can't suspend the last active admin.")
was_suspended = target.is_suspended
target.is_suspended = suspended
db.commit()
@ -238,12 +233,8 @@ def admin_delete_user(
raise HTTPException(
status_code=400, detail="Delete your own account from Settings → Account."
)
if target.role == "admin":
admin_count = db.scalar(
select(func.count()).select_from(User).where(User.role == "admin")
)
if (admin_count or 0) <= 1:
raise HTTPException(status_code=400, detail="Can't delete the last admin.")
if target.role == "admin" and count_admins(db) <= 1:
raise HTTPException(status_code=400, detail="Can't delete the last admin.")
purge_user(db, target, background)
return {"deleted": user_id}
@ -281,7 +272,7 @@ def add_demo_whitelist(
"""Add an email that may enter the shared demo account from the login page (no Google
sign-in). Idempotent re-adding an existing email just returns it."""
email = (payload.get("email") or "").strip().lower()
if not _EMAIL_RE.match(email):
if not valid_email(email):
raise HTTPException(status_code=400, detail="Enter a valid email address.")
row = db.execute(
select(DemoWhitelist).where(DemoWhitelist.email == email)

View file

@ -3,6 +3,7 @@ from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.auth import (
count_admins,
current_user,
google_enabled,
has_read_scope,
@ -111,13 +112,11 @@ def delete_my_account(
(that would lock everyone out of the admin surfaces)."""
if user.is_demo:
raise HTTPException(status_code=403, detail="The shared demo account can't be deleted.")
if user.role == "admin":
admins = db.scalar(select(func.count()).select_from(User).where(User.role == "admin")) or 0
if admins <= 1:
raise HTTPException(
status_code=400,
detail="You're the only admin — promote another admin before deleting your account.",
)
if user.role == "admin" and count_admins(db) <= 1:
raise HTTPException(
status_code=400,
detail="You're the only admin — promote another admin before deleting your account.",
)
user_id = user.id
# Full erasure (cascades) + access-request cleanup + Google-grant revocation; shared with the
# admin delete path. Capture the id first — `user` is detached once purge_user deletes the row.

View file

@ -3,7 +3,7 @@ from sqlalchemy import func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.auth import current_user
from app.auth import admin_user, current_user
from app.db import get_db
from app.models import ChannelTag, Tag, User
from app.sync.autotag import run_autotag_all
@ -103,10 +103,8 @@ def delete_tag(
@router.post("/recompute")
def recompute(
user: User = Depends(current_user),
_: User = Depends(admin_user),
db: Session = Depends(get_db),
only_missing: bool = False,
) -> dict:
if user.role != "admin":
raise HTTPException(status_code=403, detail="Admin only")
return run_autotag_all(db, only_missing=only_missing)