admin.schubert/Olympus
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(rbac): add roles and permissions

Browse source
This commit is contained in:
Schubert Ferenc 2026-07-02 23:05:59 +02:00
parent 86a32a942c
commit 694b7bd09a
37 changed files with 2682 additions and 218 deletions
Download patch file Download diff file Expand all files Collapse all files

16
backend/hermes/app/api/auth.py
View file

@ -6,8 +6,11 @@ from app.core.security import (
get_access_token_expire_seconds,
verify_password,
)
from app.core.rbac import get_current_active_user, get_user_permission_names
from app.db.database import get_db
from app.models.user import User
from app.repositories.user_repository import UserRepository
from app.schemas.rbac import CurrentUserResponse
from app.schemas.user import LoginRequest, LoginResponse
router = APIRouter(
@ -21,7 +24,7 @@ def login(
login: LoginRequest,
db: Session = Depends(get_db),
):
user = db.query(User).filter(User.username == login.username).first()
user = UserRepository.get_by_username(db, login.username)
if not user:
raise HTTPException(
@ -43,3 +46,14 @@ def login(
"expires_in": get_access_token_expire_seconds(),
"user": user,
}
@router.get("/me", response_model=CurrentUserResponse)
def me(current_user: User = Depends(get_current_active_user)):
return {
"id": current_user.id,
"username": current_user.username,
"email": current_user.email,
"role": current_user.primary_role.name,
"permissions": sorted(get_user_permission_names(current_user)),
}

27
backend/hermes/app/api/permissions.py Normal file
View file

@ -0,0 +1,27 @@
import logging
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.core.rbac import require_permission
from app.db.database import get_db
from app.models.user import User
from app.repositories.rbac_repository import RbacRepository
from app.schemas.rbac import PermissionResponse
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/permissions",
tags=["Permissions"],
)
@router.get("", response_model=list[PermissionResponse])
@router.get("/", response_model=list[PermissionResponse], include_in_schema=False)
def get_permissions(
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("roles.read")),
):
logger.info("permissions.list", extra={"actor_user_id": current_user.id})
return RbacRepository.get_permissions(db)

153
backend/hermes/app/api/roles.py Normal file
View file

@ -0,0 +1,153 @@
import logging
from fastapi import APIRouter, Depends, HTTPException, Response, status
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.rbac import require_permission
from app.db.database import get_db
from app.models.user import User
from app.repositories.rbac_repository import RbacRepository
from app.schemas.rbac import RoleCreate, RolePermissionUpdate, RoleResponse, RoleUpdate
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/roles",
tags=["Roles"],
)
def get_role_or_404(db: Session, role_id: int):
role = RbacRepository.get_role_by_id(db, role_id)
if role is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Rolle nicht gefunden",
)
return role
def ensure_unique_role_name(db: Session, name: str, role_id: int | None = None) -> None:
role = RbacRepository.get_role_by_name(db, name)
if role is not None and role.id != role_id:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Rollenname ist bereits vergeben",
)
def ensure_permission_ids_exist(db: Session, permission_ids: list[int]) -> None:
permissions = RbacRepository.get_permissions_by_ids(db, permission_ids)
if len(permissions) != len(set(permission_ids)):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Mindestens eine Berechtigung existiert nicht",
)
@router.get("", response_model=list[RoleResponse])
@router.get("/", response_model=list[RoleResponse], include_in_schema=False)
def get_roles(
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("roles.read")),
):
logger.info("roles.list", extra={"actor_user_id": current_user.id})
return RbacRepository.get_roles(db)
@router.get("/{role_id}", response_model=RoleResponse)
def get_role(
role_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("roles.read")),
):
logger.info(
"roles.detail",
extra={"actor_user_id": current_user.id, "target_role_id": role_id},
)
return get_role_or_404(db, role_id)
@router.post("", response_model=RoleResponse, status_code=status.HTTP_201_CREATED)
@router.post(
"/",
response_model=RoleResponse,
status_code=status.HTTP_201_CREATED,
include_in_schema=False,
)
def create_role(
role: RoleCreate,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("roles.create")),
):
ensure_unique_role_name(db, role.name)
ensure_permission_ids_exist(db, role.permission_ids)
logger.info("roles.create", extra={"actor_user_id": current_user.id})
return RbacRepository.create_role(db, role)
@router.put("/{role_id}", response_model=RoleResponse)
def update_role(
role_id: int,
role: RoleUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("roles.update")),
):
db_role = get_role_or_404(db, role_id)
ensure_unique_role_name(db, role.name, role_id)
ensure_permission_ids_exist(db, role.permission_ids)
logger.info(
"roles.update",
extra={"actor_user_id": current_user.id, "target_role_id": role_id},
)
return RbacRepository.update_role(db, db_role, role)
@router.delete("/{role_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_role(
role_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("roles.delete")),
):
db_role = get_role_or_404(db, role_id)
if db_role.is_system:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Systemrollen dürfen nicht gelöscht werden",
)
assigned_user_id = db.scalar(select(User.id).where(User.role_id == role_id).limit(1))
if assigned_user_id is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Rolle ist Benutzern zugewiesen und kann nicht gelöscht werden",
)
logger.info(
"roles.delete",
extra={"actor_user_id": current_user.id, "target_role_id": role_id},
)
RbacRepository.delete_role(db, db_role)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.put("/{role_id}/permissions", response_model=RoleResponse)
def update_role_permissions(
role_id: int,
permission_update: RolePermissionUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("roles.assign")),
):
db_role = get_role_or_404(db, role_id)
ensure_permission_ids_exist(db, permission_update.permission_ids)
logger.info(
"roles.permissions_update",
extra={"actor_user_id": current_user.id, "target_role_id": role_id},
)
return RbacRepository.update_role_permissions(db, db_role, permission_update)

