feat: granular RBAC system with user groups, deny-list, permission registry, system-admin, self-mod prevention

app/deps.py

@@ -25,7 +25,9 @@ async def get_current_user(
     redis: aioredis.Redis = Depends(get_redis_dep),
 ) -> dict[str, Any]:
     """Get the current authenticated user from session cookie.
-    Returns session data dict with user_id, tenant_id, email, name, role.
+
+    Returns session data dict with user_id, tenant_id, email, name, role,
+    and resolved permissions from Redis cache.
     """
     settings = get_settings()
     session_id = request.cookies.get(settings.session_cookie_name)
@@ -52,32 +54,107 @@ async def get_current_user(
     tenant_id = uuid.UUID(session_data["tenant_id"])
     await set_tenant_context(db, tenant_id)
 
+    # Load resolved permissions from cache (or DB on miss)
+    from app.core.permissions import get_cached_permissions
+
+    user_id = uuid.UUID(session_data["user_id"])
+    resolved = await get_cached_permissions(db, redis, user_id, tenant_id)
+    session_data["permissions"] = resolved.get("permissions", [])
+    session_data["denied_permissions"] = resolved.get("denied", [])
+    session_data["field_permissions"] = resolved.get("field_permissions", {})
+    session_data["is_system_admin"] = resolved.get("is_system_admin", False)
+
     return session_data
 
 
 async def require_admin(
     current_user: dict[str, Any] = Depends(get_current_user),
 ) -> dict[str, Any]:
-    """Require admin role."""
-    if current_user.get("role") != "admin":
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail={"detail": "Admin access required", "code": "forbidden"},
-        )
-    return current_user
+    """Require admin role (legacy + new permission system)."""
+    if current_user.get("is_system_admin") or current_user.get("role") == "admin":
+        return current_user
+    # Also check via permission system
+    from app.core.permissions import check_permission
+
+    if check_permission(current_user, "*:*"):
+        return current_user
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail={"detail": "Admin access required", "code": "forbidden"},
+    )
 
 
 async def require_write(
     current_user: dict[str, Any] = Depends(get_current_user),
 ) -> dict[str, Any]:
-    """Require write permission (admin or editor)."""
+    """Require write permission (admin, editor, or custom role with write perms)."""
+    if current_user.get("is_system_admin"):
+        return current_user
     role = current_user.get("role", "viewer")
-    if role not in ("admin", "editor"):
+    if role in ("admin", "editor"):
+        return current_user
+    # Check via permission system for custom roles
+    from app.core.permissions import check_permission
+
+    if check_permission(current_user, "*:write") or check_permission(current_user, "*:create"):
+        return current_user
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail={"detail": "Write access required", "code": "forbidden"},
+    )
+
+
+def require_permission(permission: str):
+    """FastAPI dependency factory: require a specific permission.
+
+    Usage:
+        @router.get("/companies", dependencies=[Depends(require_permission("companies:read"))])
+    """
+    async def _check(
+        current_user: dict[str, Any] = Depends(get_current_user),
+    ) -> dict[str, Any]:
+        if current_user.get("is_system_admin"):
+            return current_user
+        from app.core.permissions import check_permission
+
+        if check_permission(current_user, permission):
+            return current_user
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
-            detail={"detail": "Write access required", "code": "forbidden"},
+            detail={
+                "detail": f"Permission '{permission}' required",
+                "code": "forbidden",
+            },
         )
-    return current_user
+
+    return _check
+
+
+def require_field_access(module: str, field: str, default: str = "read"):
+    """FastAPI dependency factory: require field-level access.
+
+    Usage:
+        @router.get("/contacts/{id}", dependencies=[Depends(require_field_access("contacts", "annual_revenue"))])
+    """
+    async def _check(
+        current_user: dict[str, Any] = Depends(get_current_user),
+    ) -> dict[str, Any]:
+        if current_user.get("is_system_admin"):
+            return current_user
+        from app.core.permissions import check_field_access
+
+        access = check_field_access(current_user, module, field, default)
+        if access == "hidden":
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail={
+                    "detail": f"Field '{field}' is hidden",
+                    "code": "field_hidden",
+                },
+            )
+        return current_user
+
+    return _check
 
 
 async def get_tenant_id(