arrr-erp/app/modules/core/rbac/ui.py

from __future__ import annotations

from fastapi import APIRouter, Form, Request
from fastapi.responses import RedirectResponse
from app.core.http_responses import not_found_response
from sqlalchemy import select

from app.core.db.common import CommonSessionLocal
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
from app.core.security.session_auth import get_current_user
from app.core.templating import templates
from app.modules.core.audit.service import write_audit_log
from app.modules.core.iam.scope import build_scope, assert_can_manage_role_object
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
from app.modules.core.rbac.models import Permission, Role, RolePermission
from app.modules.core.rbac.services import build_permissions_payload, build_roles_payload

router = APIRouter(prefix="/system-settings/rbac", tags=["rbac-ui"])


def _csrf_rejected(request: Request):
    from app.core.http_responses import forbidden_response
    return forbidden_response(request, "CSRF validation failed")



def _redirect_login():
    return RedirectResponse(url="/login", status_code=303)


def _redirect_denied():
    from app.core.http_responses import ui_access_denied
    return ui_access_denied()


def _is_system_admin(db, current_user) -> bool:
    return "System Admin" in get_user_roles(db, current_user.id)


def _base_ctx(request: Request, current_user, db, **ctx):
    base = {
        "request": request,
        "current_user": current_user,
        "current_user_roles": get_user_roles(db, current_user.id),
        "current_user_permissions": get_user_permissions(db, current_user.id),
        "csrf_token": get_or_create_csrf_token(request),
    }
    base.update(ctx)
    return base


@router.get("")
def rbac_dashboard(request: Request):
    return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)




# Phase 3 security hardening: the roles list endpoint must not accept direct unsafe POSTs.
# Existing role creation continues to use /roles/new.
@router.post("/roles")
def roles_list_post_rejected(request: Request):
    return _csrf_rejected(request)


@router.get("/roles")
def roles_list(request: Request, q: str = "", page: int = 1, per_page: int = 10):
    db = CommonSessionLocal()
    try:
        current_user = get_current_user(request, db=db)
        if not current_user:
            return _redirect_login()

        if not _is_system_admin(db, current_user):
            return _redirect_denied()

        scope = build_scope(db, current_user)
        payload = build_roles_payload(db, scope, q=q, page=page, per_page=per_page)

        return templates.TemplateResponse(
            "modules/core/rbac/templates/roles_list.html",
            _base_ctx(request, current_user, db, title="RBAC Roles", **payload),
        )
    finally:
        db.close()


@router.post("/roles/new")
def role_create_submit(
    request: Request,
    name: str = Form(...),
    is_active: str | None = Form(None),
    csrf_token: str = Form(...),
):
    try:
        validate_csrf(request, csrf_token)
    except PermissionError:
        return _csrf_rejected(request)
    db = CommonSessionLocal()
    try:
        current_user = get_current_user(request, db=db)
        if not current_user:
            return _redirect_login()

        if not _is_system_admin(db, current_user):
            return _redirect_denied()

        name = name.strip()
        if name and not db.execute(select(Role).where(Role.name == name)).scalar_one_or_none():
            role = Role(name=name, is_active=is_active is not None)
            db.add(role)
            db.commit()
            db.refresh(role)

            write_audit_log(
                db,
                action="role.create",
                entity_type="role",
                actor=current_user,
                request=request,
                entity_id=role.id,
                entity_name=role.name,
                target_tenant_id=current_user.tenant_id,
                target_branch_id=current_user.branch_id,
                details={"name": role.name, "is_active": role.is_active},
            )

        return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
    finally:
        db.close()


