119 lines
4.1 KiB
Python
119 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
|
|
import ldap3
|
|
import json
|
|
import os
|
|
from .. import models, schemas, database
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
|
|
|
def get_ldap_config():
|
|
config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "ldap_config.json")
|
|
if os.path.exists(config_path):
|
|
with open(config_path, "r") as f:
|
|
return json.load(f)
|
|
return {"ldap_enabled": False}
|
|
|
|
def authenticate_ldap(username, password):
|
|
config = get_ldap_config()
|
|
if not config.get("ldap_enabled"):
|
|
return False
|
|
|
|
try:
|
|
server = ldap3.Server(config["server_uri"], use_ssl=config.get("use_tls", False), get_info=ldap3.ALL)
|
|
user_dn = config["user_template"].format(username=username)
|
|
conn = ldap3.Connection(server, user=user_dn, password=password, auto_bind=True)
|
|
# If bound, user is authenticated
|
|
return True
|
|
except Exception as e:
|
|
print(f"LDAP Auth Error: {e}")
|
|
return False
|
|
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)):
|
|
users = db.query(models.User).all()
|
|
# Auto-seed if empty
|
|
if not users:
|
|
new_user = models.User(
|
|
username="Admin",
|
|
role="admin",
|
|
hashed_password=get_password_hash("admin") # Default password
|
|
)
|
|
db.add(new_user)
|
|
db.commit()
|
|
db.refresh(new_user)
|
|
return [new_user]
|
|
return users
|
|
|
|
@router.post("/", response_model=schemas.User)
|
|
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
|
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, hashed_password=hashed)
|
|
db.add(new_user)
|
|
db.commit()
|
|
db.refresh(new_user)
|
|
return new_user
|
|
|
|
@router.post("/login")
|
|
def login(form_data: schemas.UserLogin, db: Session = Depends(get_db)):
|
|
user = db.query(models.User).filter(models.User.username == form_data.username).first()
|
|
|
|
# Try local authentication
|
|
authenticated = False
|
|
if user and user.hashed_password:
|
|
if verify_password(form_data.password, user.hashed_password):
|
|
authenticated = True
|
|
elif user and not user.hashed_password:
|
|
# Legacy user without password - allow skip for now or force set
|
|
authenticated = True
|
|
|
|
# If local failed, try LDAP
|
|
if not authenticated:
|
|
if authenticate_ldap(form_data.username, form_data.password):
|
|
authenticated = True
|
|
# If user doesn't exist locally, create a stub for role management
|
|
if not user:
|
|
user = models.User(username=form_data.username, role="user")
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
else:
|
|
raise HTTPException(status_code=400, detail="Invalid username or password")
|
|
|
|
return user
|
|
|
|
@router.delete("/{user_id}")
|
|
def delete_user(user_id: int, db: Session = Depends(get_db)):
|
|
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")
|
|
|
|
db.delete(user)
|
|
db.commit()
|
|
return {"message": "User deleted"}
|