2add173760
Move the access whitelist from the ALLOWED_EMAILS env var into a DB Invite table (env kept as bootstrap fallback), and add a self-service request + admin approval flow with fail-soft email. - models: Invite(email, status pending|approved|denied, requested_at, decided_*) - migration 0008: invites table; seed env ALLOWED_EMAILS u ADMIN_EMAILS as approved - auth: is_allowed() (DB-first, env fallback); a denied Google login records a pending request and bounces to /?access=requested instead of a raw 403; public POST /auth/request-access; upsert is idempotent so repeats don't re-spam admins - routes/admin.py (admin-only): list/approve/deny invites + manual add - email.py: smtplib + Gmail App Password, fail-soft (skips if SMTP unset) - /api/me exposes pending_invites; config + .env.example gain SMTP_* - UI: Login 'Request access' form + access=requested/denied handling; Settings -> Access requests (approve/deny + add); admin nudge toast on pending requests Verified locally: request-access creates a pending invite and emails the admin; seed approved npeter83; guinea-pig yt.trash2023 denied until approved.
100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
"""Admin-only onboarding: review access requests (Invites), approve/deny, manual add."""
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import email as email_mod
|
|
from app.auth import current_user
|
|
from app.db import get_db
|
|
from app.models import Invite, User
|
|
|
|
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
|
|
|
|
|
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 {
|
|
"id": inv.id,
|
|
"email": inv.email,
|
|
"status": inv.status,
|
|
"note": inv.note,
|
|
"requested_at": inv.requested_at.isoformat() if inv.requested_at else None,
|
|
"decided_at": inv.decided_at.isoformat() if inv.decided_at else None,
|
|
"decided_by": inv.decided_by,
|
|
}
|
|
|
|
|
|
@router.get("/invites")
|
|
def list_invites(
|
|
status: str | None = None,
|
|
_: User = Depends(admin_user),
|
|
db: Session = Depends(get_db),
|
|
) -> list[dict]:
|
|
stmt = select(Invite)
|
|
if status:
|
|
stmt = stmt.where(Invite.status == status)
|
|
# Pending first, then most-recently requested.
|
|
rows = db.execute(stmt.order_by(Invite.requested_at.desc())).scalars().all()
|
|
order = {"pending": 0, "approved": 1, "denied": 2}
|
|
rows.sort(key=lambda i: order.get(i.status, 9))
|
|
return [_serialize(i) for i in rows]
|
|
|
|
|
|
def _decide(db: Session, invite_id: int, admin: User, approved: bool) -> Invite:
|
|
inv = db.get(Invite, invite_id)
|
|
if inv is None:
|
|
raise HTTPException(status_code=404, detail="No such invite")
|
|
inv.status = "approved" if approved else "denied"
|
|
inv.decided_at = datetime.now(timezone.utc)
|
|
inv.decided_by = admin.email
|
|
db.commit()
|
|
return inv
|
|
|
|
|
|
@router.post("/invites/{invite_id}/approve")
|
|
def approve_invite(
|
|
invite_id: int,
|
|
background: BackgroundTasks,
|
|
admin: User = Depends(admin_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
inv = _decide(db, invite_id, admin, approved=True)
|
|
background.add_task(email_mod.send_access_approved, inv.email)
|
|
return _serialize(inv)
|
|
|
|
|
|
@router.post("/invites/{invite_id}/deny")
|
|
def deny_invite(
|
|
invite_id: int,
|
|
admin: User = Depends(admin_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
return _serialize(_decide(db, invite_id, admin, approved=False))
|
|
|
|
|
|
@router.post("/invites")
|
|
def add_invite(
|
|
payload: dict,
|
|
admin: User = Depends(admin_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Manually whitelist an email (approved straight away). Convenience for the admin."""
|
|
email = (payload.get("email") or "").strip().lower()
|
|
if "@" not in 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:
|
|
inv = Invite(email=email)
|
|
db.add(inv)
|
|
inv.status = "approved"
|
|
inv.decided_at = datetime.now(timezone.utc)
|
|
inv.decided_by = admin.email
|
|
db.commit()
|
|
return _serialize(inv)
|