B0rbor4d
d1a757d72b
- Backend: /api/users (LIST/CREATE/PATCH, admin-only) mit Self-Protection (Admin kann sich nicht selbst deaktivieren/entziehen), lokale User anlegbar, LDAP-User verwaltbar (aktiv/admin), Passwort-Reset nur lokal - Frontend: Benutzer-View (Liste, anlegen, toggles, Passwort setzen), Nav-Eintrag + Route - Bugfix 'Speichern tut nichts': submit() brach stillschweigend ab wenn kein Kunde gewaehlt war oder die API einen Fehler lieferte - jetzt wird eine Fehlermeldung angezeigt (Servers, Customers, Satellites)
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
"""Dashboard user management (admin only).
|
|
|
|
LDAP users are provisioned automatically on first login; they can be
|
|
deactivated or promoted here, but their password always lives in AD.
|
|
Local users can additionally be created and re-passworded.
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import client_ip, require_admin
|
|
from app.core.database import get_db
|
|
from app.core.exceptions import ConflictError, NotFoundError, ValidationError
|
|
from app.core.security import hash_password
|
|
from app.models.user import User
|
|
from app.schemas.user import UserCreate, UserRead, UserUpdate
|
|
from app.services.audit import AuditService
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", response_model=list[UserRead])
|
|
async def list_users(
|
|
db: AsyncSession = Depends(get_db),
|
|
_admin: User = Depends(require_admin),
|
|
) -> list[User]:
|
|
result = await db.execute(select(User).order_by(User.username))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
@router.post("", response_model=UserRead, status_code=201)
|
|
async def create_user(
|
|
payload: UserCreate,
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
) -> User:
|
|
existing = await db.execute(select(User).where(User.username == payload.username))
|
|
if existing.scalar_one_or_none():
|
|
raise ConflictError("Benutzername ist bereits vergeben")
|
|
|
|
user = User(
|
|
username=payload.username,
|
|
email=payload.email,
|
|
full_name=payload.full_name,
|
|
password_hash=hash_password(payload.password),
|
|
is_ldap=False,
|
|
is_admin=payload.is_admin,
|
|
)
|
|
db.add(user)
|
|
await db.flush()
|
|
await AuditService(db).log(
|
|
username=admin.username,
|
|
action="user.create",
|
|
target=user.username,
|
|
ip_address=client_ip(request),
|
|
)
|
|
return user
|
|
|
|
|
|
@router.patch("/{user_id}", response_model=UserRead)
|
|
async def update_user(
|
|
user_id: int,
|
|
payload: UserUpdate,
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
) -> User:
|
|
user = await db.get(User, user_id)
|
|
if not user:
|
|
raise NotFoundError("Benutzer nicht gefunden")
|
|
|
|
# Self-protection: an admin must not lock themselves out
|
|
if user.id == admin.id:
|
|
if payload.is_active is False:
|
|
raise ValidationError("Du kannst dich nicht selbst deaktivieren")
|
|
if payload.is_admin is False:
|
|
raise ValidationError("Du kannst dir nicht selbst die Adminrechte entziehen")
|
|
|
|
if payload.is_admin is not None:
|
|
user.is_admin = payload.is_admin
|
|
if payload.is_active is not None:
|
|
user.is_active = payload.is_active
|
|
if payload.full_name is not None:
|
|
user.full_name = payload.full_name
|
|
if payload.email is not None:
|
|
user.email = payload.email
|
|
if payload.password is not None:
|
|
if user.is_ldap:
|
|
raise ValidationError("LDAP-User haben kein lokales Passwort")
|
|
user.password_hash = hash_password(payload.password)
|
|
|
|
await AuditService(db).log(
|
|
username=admin.username,
|
|
action="user.update",
|
|
target=user.username,
|
|
details={"fields": sorted(payload.model_dump(exclude_unset=True).keys())},
|
|
ip_address=client_ip(request),
|
|
)
|
|
return user
|