from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from typing import List from passlib.context import CryptContext from .. import models, schemas, database, auth from ..logger import log router = APIRouter(prefix="/users", tags=["users"]) pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto") def get_db(): db = database.SessionLocal() try: yield db finally: db.close() def get_password_hash(password): return pwd_context.hash(password) def verify_password(plain_password, hashed_password): if not hashed_password: return False return pwd_context.verify(plain_password, hashed_password) @router.get("/", response_model=List[schemas.User]) def get_users(db: Session = Depends(get_db)): """[C-01] User list — public endpoint for login page to enumerate local users.""" users = db.query(models.User).all() # Auto-seed if empty if not users: # [SECURITY] For initial setup and recovery, we use a predictable default. # User MUST change this immediately in Settings. initial_password = "Admin123!" new_user = models.User( username="Admin", role="admin", origin="local", hashed_password=get_password_hash(initial_password) ) db.add(new_user) db.commit() db.refresh(new_user) log.warning(f"[SECURITY] Admin initial seeded. Credentials: Admin / {initial_password} — CHANGE IMMEDIATELY!") return [new_user] return users @router.post("/", response_model=schemas.User) def create_user( user: schemas.UserCreate, db: Session = Depends(get_db), current_user: auth.TokenData = Depends(auth.get_current_admin) ): """[C-01] Create user — admin only.""" existing = db.query(models.User).filter(models.User.username == user.username).first() if existing: raise HTTPException(status_code=400, detail="Username already exists") hashed = get_password_hash(user.password) if user.password else None new_user = models.User(username=user.username, role=user.role, origin="local", hashed_password=hashed) db.add(new_user) db.commit() db.refresh(new_user) return new_user @router.put("/{user_id}", response_model=schemas.User) def update_user( user_id: int, user_update: schemas.UserUpdate, db: Session = Depends(get_db), current_user: auth.TokenData = Depends(auth.get_current_admin) ): """[C-01] Update user — admin only.""" db_user = db.query(models.User).filter(models.User.id == user_id).first() if not db_user: raise HTTPException(status_code=404, detail="User not found") if user_update.username and db_user.username == "Admin" and user_update.username != "Admin": raise HTTPException(status_code=400, detail="Cannot change Admin username") if user_update.username: # Check if username already taken by another user existing = db.query(models.User).filter(models.User.username == user_update.username, models.User.id != user_id).first() if existing: raise HTTPException(status_code=400, detail="Username already exists") db_user.username = user_update.username if user_update.password: db_user.hashed_password = get_password_hash(user_update.password) if user_update.role: db_user.role = user_update.role db.commit() db.refresh(db_user) return db_user @router.delete("/{user_id}") def delete_user( user_id: int, db: Session = Depends(get_db), current_user: auth.TokenData = Depends(auth.get_current_admin) ): """[C-01] Delete user — admin only.""" user = db.query(models.User).filter(models.User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") if user.username == "Admin": raise HTTPException(status_code=400, detail="Cannot delete default Admin") is_ldap = user.origin == "ldap" db.delete(user) db.commit() log.info(f"User {user_id} ({user.username}) deleted by Admin. Source: {user.origin}") return {"message": "User deleted" if not is_ldap else "LDAP cache cleared for this user"}