@router.get("/roles/{role_id}")
def role_permissions_edit(request: Request, role_id: int):
    db = CommonSessionLocal()
    try:
        current_user = get_current_user(request, db=db)
        if not current_user:
            return _redirect_login()

        if not _is_system_admin(db, current_user):
            return _redirect_denied()

        role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
        if not role:
            return not_found_response(request, "Role not found")

        scope = build_scope(db, current_user)
        try:
            assert_can_manage_role_object(scope, role)
        except Exception:
            return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)

        permissions = db.execute(select(Permission).order_by(Permission.code)).scalars().all()
        assigned_permission_ids = db.execute(
            select(RolePermission.permission_id).where(RolePermission.role_id == role_id)
        ).scalars().all()

        return templates.TemplateResponse(
            "modules/core/rbac/templates/role_permissions_form.html",
            _base_ctx(
                request,
                current_user,
                db,
                title=f"Role Permissions - {role.name}",
                role=role,
                permissions=permissions,
                assigned_permission_ids=list(assigned_permission_ids),
            ),
        )
    finally:
        db.close()


@router.post("/roles/{role_id}")
def role_permissions_submit(
    request: Request,
    role_id: int,
    permission_ids: list[int] = Form([]),
    csrf_token: str = Form(...),
):
    try:
        validate_csrf(request, csrf_token)
    except PermissionError:
        return _csrf_rejected(request)
    db = CommonSessionLocal()
    try:
        current_user = get_current_user(request, db=db)
        if not current_user:
            return _redirect_login()

        if not _is_system_admin(db, current_user):
            return _redirect_denied()

        role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
        if not role:
            return not_found_response(request, "Role not found")

        old_permission_ids = db.execute(
            select(RolePermission.permission_id).where(RolePermission.role_id == role.id)
        ).scalars().all()

        db.execute(RolePermission.__table__.delete().where(RolePermission.role_id == role.id))
        for permission_id in permission_ids:
            db.add(RolePermission(role_id=role.id, permission_id=permission_id))
        db.commit()

        write_audit_log(
            db,
            action="role.permissions.update",
            entity_type="role",
            actor=current_user,
            request=request,
            entity_id=role.id,
            entity_name=role.name,
            target_tenant_id=current_user.tenant_id,
            target_branch_id=current_user.branch_id,
            details={
                "old_permission_ids": list(old_permission_ids),
                "new_permission_ids": list(permission_ids),
            },
        )

        return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
    finally:
        db.close()


@router.get("/permissions")
def permissions_list(request: Request, q: str = "", page: int = 1, per_page: int = 15):
    db = CommonSessionLocal()
    try:
        current_user = get_current_user(request, db=db)
        if not current_user:
            return _redirect_login()

        if not _is_system_admin(db, current_user):
            return _redirect_denied()

        payload = build_permissions_payload(db, q=q, page=page, per_page=per_page)

        return templates.TemplateResponse(
            "modules/core/rbac/templates/permissions_list.html",
            _base_ctx(request, current_user, db, title="RBAC Permissions", **payload),
        )
    finally:
        db.close()


@router.post("/permissions/new")
def permission_create_submit(
    request: Request,
    code: str = Form(...),
    name: str = Form(...),
    is_active: str | None = Form(None),
    csrf_token: str = Form(...),
):
    try:
        validate_csrf(request, csrf_token)
    except PermissionError:
        return _csrf_rejected(request)
    db = CommonSessionLocal()
    try:
        current_user = get_current_user(request, db=db)
        if not current_user:
            return _redirect_login()

        if not _is_system_admin(db, current_user):
            return _redirect_denied()

        code = code.strip()
        if code and not db.execute(select(Permission).where(Permission.code == code)).scalar_one_or_none():
            permission = Permission(code=code, name=name.strip(), is_active=is_active is not None)
            db.add(permission)
            db.commit()
            db.refresh(permission)

            write_audit_log(
                db,
                action="permission.create",
                entity_type="permission",
                actor=current_user,
                request=request,
                entity_id=permission.id,
                entity_name=permission.code,
                target_tenant_id=current_user.tenant_id,
                target_branch_id=current_user.branch_id,
                details={"code": permission.code, "name": permission.name, "is_active": permission.is_active},
            )

        return RedirectResponse(url="/system-settings/rbac/permissions", status_code=303)
    finally:
        db.close()