Phase 3 harden CSRF invalid POST and work access handling

app/modules/core/rbac/ui.py

@@ -2,6 +2,7 @@ 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
@@ -17,6 +18,12 @@ from app.modules.core.rbac.services import build_permissions_payload, build_role
 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)
 
@@ -47,6 +54,15 @@ 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()
@@ -76,7 +92,10 @@ def role_create_submit(
     is_active: str | None = Form(None),
     csrf_token: str = Form(...),
 ):
-    validate_csrf(request, csrf_token)
+    try:
+        validate_csrf(request, csrf_token)
+    except PermissionError:
+        return _csrf_rejected(request)
     db = CommonSessionLocal()
     try:
         current_user = get_current_user(request, db=db)
@@ -124,7 +143,7 @@ def role_permissions_edit(request: Request, role_id: int):
 
         role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
         if not role:
-            return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
+            return not_found_response(request, "Role not found")
 
         scope = build_scope(db, current_user)
         try:
@@ -160,7 +179,10 @@ def role_permissions_submit(
     permission_ids: list[int] = Form([]),
     csrf_token: str = Form(...),
 ):
-    validate_csrf(request, csrf_token)
+    try:
+        validate_csrf(request, csrf_token)
+    except PermissionError:
+        return _csrf_rejected(request)
     db = CommonSessionLocal()
     try:
         current_user = get_current_user(request, db=db)
@@ -172,7 +194,7 @@ def role_permissions_submit(
 
         role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
         if not role:
-            return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
+            return not_found_response(request, "Role not found")
 
         old_permission_ids = db.execute(
             select(RolePermission.permission_id).where(RolePermission.role_id == role.id)
@@ -233,7 +255,10 @@ def permission_create_submit(
     is_active: str | None = Form(None),
     csrf_token: str = Form(...),
 ):
-    validate_csrf(request, csrf_token)
+    try:
+        validate_csrf(request, csrf_token)
+    except PermissionError:
+        return _csrf_rejected(request)
     db = CommonSessionLocal()
     try:
         current_user = get_current_user(request, db=db)
@@ -265,4 +290,4 @@ def permission_create_submit(
 
         return RedirectResponse(url="/system-settings/rbac/permissions", status_code=303)
     finally:
-        db.close()
+        db.close()