348 lines
12 KiB
Python
348 lines
12 KiB
Python
from app.modules.auth.schemas.access_schema import RoleAssignedOut
|
|
"""
|
|
Role Management Routes - RBAC endpoints for creating, updating, deleting roles.
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, Query, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional, List
|
|
from uuid import UUID
|
|
from pydantic import BaseModel
|
|
|
|
from app.db.database import get_db
|
|
from app.middleware.auth import get_current_user
|
|
from app.modules.auth.models.user_model import User
|
|
from app.modules.auth.controllers.role_controller import RoleController
|
|
from app.modules.auth.schemas.role_schema import (
|
|
RoleCreate,
|
|
RoleUpdate,
|
|
RoleOut,
|
|
RoleWithAccessesOut,
|
|
RolePaginatedOut,
|
|
)
|
|
from app.modules.auth.dependencies.access_dependency import require_access
|
|
from app.modules.auth.services.privilege import holds_superadmin_access
|
|
from app.modules.auth.services.user_role_service import UserRoleReader
|
|
from app.core.schemas import MessageOut
|
|
|
|
|
|
class AssignUserRoleIn(BaseModel):
|
|
user_id: int
|
|
role_id: UUID
|
|
|
|
|
|
router = APIRouter(prefix="/api/roles", tags=["Roles"])
|
|
|
|
|
|
def _is_superadmin(current_user: User) -> bool:
|
|
"""True when the caller holds any superadmin access code."""
|
|
return holds_superadmin_access(current_user)
|
|
|
|
|
|
def _assert_role_in_callers_tenant(role, current_user: User, db: Session = None, allow_system_roles: bool = False) -> None:
|
|
"""
|
|
Refuse to act on a role belonging to another tenant.
|
|
|
|
`RoleController.get_role_by_id` looks a role up by primary key alone, so
|
|
without this every by-id route was reachable across tenants — a tenant
|
|
administrator could read, rename and delete another tenant's roles by
|
|
guessing or observing a UUID. Raises 404 rather than 403 so the response
|
|
does not confirm that the role exists.
|
|
"""
|
|
if _is_superadmin(current_user):
|
|
return
|
|
if role.tenant_id == current_user.tenant_id:
|
|
return
|
|
|
|
if allow_system_roles:
|
|
if getattr(role, "tenant_id", None) is None and getattr(role, "name", "").lower() != "superadmin":
|
|
return
|
|
if getattr(role, "is_system", False) and current_user.tenant_id and db:
|
|
from app.modules.billing.models.plan_model import TenantSubscription, PlanRole
|
|
sub = db.query(TenantSubscription).filter(
|
|
TenantSubscription.tenant_id == current_user.tenant_id,
|
|
TenantSubscription.status == 'active'
|
|
).order_by(TenantSubscription.created_at.desc()).first()
|
|
if sub and sub.plan_id:
|
|
has_role = db.query(PlanRole).filter(
|
|
PlanRole.plan_id == sub.plan_id,
|
|
PlanRole.role_id == role.id
|
|
).first()
|
|
if has_role:
|
|
return
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Role not found"
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/", response_model=RoleOut, dependencies=[require_access("admin.role.create")]
|
|
)
|
|
def create_role(
|
|
payload: RoleCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Create a new role.
|
|
|
|
Required access: admin.role.create | superadmin.role.create
|
|
|
|
- **role_name**: Name of the role
|
|
- **description**: Optional description
|
|
- **access_ids**: List of access IDs to assign
|
|
- **tenant_id**: Tenant ID (for tenant-specific roles)
|
|
"""
|
|
if not payload.tenant_id and current_user.tenant_id:
|
|
payload.tenant_id = current_user.tenant_id
|
|
|
|
result = RoleController.create_role(db, payload)
|
|
try:
|
|
from app.modules.activity_logs.service import log_event
|
|
from app.modules.activity_logs.constants import ActivityLogModule, ActivityLogAction, ActivityLogStatus, ActivityLogTargetType
|
|
if payload.tenant_id:
|
|
log_event(
|
|
tenant_id=payload.tenant_id,
|
|
user_id=current_user.id,
|
|
user_email=current_user.email,
|
|
module=ActivityLogModule.TENANT,
|
|
action=ActivityLogAction.ROLE_CREATED,
|
|
target_id=str(result.id),
|
|
target_type=ActivityLogTargetType.ROLE,
|
|
metadata={
|
|
"role_name": payload.role_name
|
|
},
|
|
status=ActivityLogStatus.SUCCESS
|
|
)
|
|
except Exception as e:
|
|
import logging
|
|
logging.getLogger(__name__).error(f"Failed to log ROLE_CREATED: {e}")
|
|
|
|
return result
|
|
|
|
|
|
@router.get(
|
|
"/", response_model=List[RoleOut], dependencies=[require_access("admin.role.read")]
|
|
)
|
|
def get_all_roles(
|
|
tenant_id: Optional[UUID] = Query(None),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get all roles.
|
|
|
|
Required access: admin.role.read | superadmin.role.read
|
|
|
|
Superadmins see roles across all tenants by default, or a single
|
|
tenant when `?tenant_id=<uuid>` is provided. Tenant admins are
|
|
always scoped to their own tenant regardless of the query param.
|
|
"""
|
|
effective_tenant_id = (
|
|
tenant_id if _is_superadmin(current_user) else current_user.tenant_id
|
|
)
|
|
return RoleController.get_all_roles(db, effective_tenant_id)
|
|
|
|
|
|
@router.get(
|
|
"/{role_id}",
|
|
response_model=RoleOut,
|
|
dependencies=[require_access("admin.role.read")],
|
|
)
|
|
def get_role(
|
|
role_id: UUID,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get a specific role by ID.
|
|
|
|
Required access: admin.role.read | superadmin.role.read
|
|
"""
|
|
role = RoleController.get_role_by_id(db, role_id)
|
|
_assert_role_in_callers_tenant(role, current_user, db, allow_system_roles=True)
|
|
return role
|
|
|
|
|
|
@router.get(
|
|
"/{role_id}/details",
|
|
response_model=RoleWithAccessesOut,
|
|
dependencies=[require_access("admin.role.read")],
|
|
)
|
|
def get_role_with_accesses(
|
|
role_id: UUID,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get a role with all assigned access codes.
|
|
|
|
Required access: admin.role.read | superadmin.role.read
|
|
|
|
Also reports how many principals hold it, so the editing screen can say what
|
|
a change affects. Counted here and not on the list endpoints, which render
|
|
up to a hundred roles.
|
|
"""
|
|
_assert_role_in_callers_tenant(
|
|
RoleController.get_role_by_id(db, role_id), current_user, db, allow_system_roles=True
|
|
)
|
|
role = RoleController.get_role_with_accesses(db, role_id)
|
|
out = RoleWithAccessesOut.model_validate(role)
|
|
out.assigned_user_count = UserRoleReader(db).users_holding_role(role_id)
|
|
return out
|
|
|
|
|
|
@router.patch(
|
|
"/{role_id}",
|
|
response_model=RoleOut,
|
|
dependencies=[require_access("admin.role.update")],
|
|
)
|
|
def update_role(
|
|
role_id: UUID,
|
|
payload: RoleUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Update a role's details and accesses.
|
|
|
|
Required access: admin.role.update | superadmin.role.update
|
|
|
|
- **role_name**: New role name (optional)
|
|
- **description**: New description (optional)
|
|
- **access_ids**: New list of access IDs (optional)
|
|
"""
|
|
role = RoleController.get_role_by_id(db, role_id)
|
|
_assert_role_in_callers_tenant(role, current_user)
|
|
if role.is_default and current_user.tenant_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Default roles cannot be modified by tenant admins."
|
|
)
|
|
return RoleController.update_role(db, role_id, payload)
|
|
|
|
|
|
@router.delete("/{role_id}", dependencies=[require_access("admin.role.update")], response_model=MessageOut)
|
|
def delete_role(
|
|
role_id: UUID,
|
|
force: bool = Query(
|
|
False,
|
|
description=(
|
|
"Delete even though the role is still held. Required once anybody "
|
|
"holds it, so the removal of their authority is a decision rather "
|
|
"than a side effect."
|
|
),
|
|
),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Delete a role.
|
|
|
|
Required access: admin.role.update | superadmin.role.delete
|
|
|
|
**Refuses with 409 while the role is still held**, unless `?force=true`.
|
|
|
|
Deleting a role has always taken authority away silently: `user_roles.role_id`
|
|
is `ON DELETE CASCADE`, so every grant of it disappears, and `users.role_id`
|
|
is `ON DELETE SET NULL`, so every holder loses their primary role. Neither
|
|
leaves a trace and neither was ever counted for the operator. With one role
|
|
per user that was survivable; with roles granted in several places it is
|
|
not, because the operator can no longer hold the affected set in their head.
|
|
|
|
The cascade behaviour is unchanged. Only the silence is.
|
|
"""
|
|
role = RoleController.get_role_by_id(db, role_id)
|
|
_assert_role_in_callers_tenant(role, current_user)
|
|
if role.is_default and current_user.tenant_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Default roles cannot be deleted by tenant admins."
|
|
)
|
|
|
|
if not force:
|
|
holders = UserRoleReader(db).users_holding_role(role_id)
|
|
if holders:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=(
|
|
f"This role cannot be deleted because there are currently {holders} "
|
|
f"active {'assignment' if holders == 1 else 'assignments'} of it. "
|
|
"Please reassign or remove them first."
|
|
),
|
|
)
|
|
|
|
return RoleController.delete_role(db, role_id)
|
|
|
|
|
|
@router.get(
|
|
"/tenant/{tenant_id}",
|
|
response_model=List[RoleOut],
|
|
dependencies=[require_access("admin.role.read")],
|
|
)
|
|
def get_roles_by_tenant(
|
|
tenant_id: UUID,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get all roles for a specific tenant.
|
|
|
|
The tenant is named by the caller, so a tenant administrator must not be
|
|
able to name somebody else's.
|
|
"""
|
|
if not _is_superadmin(current_user) and tenant_id != current_user.tenant_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found"
|
|
)
|
|
return RoleController.get_all_roles(db, tenant_id)
|
|
|
|
|
|
@router.post(
|
|
"/assign-user",
|
|
dependencies=[require_access("admin.role.assign")],
|
|
response_model=RoleAssignedOut)
|
|
def assign_role_to_user(
|
|
payload: AssignUserRoleIn,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Assign a role to a user."""
|
|
return RoleController.assign_role_to_user(db, payload.user_id, payload.role_id)
|
|
|
|
|
|
@router.get(
|
|
"",
|
|
response_model=RolePaginatedOut,
|
|
dependencies=[require_access("admin.role.read")],
|
|
)
|
|
def list_roles_paginated(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(10, ge=1, le=100),
|
|
search: Optional[str] = Query(None),
|
|
tenant_id: Optional[UUID] = Query(None),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get paginated list of roles with optional search.
|
|
|
|
Required access: admin.role.read | superadmin.role.read
|
|
|
|
Superadmins see roles across all tenants by default, or a single
|
|
tenant when `?tenant_id=<uuid>` is provided. Tenant admins are
|
|
always scoped to their own tenant regardless of the query param.
|
|
"""
|
|
effective_tenant_id = (
|
|
tenant_id if _is_superadmin(current_user) else current_user.tenant_id
|
|
)
|
|
|
|
return RoleController.get_roles_paginated(
|
|
db=db,
|
|
tenant_id=effective_tenant_id,
|
|
page=page,
|
|
page_size=page_size,
|
|
search=search,
|
|
)
|