derekc
aa12648228
- Rename app from Eggtracker to Yolkbook throughout - Add JWT-based authentication (python-jose, passlib/bcrypt) - Add users table; all data tables gain user_id FK for full data isolation - Super admin credentials sourced from ADMIN_USERNAME/ADMIN_PASSWORD env vars, synced on every startup; orphaned rows auto-assigned to admin post-migration - Login page with self-registration; JWT stored in localStorage (30-day expiry) - Admin panel (/admin): list users, reset passwords, disable/enable, delete, and impersonate (Login As) with Return to Admin banner - Settings modal (gear icon in nav): timezone selector and change password - Timezone stored per-user; stats date windows computed in user's timezone; date input setToday() respects user timezone via Intl API - migrate_v2.sql for existing single-user installs - Auto-migration adds timezone column to users on startup - Updated README with full setup, auth, admin, and migration docs Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from database import get_db
|
|
from models import User
|
|
from schemas import UserCreate, UserOut, ResetPasswordRequest, TokenResponse
|
|
from auth import hash_password, create_access_token, get_current_admin, get_current_user
|
|
|
|
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
|
|
|
|
|
@router.get("/users", response_model=list[UserOut])
|
|
def list_users(
|
|
_: User = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return db.scalars(select(User).order_by(User.created_at)).all()
|
|
|
|
|
|
@router.post("/users", response_model=UserOut, status_code=201)
|
|
def create_user(
|
|
body: UserCreate,
|
|
_: User = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
existing = db.scalars(select(User).where(User.username == body.username)).first()
|
|
if existing:
|
|
raise HTTPException(status_code=409, detail="Username already taken")
|
|
user = User(
|
|
username=body.username,
|
|
hashed_password=hash_password(body.password),
|
|
is_admin=False,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.post("/users/{user_id}/reset-password")
|
|
def reset_password(
|
|
user_id: int,
|
|
body: ResetPasswordRequest,
|
|
current_admin: User = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = db.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
user.hashed_password = hash_password(body.new_password)
|
|
db.commit()
|
|
return {"detail": f"Password reset for {user.username}"}
|
|
|
|
|
|
@router.post("/users/{user_id}/disable")
|
|
def disable_user(
|
|
user_id: int,
|
|
current_admin: User = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = db.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if user.id == current_admin.id:
|
|
raise HTTPException(status_code=400, detail="Cannot disable your own account")
|
|
user.is_disabled = True
|
|
db.commit()
|
|
return {"detail": f"User {user.username} disabled"}
|
|
|
|
|
|
@router.post("/users/{user_id}/enable")
|
|
def enable_user(
|
|
user_id: int,
|
|
_: User = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = db.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
user.is_disabled = False
|
|
db.commit()
|
|
return {"detail": f"User {user.username} enabled"}
|
|
|
|
|
|
@router.delete("/users/{user_id}", status_code=204)
|
|
def delete_user(
|
|
user_id: int,
|
|
current_admin: User = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = db.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if user.id == current_admin.id:
|
|
raise HTTPException(status_code=400, detail="Cannot delete your own account")
|
|
db.delete(user)
|
|
db.commit()
|
|
|
|
|
|
@router.post("/users/{user_id}/impersonate", response_model=TokenResponse)
|
|
def impersonate_user(
|
|
user_id: int,
|
|
_: User = Depends(get_current_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = db.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
token = create_access_token(user.id, user.username, user.is_admin, user.timezone)
|
|
return TokenResponse(access_token=token)
|