Files
data-entry-app/backend/app/api/access.py
T

785 lines
28 KiB
Python
Raw Normal View History

"""Internal-user authentication and permission introspection routes.
Frontends should call ``GET /api/access/me`` to discover which permission keys
the current user has, then use those keys to hide/show navigation items.
**Visibility is not security** — every privileged backend route must depend on
``require_permission`` (or one of its siblings) directly.
"""
from __future__ import annotations
2026-05-10 09:46:07 +12:00
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from pydantic import BaseModel
2026-06-17 21:55:04 +12:00
from sqlalchemy import func, select
from sqlalchemy.orm import Session, selectinload
from app.core.access import (
INTERNAL_USER_SUBJECT,
INTERNAL_USER_TENANT_ID,
get_current_user,
get_user_permissions,
permissions_to_module_map,
require_permission,
)
from app.core.config import settings
2026-05-10 09:46:07 +12:00
from app.core.http import CLIENT_AUTH_COOKIE
from app.core.rate_limit import SlidingWindowRateLimiter, request_client_key
from app.core.security_logging import log_security_event
2026-05-08 09:06:14 +12:00
from app.core.security import hash_password, issue_token, verify_password
from app.db.session import get_db
from app.models.access import Permission, Role, User
router = APIRouter(prefix="/api/access", tags=["access"])
2026-05-10 09:46:07 +12:00
login_rate_limiter = SlidingWindowRateLimiter(
limit=settings.login_rate_limit_attempts,
window_seconds=settings.login_rate_limit_window_seconds,
)
class LoginRequest(BaseModel):
email: str
password: str
class UserSession(BaseModel):
# Mirrors the existing `LoginResponse` shape so the frontend's `AppSession`
# store can consume this response without a separate type. `permissions`
# is the new permission-key array; `module_permissions` is the legacy
# module→access-level map for nav gating.
user_id: int
email: str
name: str
role: str
role_name: str | None = None
is_active: bool
tenant_id: str | None = None
client_role: str | None = None
client_account_id: int | None = None
module_permissions: dict[str, str] = {}
permissions: list[str]
token: str | None = None
class RoleRead(BaseModel):
id: int
name: str
description: str | None
permissions: list[str]
2026-06-17 21:55:04 +12:00
module_permissions: dict[str, str]
is_protected: bool = False
user_count: int = 0
class UserRead(BaseModel):
id: int
email: str
name: str
is_active: bool
role: str | None
2026-06-17 21:55:04 +12:00
role_id: int | None
# True when this user can never be deleted (lean owner accounts). The UI
# uses this to disable the delete control rather than re-deriving the rule.
is_protected: bool = False
class AssignableRole(BaseModel):
id: int
name: str
description: str | None
class RoleModuleDefinition(BaseModel):
key: str
label: str
description: str
levels: list[str]
class CreateUserRequest(BaseModel):
email: str
name: str
role_id: int | None = None
is_active: bool = True
password: str | None = None
class AdminUpdateUserRequest(BaseModel):
name: str | None = None
email: str | None = None
role_id: int | None = None
is_active: bool | None = None
class AdminSetPasswordRequest(BaseModel):
new_password: str
class CreateRoleRequest(BaseModel):
name: str
description: str | None = None
module_permissions: dict[str, str] = {}
class UpdateRoleRequest(BaseModel):
name: str | None = None
description: str | None = None
module_permissions: dict[str, str] | None = None
# Lean owner accounts are permanent: they may be edited but never deleted, so a
# tenant can't accidentally lock itself out of the highest level of access.
LEAN_ROLE_NAME = "lean"
ADMIN_ROLE_NAME = "admin"
ROLE_MANAGEMENT_ALLOWED_ROLES = {LEAN_ROLE_NAME, ADMIN_ROLE_NAME}
PROTECTED_ROLE_NAMES = ROLE_MANAGEMENT_ALLOWED_ROLES
ROLE_MODULE_DEFINITIONS: tuple[dict[str, object], ...] = (
{
"key": "dashboard",
"label": "Dashboard",
"description": "Home dashboard visibility.",
"levels": {"none": (), "view": ("view_dashboard",)},
},
{
"key": "mix_calculator",
"label": "Mix Calculator",
"description": "Open the calculator and save sessions.",
"levels": {
"none": (),
"view": ("view_mix_calculator",),
"edit": ("view_mix_calculator", "use_mix_calculator", "save_mix_calculator_session"),
},
},
{
"key": "raw_materials",
"label": "Raw Materials",
"description": "View or edit raw materials.",
"levels": {"none": (), "view": ("view_raw_materials",), "edit": ("view_raw_materials", "edit_raw_materials")},
},
{
"key": "products",
"label": "Products",
"description": "View or edit finished products.",
"levels": {"none": (), "view": ("view_products",), "edit": ("view_products", "edit_products")},
},
{
"key": "mix_master",
"label": "Mix Master",
"description": "View or edit mix recipes.",
"levels": {"none": (), "view": ("view_mixes",), "edit": ("view_mixes", "edit_mixes")},
},
{
"key": "operations_throughput",
"label": "Throughput",
"description": "View or edit throughput entries.",
"levels": {"none": (), "view": ("view_throughput",), "edit": ("view_throughput", "edit_throughput")},
},
{
"key": "ordering",
"label": "Ordering",
"description": "Access customer ordering and ordering administration.",
"levels": {
"none": (),
"view": ("view_ordering",),
"edit": ("view_ordering", "edit_ordering"),
"manage": ("view_ordering", "edit_ordering", "manage_ordering"),
},
},
{
"key": "scenarios",
"label": "Scenarios",
"description": "View or run scenarios.",
"levels": {"none": (), "view": ("view_scenarios",), "edit": ("view_scenarios", "edit_scenarios")},
},
{
"key": "client_access",
"label": "Client Access",
"description": "Manage customer portal accounts and access.",
"levels": {"none": (), "manage": ("manage_client_access",)},
},
{
"key": "users",
"label": "Users",
"description": "View or manage internal users.",
"levels": {"none": (), "view": ("view_users",), "manage": ("view_users", "manage_users")},
},
{
"key": "roles",
"label": "Roles",
"description": "Manage roles and permission assignments.",
"levels": {"none": (), "manage": ("manage_permissions",)},
},
{
"key": "settings",
"label": "Settings",
"description": "Open settings and edit system configuration.",
"levels": {"none": (), "view": ("view_settings",), "edit": ("view_settings", "edit_settings")},
},
)
def _serialize_user_read(user: User) -> UserRead:
role_name = user.role.name if user.role else None
return UserRead(
id=user.id,
email=user.email,
name=user.name,
is_active=user.is_active,
role=role_name,
role_id=user.role_id,
is_protected=(role_name or "").lower() == LEAN_ROLE_NAME,
)
def _role_name_lower(role: Role | None) -> str:
return (role.name if role else "").strip().lower()
def _is_protected_role_name(role_name: str | None) -> bool:
return (role_name or "").strip().lower() in PROTECTED_ROLE_NAMES
def _module_definitions_response() -> list[RoleModuleDefinition]:
return [
RoleModuleDefinition(
key=definition["key"],
label=definition["label"],
description=definition["description"],
levels=list(definition["levels"].keys()),
)
for definition in ROLE_MODULE_DEFINITIONS
]
def _permissions_to_role_module_map(permission_keys: set[str]) -> dict[str, str]:
result: dict[str, str] = {}
for definition in ROLE_MODULE_DEFINITIONS:
selected = "none"
levels = definition["levels"]
for level, required in levels.items():
required_keys = set(required)
if not required_keys or required_keys.issubset(permission_keys):
selected = level
result[definition["key"]] = selected
return result
def _role_payload_to_permission_keys(module_permissions: dict[str, str]) -> set[str]:
known_modules = {definition["key"] for definition in ROLE_MODULE_DEFINITIONS}
unknown_modules = sorted(set(module_permissions) - known_modules)
if unknown_modules:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Unknown modules: {unknown_modules}",
)
granted: set[str] = set()
for definition in ROLE_MODULE_DEFINITIONS:
key = definition["key"]
level = module_permissions.get(key, "none")
available_levels: dict[str, tuple[str, ...]] = definition["levels"] # type: ignore[assignment]
if level not in available_levels:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Invalid access level '{level}' for module '{key}'",
)
granted.update(available_levels[level])
return granted
def _serialize_role_read(role: Role, *, user_count: int = 0) -> RoleRead:
permission_keys = {permission.key for permission in role.permissions}
return RoleRead(
id=role.id,
name=role.name,
description=role.description,
permissions=sorted(permission_keys),
module_permissions=_permissions_to_role_module_map(permission_keys),
is_protected=_is_protected_role_name(role.name),
user_count=user_count,
)
def _require_role_management_actor(user: User = Depends(get_current_user)) -> User:
if _role_name_lower(user.role) not in ROLE_MANAGEMENT_ALLOWED_ROLES:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only lean and admin accounts can manage roles",
)
return user
def _load_role(db: Session, role_id: int) -> Role:
role = db.scalar(
select(Role).where(Role.id == role_id).options(selectinload(Role.permissions))
)
if role is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Role not found")
return role
def _apply_role_updates(
role: Role,
*,
name: str | None,
description: str | None,
module_permissions: dict[str, str] | None,
permissions_by_key: dict[str, Permission],
) -> None:
if name is not None:
trimmed_name = name.strip()
if not trimmed_name:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Role name cannot be empty")
role.name = trimmed_name
if description is not None:
role.description = description.strip() or None
if module_permissions is not None:
desired_permission_keys = _role_payload_to_permission_keys(module_permissions)
desired = {permissions_by_key[key] for key in desired_permission_keys}
current = set(role.permissions)
for permission in desired - current:
role.permissions.append(permission)
for permission in current - desired:
role.permissions.remove(permission)
def _serialize_session(user: User, *, include_token: bool = False) -> UserSession:
permission_set = get_user_permissions(user)
permissions = sorted(permission_set)
module_permissions = permissions_to_module_map(permission_set)
role_name = user.role.name if user.role else None
token = None
if include_token:
2026-05-10 09:46:07 +12:00
token = issue_token(
{"sub": INTERNAL_USER_SUBJECT, "user_id": user.id, "email": user.email},
ttl_seconds=settings.session_ttl_seconds,
)
# role="internal" is a marker the shared auth deps recognise so internal
# users can hit the same routes as client-portal users without being
# confused with them. Display name lives in role_name / client_role.
return UserSession(
user_id=user.id,
email=user.email,
name=user.name,
role="internal",
role_name=role_name,
is_active=user.is_active,
tenant_id=INTERNAL_USER_TENANT_ID,
client_role=role_name,
client_account_id=None,
module_permissions=module_permissions,
permissions=permissions,
token=token,
)
@router.post("/login", response_model=UserSession)
2026-05-10 09:46:07 +12:00
def login(payload: LoginRequest, response: Response, request: Request, db: Session = Depends(get_db)):
"""Internal-user login.
2026-05-31 20:19:44 +12:00
Authenticates against the per-user password hash stored on ``users``.
Inactive or unknown users are rejected with a generic 401 to avoid
leaking which emails are valid.
"""
2026-05-10 09:46:07 +12:00
login_rate_limiter.hit(request_client_key(request, suffix="internal-login"))
email = payload.email.strip().lower()
user = db.scalar(
select(User)
.where(User.email == email)
.options(selectinload(User.role).selectinload(Role.permissions))
)
if user is None or not user.is_active:
2026-05-10 09:46:07 +12:00
log_security_event("auth.login_failed", audience="internal", ip=request_client_key(request))
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password")
2026-05-31 20:19:44 +12:00
if not (
verify_password(payload.password, user.password_hash)
or (user.password_hash is None and payload.password == settings.admin_password)
):
log_security_event("auth.login_failed", audience="internal", ip=request_client_key(request))
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password")
2026-05-10 09:46:07 +12:00
session = _serialize_session(user, include_token=True)
if session.token:
CLIENT_AUTH_COOKIE.apply(response, session.token)
log_security_event("auth.login_succeeded", audience="internal", role=user.role.name if user.role else None, user_id=user.id)
return session.model_copy(update={"token": None})
@router.get("/me", response_model=UserSession)
def read_me(user: User = Depends(get_current_user)):
"""Return the current user with permission keys for UI navigation gating."""
2026-05-10 09:46:07 +12:00
return _serialize_session(user).model_copy(update={"token": None})
@router.get("/me/permissions", response_model=list[str])
def read_my_permissions(user: User = Depends(get_current_user)):
return sorted(get_user_permissions(user))
2026-05-08 09:06:14 +12:00
class UpdateMeRequest(BaseModel):
name: str | None = None
email: str | None = None
current_password: str | None = None
new_password: str | None = None
@router.patch("/me", response_model=UserSession)
def update_me(
payload: UpdateMeRequest,
db: Session = Depends(get_db),
user: User = Depends(get_current_user),
):
"""Allow an internal user to update their own name, email, or password."""
if payload.new_password:
2026-05-31 20:19:44 +12:00
# Require current password verification before allowing a password
# change. Keep a narrow fallback for legacy rows that still have no
# password hash yet.
current_ok = verify_password(payload.current_password or "", user.password_hash) or (
user.password_hash is None and (payload.current_password or "") == settings.admin_password
2026-05-08 09:06:14 +12:00
)
if not current_ok:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Current password is incorrect",
)
if len(payload.new_password) < 8:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="New password must be at least 8 characters",
)
user.password_hash = hash_password(payload.new_password)
if payload.name is not None:
name = payload.name.strip()
if not name:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Name cannot be empty")
user.name = name
if payload.email is not None:
email = payload.email.strip().lower()
if not email or "@" not in email:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid email address")
existing = db.scalar(select(User).where(User.email == email, User.id != user.id))
if existing:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email is already in use")
user.email = email
db.commit()
db.refresh(user)
2026-05-10 09:46:07 +12:00
return _serialize_session(user, include_token=True).model_copy(update={"token": None})
@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
def logout(response: Response):
CLIENT_AUTH_COOKIE.clear(response)
response.status_code = status.HTTP_204_NO_CONTENT
return None
2026-05-08 09:06:14 +12:00
# Permission-enforced administrative endpoints. Route bodies should not check
# role names — every gate is the require_permission(...) dependency.
@router.get("/users", response_model=list[UserRead])
def list_users(
db: Session = Depends(get_db),
_: User = Depends(require_permission("view_users")), # gated by permission key
):
2026-06-17 21:55:04 +12:00
users = db.scalars(
select(User).options(selectinload(User.role)).order_by(User.name)
).all()
return [_serialize_user_read(user) for user in users]
@router.get("/assignable-roles", response_model=list[AssignableRole])
def list_assignable_roles(
db: Session = Depends(get_db),
_: User = Depends(require_permission("manage_users")), # gated by permission key
):
"""Roles that a user-manager can assign — used to populate the role picker.
Separate from ``/roles`` (which exposes full permission sets and is gated by
``manage_permissions``); managing users only needs the role list itself.
"""
roles = db.scalars(select(Role).order_by(Role.name)).all()
return [
2026-06-17 21:55:04 +12:00
AssignableRole(id=role.id, name=role.name, description=role.description)
for role in roles
]
2026-06-17 21:55:04 +12:00
def _load_managed_user(db: Session, user_id: int) -> User:
user = db.scalar(
select(User).where(User.id == user_id).options(selectinload(User.role))
)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
return user
def _resolve_role(db: Session, role_id: int | None) -> Role | None:
if role_id is None:
return None
role = db.scalar(select(Role).where(Role.id == role_id))
if role is None:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Unknown role")
return role
@router.post("/users", response_model=UserRead, status_code=status.HTTP_201_CREATED)
def create_user(
payload: CreateUserRequest,
db: Session = Depends(get_db),
actor: User = Depends(require_permission("manage_users")), # gated by permission key
):
"""Create a new internal user.
A user with no password can still sign in with the shared internal password
until they set a personal one in their own settings.
"""
email = payload.email.strip().lower()
if not email or "@" not in email:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid email address")
name = payload.name.strip()
if not name:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Name cannot be empty")
if db.scalar(select(User).where(User.email == email)):
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email is already in use")
role = _resolve_role(db, payload.role_id)
password_hash = None
if payload.password:
if len(payload.password) < 8:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Password must be at least 8 characters",
)
password_hash = hash_password(payload.password)
user = User(
email=email,
name=name,
role_id=role.id if role else None,
is_active=payload.is_active,
password_hash=password_hash,
)
db.add(user)
db.commit()
db.refresh(user)
log_security_event("users.created", audience="internal", actor_user_id=actor.id, user_id=user.id)
return _serialize_user_read(user)
@router.patch("/users/{user_id}", response_model=UserRead)
def update_user(
user_id: int,
payload: AdminUpdateUserRequest,
db: Session = Depends(get_db),
actor: User = Depends(require_permission("manage_users")), # gated by permission key
):
"""Update another user's name, email, role, or active status."""
user = _load_managed_user(db, user_id)
if payload.is_active is not None:
# Guard against locking yourself out of your own management session.
if user.id == actor.id and not payload.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="You cannot deactivate your own account",
)
user.is_active = payload.is_active
if payload.name is not None:
name = payload.name.strip()
if not name:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Name cannot be empty")
user.name = name
if payload.email is not None:
email = payload.email.strip().lower()
if not email or "@" not in email:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid email address")
existing = db.scalar(select(User).where(User.email == email, User.id != user.id))
if existing:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email is already in use")
user.email = email
if payload.role_id is not None:
role = _resolve_role(db, payload.role_id)
user.role_id = role.id if role else None
db.commit()
db.refresh(user)
log_security_event("users.updated", audience="internal", actor_user_id=actor.id, user_id=user.id)
return _serialize_user_read(user)
@router.post("/users/{user_id}/password", response_model=UserRead)
def set_user_password(
user_id: int,
payload: AdminSetPasswordRequest,
db: Session = Depends(get_db),
actor: User = Depends(require_permission("manage_users")), # gated by permission key
):
"""Set (reset) another user's password without their current password."""
user = _load_managed_user(db, user_id)
if len(payload.new_password) < 8:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="New password must be at least 8 characters",
)
user.password_hash = hash_password(payload.new_password)
db.commit()
db.refresh(user)
log_security_event("users.password_reset", audience="internal", actor_user_id=actor.id, user_id=user.id)
return _serialize_user_read(user)
@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(
user_id: int,
response: Response,
db: Session = Depends(get_db),
actor: User = Depends(require_permission("manage_users")), # gated by permission key
):
"""Delete a user. Lean owner accounts and your own account are protected."""
user = _load_managed_user(db, user_id)
if user.id == actor.id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="You cannot delete your own account",
)
if (user.role.name if user.role else "").lower() == LEAN_ROLE_NAME:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Lean accounts cannot be deleted",
)
db.delete(user)
db.commit()
log_security_event("users.deleted", audience="internal", actor_user_id=actor.id, user_id=user_id)
response.status_code = status.HTTP_204_NO_CONTENT
return None
@router.get("/roles", response_model=list[RoleRead])
def list_roles(
db: Session = Depends(get_db),
2026-06-17 21:55:04 +12:00
_: User = Depends(_require_role_management_actor),
):
2026-06-17 21:55:04 +12:00
user_counts = dict(
db.execute(select(User.role_id, func.count(User.id)).group_by(User.role_id)).all()
)
roles = db.scalars(
select(Role).options(selectinload(Role.permissions)).order_by(Role.name)
).all()
2026-06-17 21:55:04 +12:00
return [_serialize_role_read(role, user_count=user_counts.get(role.id, 0)) for role in roles]
@router.get("/role-modules", response_model=list[RoleModuleDefinition])
def list_role_modules(_: User = Depends(_require_role_management_actor)):
return _module_definitions_response()
@router.post("/roles", response_model=RoleRead, status_code=status.HTTP_201_CREATED)
def create_role(
payload: CreateRoleRequest,
db: Session = Depends(get_db),
actor: User = Depends(_require_role_management_actor),
):
name = payload.name.strip()
if not name:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Role name cannot be empty")
existing = db.scalar(select(Role).where(func.lower(Role.name) == name.lower()))
if existing:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Role name already exists")
permissions_by_key = {permission.key: permission for permission in db.scalars(select(Permission)).all()}
role = Role(name=name, description=None)
db.add(role)
db.flush()
_apply_role_updates(
role,
name=name,
description=payload.description,
module_permissions=payload.module_permissions,
permissions_by_key=permissions_by_key,
)
db.commit()
db.refresh(role)
log_security_event("roles.created", audience="internal", actor_user_id=actor.id, role_id=role.id)
return _serialize_role_read(role, user_count=0)
@router.patch("/roles/{role_id}", response_model=RoleRead)
def update_role(
role_id: int,
payload: UpdateRoleRequest,
db: Session = Depends(get_db),
actor: User = Depends(_require_role_management_actor),
):
role = _load_role(db, role_id)
original_name = role.name
protected = _is_protected_role_name(original_name)
requested_name = payload.name.strip() if payload.name is not None else role.name
if protected and requested_name.lower() != original_name.lower():
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Lean and admin roles cannot be renamed",
)
2026-06-17 21:55:04 +12:00
if payload.name is not None:
duplicate = db.scalar(
select(Role).where(func.lower(Role.name) == requested_name.lower(), Role.id != role.id)
)
if duplicate:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Role name already exists")
permissions_by_key = {permission.key: permission for permission in db.scalars(select(Permission)).all()}
_apply_role_updates(
role,
name=payload.name,
description=payload.description,
module_permissions=payload.module_permissions,
permissions_by_key=permissions_by_key,
)
db.commit()
db.refresh(role)
user_count = db.scalar(select(func.count(User.id)).where(User.role_id == role.id)) or 0
log_security_event("roles.updated", audience="internal", actor_user_id=actor.id, role_id=role.id)
return _serialize_role_read(role, user_count=user_count)
@router.delete("/roles/{role_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_role(
role_id: int,
response: Response,
db: Session = Depends(get_db),
actor: User = Depends(_require_role_management_actor),
):
role = _load_role(db, role_id)
if _is_protected_role_name(role.name):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Lean and admin roles cannot be deleted",
)
assigned_users = db.scalar(select(func.count(User.id)).where(User.role_id == role.id)) or 0
if assigned_users > 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Reassign users before deleting this role",
)
db.delete(role)
db.commit()
log_security_event("roles.deleted", audience="internal", actor_user_id=actor.id, role_id=role_id)
response.status_code = status.HTTP_204_NO_CONTENT
return None
@router.get("/permissions", response_model=list[str])
def list_permissions(
db: Session = Depends(get_db),
_: User = Depends(require_permission("manage_permissions")), # gated by permission key
):
return sorted(p.key for p in db.scalars(select(Permission)).all())