feat: permissions added to controllers

src/controllers/_helpers.py

@@ -1,9 +1,9 @@
-import jwt
 from typing import Annotated, Any
 from fastapi import Depends, HTTPException
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 
 from src.services.userService import UserService
+from src.infra.database.database import PgDatabase
 from src.infra.security.hashing import ALGORITHM, JWT_ACCESS_SECRET_KEY, decode_token
 
 oauth2_bearer = OAuth2PasswordBearer(tokenUrl="auth/login")
@@ -22,4 +22,27 @@ def get_current_user(token: token_dependency) -> dict[str, Any]:
     return user
 
 user_dependency = Annotated[dict[str, Any], Depends(get_current_user)]
-form_auth_dependency = Annotated[OAuth2PasswordRequestForm, Depends()]
+form_auth_dependency = Annotated[OAuth2PasswordRequestForm, Depends()]
+
+class PermissionChecker():
+    def __init__(self, required_permission: str):
+        self.required_permission = required_permission
+
+    def __call__(self, user: user_dependency):
+        controller, action = self.required_permission.split("-")
+        user_type_id = user["tipo_usuario_id"]
+
+        query = """
+            SELECT regras.acao, regras.permitir, controllers.nome, tipo_usuario.nome 
+            FROM regras
+            JOIN controllers ON controllers.id = regras.controller_id
+            JOIN tipo_usuario ON tipo_usuario.id = regras.tipo_usuario_id
+            WHERE tipo_usuario.id = %s AND regras.permitir = True AND controllers.nome = %s AND regras.acao = %s
+        """
+
+        with PgDatabase() as db:
+            db.cursor.execute(query, (user_type_id, controller, action))
+            row = db.cursor.fetchone()
+
+            if row is None:
+                raise HTTPException(status_code=403, detail={"message": "Usuário não autorizado", "error": True})

src/controllers/userController.py

@@ -1,5 +1,6 @@
-from fastapi import APIRouter, Request
+from fastapi import APIRouter, Depends, Request
 
+from ._helpers import PermissionChecker
 from src.services.userService import UserService
 from src.schemas.userSchema import UserAddSchema, UserEditSchema
 
@@ -12,22 +13,36 @@ def user_all(
     page: int = 1,
     rows_per_page: int = 10,
     sort_by: str | None = None,
-    show_fk_id: int | None = 1
+    show_fk_id: int | None = 1,
+    perms = Depends(PermissionChecker("usuario-all"))
 ):
     return UserService().all(request.query_params)
 
 @router.get("/{user_id:int}")
-def user_view(user_id: int):
+def user_view(
+    user_id: int, 
+    perms = Depends(PermissionChecker("usuario-view"))
+):
     return UserService().view(user_id)
 
 @router.post("/")
-def user_add(user: UserAddSchema):
+def user_add(
+    user: UserAddSchema,
+    perms = Depends(PermissionChecker("usuario-add"))
+):
     return UserService().add(user)
 
 @router.put("/{user_id:int}")
-def user_edit(user_id: int, user: UserEditSchema):
+def user_edit(
+    user_id: int, 
+    user: UserEditSchema,
+    perms = Depends(PermissionChecker("usuario-edit"))
+):
     return UserService().edit(user_id, user)
 
 @router.delete("/{user_id:int}")
-def user_delete(user_id: int):
+def user_delete(
+    user_id: int,
+    perms = Depends(PermissionChecker("usuario-delete"))
+):
     return UserService().delete(user_id)

src/controllers/userTypeController.py

@@ -1,5 +1,6 @@
-from fastapi import APIRouter, Request
+from fastapi import APIRouter, Depends, Request
 
+from ._helpers import PermissionChecker
 from src.schemas.userTypeSchema import UserTypeSchema
 from src.services.userTypeService import UserTypeService
 
@@ -10,26 +11,40 @@ def user_type_all(
     request: Request,
     page: int = 1,
     rows_per_page: int = 10,
-    sort_by: str | None = None
+    sort_by: str | None = None,
+    perms = Depends(PermissionChecker("tipo_usuario-all"))
 ):
     return UserTypeService().all(request.query_params)
 
 @router.get("/{user_type_id:int}")
-def user_type_view(user_type_id: int):
+def user_type_view(
+    user_type_id: int,
+    perms = Depends(PermissionChecker("tipo_usuario-view"))
+):
     response = UserTypeService().view(user_type_id)
     return response
 
 @router.post("/")