85
backend/hermes/app/api/users.py
View file

@ -1,12 +1,12 @@
import logging
from fastapi import APIRouter, Depends, HTTPException, Response, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from app.core.security import verify_access_token
from app.core.rbac import get_user_permission_names, require_permission
from app.db.database import get_db
from app.models.user import User
from app.repositories.rbac_repository import RbacRepository
from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate, UserPasswordUpdate, UserResponse, UserUpdate
@ -17,45 +17,6 @@ router = APIRouter(
tags=["Users"],
)
bearer_scheme = HTTPBearer(auto_error=False)
def get_current_user(
credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
db: Session = Depends(get_db),
) -> User:
if credentials is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Nicht authentifiziert",
headers={"WWW-Authenticate": "Bearer"},
)
username = verify_access_token(credentials.credentials)
user = UserRepository.get_by_username(db, username)
if user is None or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Nicht authentifiziert",
headers={"WWW-Authenticate": "Bearer"},
)
return user
def require_role(*allowed_roles: str):
def dependency(current_user: User = Depends(get_current_user)) -> User:
if current_user.role not in allowed_roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Keine Berechtigung",
)
return current_user
return dependency
def get_user_or_404(db: Session, user_id: int) -> User:
db_user = UserRepository.get_by_id(db, user_id)
@ -80,11 +41,36 @@ def raise_conflict(field: str) -> None:
)
def require_actor_permission(current_user: User, permission: str) -> None:
if permission not in get_user_permission_names(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Keine Berechtigung",
)
def apply_role(db: Session, user: UserCreate | UserUpdate) -> None:
role = None
if user.role_id is not None:
role = RbacRepository.get_role_by_id(db, user.role_id)
else:
role = RbacRepository.get_role_by_name(db, user.role)
if role is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Rolle nicht gefunden",
)
user.role_id = role.id
user.role = role.name
@router.get("", response_model=list[UserResponse])
@router.get("/", response_model=list[UserResponse], include_in_schema=False)
def get_users(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
current_user: User = Depends(require_permission("users.read")),
):
logger.info("users.list", extra={"actor_user_id": current_user.id})
return UserRepository.get_all(db)
@ -94,7 +80,7 @@ def get_users(
def get_user(
user_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
current_user: User = Depends(require_permission("users.read")),
):
logger.info(
"users.detail",
@ -113,8 +99,10 @@ def get_user(
def create_user(
user: UserCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
current_user: User = Depends(require_permission("users.create")),
):
require_actor_permission(current_user, "roles.assign")
apply_role(db, user)
conflict = UserRepository.find_conflict(
db,
username=user.username,
@ -133,9 +121,12 @@ def update_user(
user_id: int,
user: UserUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
current_user: User = Depends(require_permission("users.update")),
):
db_user = get_user_or_404(db, user_id)
apply_role(db, user)
if user.role_id != db_user.role_id:
require_actor_permission(current_user, "roles.assign")
conflict = UserRepository.find_conflict(
db,
username=user.username,
@ -158,7 +149,7 @@ def update_user_password(
user_id: int,
password_update: UserPasswordUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
current_user: User = Depends(require_permission("users.password.update")),
):
db_user = get_user_or_404(db, user_id)
@ -173,7 +164,7 @@ def update_user_password(
def delete_user(
user_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
current_user: User = Depends(require_permission("users.delete")),
):
if current_user.id == user_id:
raise HTTPException(

94
backend/hermes/app/core/rbac.py Normal file
View file

@ -0,0 +1,94 @@
from collections.abc import Iterable
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from app.core.security import verify_access_token
from app.db.database import get_db
from app.models.user import User
from app.repositories.user_repository import UserRepository
bearer_scheme = HTTPBearer(auto_error=False)
def get_current_user(
credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
db: Session = Depends(get_db),
) -> User:
if credentials is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Nicht authentifiziert",
headers={"WWW-Authenticate": "Bearer"},
)
username = verify_access_token(credentials.credentials)
user = UserRepository.get_by_username(db, username)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Nicht authentifiziert",
headers={"WWW-Authenticate": "Bearer"},
)
return user
def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
if not current_user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Nicht authentifiziert",
headers={"WWW-Authenticate": "Bearer"},
)
return current_user
def get_user_permission_names(user: User) -> set[str]:
if user.primary_role is None:
return set()
return {permission.name for permission in user.primary_role.permissions}
def require_permission(permission: str):
def dependency(current_user: User = Depends(get_current_active_user)) -> User:
if permission not in get_user_permission_names(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Keine Berechtigung",
)
return current_user
return dependency
def require_any_permission(permissions: Iterable[str]):
permission_set = set(permissions)
def dependency(current_user: User = Depends(get_current_active_user)) -> User:
if get_user_permission_names(current_user).isdisjoint(permission_set):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Keine Berechtigung",
)
return current_user
return dependency
def require_all_permissions(permissions: Iterable[str]):
permission_set = set(permissions)
def dependency(current_user: User = Depends(get_current_active_user)) -> User:
if not permission_set.issubset(get_user_permission_names(current_user)):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Keine Berechtigung",
)
return current_user
return dependency

3
backend/hermes/app/db/database.py
View file

@ -20,6 +20,7 @@ class Base(DeclarativeBase):
pass
# <<< HIER IMPORTIEREN >>>
import app.models.rbac
import app.models.user
@ -28,4 +29,4 @@ def get_db():
try:
yield db
finally:
db.close()
db.close()

15
backend/hermes/app/main.py
View file

@ -9,8 +9,12 @@ from starlette import status
from starlette.requests import Request
from app.api.auth import router as auth_router
from app.api.permissions import router as permissions_router
from app.api.roles import router as roles_router
from app.api.users import router as users_router
from app.db.database import SessionLocal
from app.db.health import check_database
from app.rbac.seed import seed_rbac
app = FastAPI(
title="Hermes API",
@ -20,10 +24,21 @@ app = FastAPI(
app.include_router(auth_router)
app.include_router(users_router)
app.include_router(roles_router)
app.include_router(permissions_router)
logger = logging.getLogger(__name__)
@app.on_event("startup")
def startup_seed_rbac():
db = SessionLocal()
try:
seed_rbac(db)
finally:
db.close()
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
logger.warning(

58
backend/hermes/app/models/rbac.py Normal file
View file

@ -0,0 +1,58 @@
from datetime import datetime
from sqlalchemy import Boolean, DateTime, ForeignKey, String, Table, Column, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.db.database import Base
role_permissions = Table(
"role_permissions",
Base.metadata,
Column("role_id", ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True),
Column("permission_id", ForeignKey("permissions.id", ondelete="CASCADE"), primary_key=True),
)
class Role(Base):
__tablename__ = "roles"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(80), unique=True, index=True)
display_name: Mapped[str] = mapped_column(String(120))
description: Mapped[str] = mapped_column(String(500), default="", server_default="")
is_system: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
)
permissions: Mapped[list["Permission"]] = relationship(
secondary=role_permissions,
back_populates="roles",
lazy="selectin",
)
class Permission(Base):
__tablename__ = "permissions"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(120), unique=True, index=True)
display_name: Mapped[str] = mapped_column(String(160))
description: Mapped[str] = mapped_column(String(500), default="", server_default="")
module: Mapped[str] = mapped_column(String(80), index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
)
roles: Mapped[list[Role]] = relationship(
secondary=role_permissions,
back_populates="permissions",
lazy="selectin",
)

9
backend/hermes/app/models/user.py
View file

@ -2,8 +2,10 @@ from datetime import datetime
from sqlalchemy import Boolean
from sqlalchemy import DateTime
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy import func
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
@ -48,6 +50,13 @@ class User(Base):
server_default="user"
)
role_id: Mapped[int] = mapped_column(
ForeignKey("roles.id"),
index=True,
)
primary_role = relationship("Role", lazy="joined")
is_active: Mapped[bool] = mapped_column(
Boolean,
default=True,

112
backend/hermes/app/rbac/defaults.py Normal file
View file

@ -0,0 +1,112 @@
STANDARD_ROLES = [
{
"name": "administrator",
"display_name": "Administrator",
"description": "Vollzugriff auf Olympus CRM",
"is_system": True,
},
{
"name": "management",
"display_name": "Management",
"description": "Lesender Zugriff auf operative Kernmodule",
"is_system": True,
},
{
"name": "sales",
"display_name": "Vertrieb",
"description": "Zugriff auf Kunden und Projekte",
"is_system": True,
},
{
"name": "technician",
"display_name": "Technik",
"description": "Zugriff auf Tickets und Projekte",
"is_system": True,
},
{
"name": "support",
"display_name": "Support",
"description": "Operativer Support-Zugriff",
"is_system": True,
},
{
"name": "warehouse",
"display_name": "Lager",
"description": "Zugriff auf lagernahe Prozesse",
"is_system": True,
},
{
"name": "guest",
"display_name": "Gast",
"description": "Minimaler Zugriff",
"is_system": True,
},
]
STANDARD_PERMISSIONS = [
("users.read", "Benutzer lesen", "Benutzer anzeigen", "users"),
("users.create", "Benutzer erstellen", "Benutzer anlegen", "users"),
("users.update", "Benutzer bearbeiten", "Benutzer aktualisieren", "users"),
("users.delete", "Benutzer loeschen", "Benutzer entfernen", "users"),
("users.password.update", "Passwort aendern", "Benutzerpasswort aendern", "users"),
("roles.read", "Rollen lesen", "Rollen anzeigen", "roles"),
("roles.create", "Rollen erstellen", "Rollen anlegen", "roles"),
("roles.update", "Rollen bearbeiten", "Rollen aktualisieren", "roles"),
("roles.delete", "Rollen loeschen", "Rollen entfernen", "roles"),
("roles.assign", "Rollen zuweisen", "Berechtigungen Rollen zuweisen", "roles"),
("customers.read", "Kunden lesen", "Kunden anzeigen", "customers"),
("customers.create", "Kunden erstellen", "Kunden anlegen", "customers"),
("customers.update", "Kunden bearbeiten", "Kunden aktualisieren", "customers"),
("customers.delete", "Kunden loeschen", "Kunden entfernen", "customers"),
("projects.read", "Projekte lesen", "Projekte anzeigen", "projects"),
("projects.create", "Projekte erstellen", "Projekte anlegen", "projects"),
("projects.update", "Projekte bearbeiten", "Projekte aktualisieren", "projects"),
("projects.delete", "Projekte loeschen", "Projekte entfernen", "projects"),
("tickets.read", "Tickets lesen", "Tickets anzeigen", "tickets"),
("tickets.create", "Tickets erstellen", "Tickets anlegen", "tickets"),
("tickets.update", "Tickets bearbeiten", "Tickets aktualisieren", "tickets"),
("tickets.delete", "Tickets loeschen", "Tickets entfernen", "tickets"),
("dashboard.read", "Dashboard lesen", "Dashboard anzeigen", "dashboard"),
("system.settings.read", "Einstellungen lesen", "Systemeinstellungen anzeigen", "system"),
("system.settings.update", "Einstellungen bearbeiten", "Systemeinstellungen aktualisieren", "system"),
]
ROLE_PERMISSION_NAMES = {
"administrator": "*",
"management": {
"dashboard.read",
"users.read",
"customers.read",
"projects.read",
"tickets.read",
},
"sales": {
"dashboard.read",
"customers.read",
"customers.create",
"customers.update",
"projects.read",
"tickets.read",
},
"technician": {
"dashboard.read",
"projects.read",
"tickets.read",
"tickets.create",
"tickets.update",
},
"support": {
"dashboard.read",
"users.read",
"customers.read",
"tickets.read",
"tickets.create",
"tickets.update",
},
"warehouse": {
"dashboard.read",
},
"guest": {
"dashboard.read",
},
}

63
backend/hermes/app/rbac/seed.py Normal file
View file

@ -0,0 +1,63 @@
import logging
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.rbac import Permission, Role
from app.rbac.defaults import ROLE_PERMISSION_NAMES, STANDARD_PERMISSIONS, STANDARD_ROLES
logger = logging.getLogger(__name__)
def seed_rbac(db: Session) -> None:
roles_by_name = {
role.name: role
for role in db.scalars(select(Role)).all()
}
permissions_by_name = {
permission.name: permission
for permission in db.scalars(select(Permission)).all()
}
for role_data in STANDARD_ROLES:
role = roles_by_name.get(role_data["name"])
if role is None:
role = Role(**role_data)
db.add(role)
roles_by_name[role.name] = role
else:
role.display_name = role_data["display_name"]
role.description = role_data["description"]
role.is_system = role_data["is_system"]
for name, display_name, description, module in STANDARD_PERMISSIONS:
permission = permissions_by_name.get(name)
if permission is None:
permission = Permission(
name=name,
display_name=display_name,
description=description,
module=module,
)
db.add(permission)
permissions_by_name[name] = permission
else:
permission.display_name = display_name
permission.description = description
permission.module = module
db.flush()
all_permissions = set(permissions_by_name.values())
for role_name, permission_names in ROLE_PERMISSION_NAMES.items():
role = roles_by_name[role_name]
if permission_names == "*":
role.permissions = list(all_permissions)
else:
role.permissions = [
permissions_by_name[permission_name]
for permission_name in permission_names
]
db.commit()
logger.info("rbac.seeded")

93
backend/hermes/app/repositories/rbac_repository.py Normal file
View file

@ -0,0 +1,93 @@
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload
from app.models.rbac import Permission, Role
from app.schemas.rbac import RoleCreate, RolePermissionUpdate, RoleUpdate
class RbacRepository:
@staticmethod
def get_roles(db: Session) -> list[Role]:
return list(
db.scalars(
select(Role)
.options(selectinload(Role.permissions))
.order_by(Role.name)
)
)
@staticmethod
def get_role_by_id(db: Session, role_id: int) -> Role | None:
return db.scalar(
select(Role)
.where(Role.id == role_id)
.options(selectinload(Role.permissions))
)
@staticmethod
def get_role_by_name(db: Session, name: str) -> Role | None:
return db.scalar(
select(Role)
.where(Role.name == name)
.options(selectinload(Role.permissions))
)
@staticmethod
def get_permissions(db: Session) -> list[Permission]:
return list(db.scalars(select(Permission).order_by(Permission.module, Permission.name)))
@staticmethod
def get_permissions_by_ids(db: Session, permission_ids: list[int]) -> list[Permission]:
if not permission_ids:
return []
return list(db.scalars(select(Permission).where(Permission.id.in_(permission_ids))))
@staticmethod
def create_role(db: Session, role: RoleCreate) -> Role:
db_role = Role(
name=role.name,
display_name=role.display_name,
description=role.description,
is_system=False,
)
db_role.permissions = RbacRepository.get_permissions_by_ids(db, role.permission_ids)
db.add(db_role)
db.commit()
db.refresh(db_role)
return RbacRepository.get_role_by_id(db, db_role.id) or db_role
@staticmethod
def update_role(db: Session, db_role: Role, role: RoleUpdate) -> Role:
db_role.name = role.name
db_role.display_name = role.display_name
db_role.description = role.description
db_role.permissions = RbacRepository.get_permissions_by_ids(db, role.permission_ids)
db.commit()
db.refresh(db_role)
return RbacRepository.get_role_by_id(db, db_role.id) or db_role
@staticmethod
def update_role_permissions(
db: Session,
db_role: Role,
permission_update: RolePermissionUpdate,
) -> Role:
db_role.permissions = RbacRepository.get_permissions_by_ids(
db,
permission_update.permission_ids,
)
db.commit()
db.refresh(db_role)
return RbacRepository.get_role_by_id(db, db_role.id) or db_role
@staticmethod
def delete_role(db: Session, db_role: Role) -> None:
db.delete(db_role)
db.commit()

32
backend/hermes/app/repositories/user_repository.py
View file

@ -1,7 +1,8 @@
from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlalchemy.orm import Session, joinedload, selectinload
from app.core.security import hash_password
from app.models.rbac import Role
from app.models.user import User
from app.schemas.user import UserCreate, UserPasswordUpdate, UserUpdate
@ -9,15 +10,31 @@ from app.schemas.user import UserCreate, UserPasswordUpdate, UserUpdate
class UserRepository:
@staticmethod
def get_all(db: Session) -> list[User]:
return list(db.scalars(select(User).order_by(User.created_at.desc())))
return list(
db.scalars(
select(User)
.options(
joinedload(User.primary_role).selectinload(Role.permissions),
)
.order_by(User.created_at.desc())
)
)
@staticmethod
def get_by_id(db: Session, user_id: int) -> User | None:
return db.get(User, user_id)
return db.scalar(
select(User)
.where(User.id == user_id)
.options(joinedload(User.primary_role).selectinload(Role.permissions))
)
@staticmethod
def get_by_username(db: Session, username: str) -> User | None:
return db.scalar(select(User).where(User.username == username))
return db.scalar(
select(User)
.where(User.username == username)
.options(joinedload(User.primary_role).selectinload(Role.permissions))
)
@staticmethod
def find_conflict(
@ -52,6 +69,7 @@ class UserRepository:
username=user.username,
email=str(user.email),
role=user.role,
role_id=user.role_id,
is_active=user.is_active,
password_hash=hash_password(user.password),
)
@ -60,7 +78,7 @@ class UserRepository:
db.commit()
db.refresh(db_user)
return db_user
return UserRepository.get_by_id(db, db_user.id) or db_user
@staticmethod
def update(db: Session, db_user: User, user: UserUpdate) -> User:
@ -69,6 +87,8 @@ class UserRepository:
db_user.username = user.username
db_user.email = str(user.email)
db_user.role = user.role
if user.role_id is not None:
db_user.role_id = user.role_id
db_user.is_active = user.is_active
if user.password:
@ -77,7 +97,7 @@ class UserRepository:
db.commit()
db.refresh(db_user)
return db_user
return UserRepository.get_by_id(db, db_user.id) or db_user
@staticmethod
def update_password(

74
backend/hermes/app/schemas/rbac.py Normal file
View file

@ -0,0 +1,74 @@
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
RoleName = str
class PermissionResponse(BaseModel):
id: int
name: str
display_name: str
description: str
module: str
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class RoleBase(BaseModel):
name: str = Field(min_length=3, max_length=80)
display_name: str = Field(min_length=2, max_length=120)
description: str = Field(default="", max_length=500)
@field_validator("name")
@classmethod
def validate_name(cls, value: str) -> str:
normalized = value.strip().lower()
allowed = normalized.replace("_", "").replace("-", "").replace(".", "")
if not allowed.isalnum():
raise ValueError(
"Rollenname darf nur Buchstaben, Zahlen, Punkt, Unterstrich und Bindestrich enthalten"
)
return normalized
@field_validator("display_name", "description", mode="before")
@classmethod
def normalize_text(cls, value: object) -> str:
if value is None:
return ""
return str(value).strip()
class RoleCreate(RoleBase):
permission_ids: list[int] = Field(default_factory=list)
class RoleUpdate(RoleBase):
permission_ids: list[int] = Field(default_factory=list)
class RolePermissionUpdate(BaseModel):
permission_ids: list[int] = Field(default_factory=list)
class RoleResponse(BaseModel):
id: int
name: str
display_name: str
description: str
is_system: bool
permissions: list[PermissionResponse]
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class CurrentUserResponse(BaseModel):
id: int
username: str
email: str
role: str
permissions: list[str]

11
backend/hermes/app/schemas/user.py
View file

@ -1,9 +1,9 @@
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
UserRole = Literal["admin", "manager", "user"]
from app.schemas.rbac import RoleResponse
UserRole = str
class UserBase(BaseModel):
@ -11,7 +11,8 @@ class UserBase(BaseModel):
last_name: str = Field(default="", max_length=100)
username: str = Field(min_length=3, max_length=50)
email: EmailStr
role: UserRole = "user"
role: UserRole = "support"
role_id: int | None = None
is_active: bool = True
@field_validator("first_name", "last_name", "username", mode="before")
@ -65,6 +66,8 @@ class UserResponse(BaseModel):
username: str
email: EmailStr
role: UserRole
role_id: int
primary_role: RoleResponse
is_active: bool
created_at: datetime
updated_at: datetime