118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
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"}
|