-def user_type_add(user_type: UserTypeSchema):
+def user_type_add(
+    user_type: UserTypeSchema,
+    perms = Depends(PermissionChecker("tipo_usuario-add"))
+):
     response = UserTypeService().add(user_type)
     return response
 
 @router.put("/{user_type_id:int}")
-def user_type_edit(user_type_id: int, user_type: UserTypeSchema):
+def user_type_edit(
+    user_type_id: int, 
+    user_type: UserTypeSchema,
+    perms = Depends(PermissionChecker("tipo_usuario-edit"))
+):
     response = UserTypeService().edit(user_type_id, user_type)
     return response
 
 @router.delete("/{user_type_id:int}")
-def user_type_delete(user_type_id: int):
+def user_type_delete(
+    user_type_id: int,
+    perms = Depends(PermissionChecker("tipo_usuario-delete"))
+):
     response = UserTypeService().delete(user_type_id)
     return response

src/infra/database/populate.py

@@ -19,6 +19,7 @@ def insert(filePath: str) -> None:
             db.connection.commit()
 
     insert("/scripts/tables.sql")
+    insert("/scripts/rules.sql")
 
 
 if __name__ == "__main__":

src/infra/database/scripts/rules.sql

@@ -0,0 +1,25 @@
+-- Tipo Usuário
+INSERT INTO tipo_usuario (id, nome) VALUES (1, 'admin');
+
+-- Usuário
+-- senha: 1234
+INSERT INTO usuario (id, email, senha, tipo_usuario_id) VALUES (1, "admin@email.com", "$6$xvhoMyeu7TlurX6Z$.j4NQGAvbEzu3IKSJAWU8IzTqAGVW8RZ2sUWmJSzJY2QR0VJhm26rOC/0.7UXT3c8YrwoS3y4pUdbWJV7GEZG1", 1)
+
+-- Controllers
+INSERT INTO controllers (id, nome) VALUES (1, 'usuario');
+INSERT INTO controllers (id, nome) VALUES (2, 'tipo_usuario');
+
+-- Admin Rules
+-- admin + usuario controller
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (1, 1, 1, 'all', True);
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (2, 1, 1, 'view', True);
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (3, 1, 1, 'add', True);
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (4, 1, 1, 'edit', True);
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (5, 1, 1, 'delete', True);
+
+-- admin + tipo_usuario controller
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (6, 2, 1, 'all', True);
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (7, 2, 1, 'view', True);
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (8, 2, 1, 'add', True);
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (9, 2, 1, 'edit', True);
+INSERT INTO regras (id, controller_id, tipo_usuario_id, acao, permitir) VALUES (10, 2, 1, 'delete', True);

src/infra/database/scripts/tables.sql

@@ -9,6 +9,24 @@ CREATE TABLE IF NOT EXISTS usuario (
     senha VARCHAR(200) NOT NULL,
     tipo_usuario_id INTEGER NOT NULL,
 
+    CONSTRAINT fk_tipo_usuario FOREIGN KEY (tipo_usuario_id)
+    REFERENCES tipo_usuario (id)
+);
+
+CREATE TABLE IF NOT EXISTS controllers (
+    id SERIAL PRIMARY KEY,
+    nome VARCHAR(50) UNIQUE NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS regras (
+    id SERIAL PRIMARY KEY,
+    acao VARCHAR(50) NOT NULL,
+    permitir BOOLEAN NOT NULL,
+    controller_id INTEGER NOT NULL,
+    tipo_usuario_id INTEGER NOT NULL,
+
+    CONSTRAINT fk_controller FOREIGN KEY (controller_id)
+    REFERENCES controllers (id),
     CONSTRAINT fk_tipo_usuario FOREIGN KEY (tipo_usuario_id)
     REFERENCES tipo_usuario (id)
 );

src/services/userService.py

@@ -17,7 +17,7 @@ def __init__(self) -> None:
         try:
             self.all_columns = reduce(lambda acc, elem: acc + ", " + str(elem), self.columns)
         except Exception:
-            self.all_columns = ""
+            self.all_columns = "*"
 
     def all(self, query_params: QueryParams) -> dict[str, Any]:
         show_fk_id = bool(int(query_params.get("show_fk_id", 1)))

src/services/userTypeService.py

@@ -19,7 +19,7 @@ def __init__(self) -> None:
         try:
             self.all_columns = reduce(lambda acc, elem: acc + ", " + str(elem), self.columns)
         except Exception:
-            self.all_columns = ""
+            self.all_columns = "*"
 
     def all(self, query_params: QueryParams) -> dict[str, Any]:
         query = f"SELECT {self.all_columns} FROM {self.table}"