refactor: introduce declarative permission pattern for authorization

pyproject.toml

@@ -123,15 +123,13 @@ ignore = [
 [tool.ruff.lint.per-file-ignores]
 "src/app/infrastructure/persistence_sqla/alembic/**" = ["ALL", ]
 "tests/**" = [
+    "ARG002", # unused-method-argument
     "PLC2801", # unnecessary-dunder-call
     "PLR2004", # magic-value-comparison
     "PT011", # pytest-raises-too-broad
     "S101", # assert
-    "S105", # hardcoded-password-string
     "S106", # hardcoded-password-func-arg
     "S107", # hardcoded-password-default
-    "SLF001", # private-member-access
-    "UP012", # unnecessary-encode-utf8
 ]
 #
 "src/app/domain/value_objects/base.py" = ["B024", ]                          # abstract-base-class-without-abstract-method

src/app/application/commands/change_password.py

@@ -1,12 +1,19 @@
 import logging
 from dataclasses import dataclass
 
-from app.application.common.exceptions.authorization import AuthorizationError
 from app.application.common.ports.transaction_manager import (
     TransactionManager,
 )
 from app.application.common.ports.user_command_gateway import UserCommandGateway
-from app.application.common.services.authorization import AuthorizationService
+from app.application.common.services.authorization.authorize import (
+    authorize,
+)
+from app.application.common.services.authorization.composite import AnyOf
+from app.application.common.services.authorization.permissions import (
+    CanManageSelf,
+    CanManageSubordinate,
+    UserManagementContext,
+)
 from app.application.common.services.current_user import CurrentUserService
 from app.domain.entities.user import User
 from app.domain.exceptions.user import UserNotFoundByUsernameError
@@ -40,13 +47,11 @@ class ChangePasswordInteractor:
     def __init__(
         self,
         current_user_service: CurrentUserService,
-        authorization_service: AuthorizationService,
         user_command_gateway: UserCommandGateway,
         user_service: UserService,
         transaction_manager: TransactionManager,
     ):
         self._current_user_service = current_user_service
-        self._authorization_service = authorization_service
         self._user_command_gateway = user_command_gateway
         self._user_service = user_service
         self._transaction_manager = transaction_manager
@@ -65,16 +70,16 @@ async def __call__(self, request_data: ChangePasswordRequest) -> None:
         if user is None:
             raise UserNotFoundByUsernameError(username)
 
-        try:
-            self._authorization_service.authorize_for_self(
-                current_user.id_,
-                target_id=user.id_,
-            )
-        except AuthorizationError:
-            self._authorization_service.authorize_for_subordinate_role(
-                current_user.role,
-                target_role=user.role,
-            )
+        authorize(
+            AnyOf(
+                CanManageSelf(),
+                CanManageSubordinate(),
+            ),
+            context=UserManagementContext(
+                subject=current_user,
+                target=user,
+            ),
+        )
 
         self._user_service.change_password(user, password)
         await self._transaction_manager.commit()

src/app/application/commands/create_user.py

@@ -7,7 +7,13 @@
     TransactionManager,
 )
 from app.application.common.ports.user_command_gateway import UserCommandGateway
-from app.application.common.services.authorization import AuthorizationService
+from app.application.common.services.authorization.authorize import (
+    authorize,
+)
+from app.application.common.services.authorization.permissions import (
+    CanManageRole,
+    RoleManagementContext,
+)
 from app.application.common.services.current_user import CurrentUserService
 from app.domain.enums.user_role import UserRole
 from app.domain.exceptions.user import UsernameAlreadyExistsError
@@ -46,13 +52,11 @@ class CreateUserInteractor:
     def __init__(
         self,
         current_user_service: CurrentUserService,
-        authorization_service: AuthorizationService,
         user_command_gateway: UserCommandGateway,
         user_service: UserService,
         transaction_manager: TransactionManager,
     ):
         self._current_user_service = current_user_service
-        self._authorization_service = authorization_service
         self._user_command_gateway = user_command_gateway
         self._user_service = user_service
         self._transaction_manager = transaction_manager
@@ -64,9 +68,13 @@ async def __call__(self, request_data: CreateUserRequest) -> CreateUserResponse:
         )
 
         current_user = await self._current_user_service.get_current_user()
-        self._authorization_service.authorize_for_subordinate_role(
-            current_user.role,
-            target_role=request_data.role,
+
+        authorize(
+            CanManageRole(),
+            context=RoleManagementContext(
+                subject=current_user,
+                target_role=request_data.role,
+            ),
         )
 
         username = Username(request_data.username)

src/app/application/commands/grant_admin.py

@@ -5,7 +5,13 @@
     TransactionManager,
 )
 from app.application.common.ports.user_command_gateway import UserCommandGateway
-from app.application.common.services.authorization import AuthorizationService
+from app.application.common.services.authorization.authorize import (
+    authorize,
+)
+from app.application.common.services.authorization.permissions import (
+    CanManageRole,
+    RoleManagementContext,
+)
 from app.application.common.services.current_user import CurrentUserService
 from app.domain.entities.user import User
 from app.domain.enums.user_role import UserRole
@@ -38,13 +44,11 @@ class GrantAdminInteractor:
     def __init__(
         self,
         current_user_service: CurrentUserService,
-        authorization_service: AuthorizationService,
         user_command_gateway: UserCommandGateway,
         user_service: UserService,
         transaction_manager: TransactionManager,
     ):
         self._current_user_service = current_user_service
-        self._authorization_service = authorization_service
         self._user_command_gateway = user_command_gateway
         self._user_service = user_service
         self._transaction_manager = transaction_manager
@@ -56,9 +60,13 @@ async def __call__(self, request_data: GrantAdminRequest) -> None:
         )
 
         current_user = await self._current_user_service.get_current_user()
-        self._authorization_service.authorize_for_subordinate_role(
-            current_user.role,
-            target_role=UserRole.ADMIN,
+
+        authorize(
+            CanManageRole(),
+            context=RoleManagementContext(
+                subject=current_user,
+                target_role=UserRole.ADMIN,
+            ),
         )
 
         username = Username(request_data.username)

src/app/application/commands/inactivate_user.py

@@ -6,7 +6,15 @@
     TransactionManager,
 )
 from app.application.common.ports.user_command_gateway import UserCommandGateway
-from app.application.common.services.authorization import AuthorizationService
+from app.application.common.services.authorization.authorize import (
+    authorize,
+)
+from app.application.common.services.authorization.permissions import (
+    CanManageRole,
+    CanManageSubordinate,
+    RoleManagementContext,
+    UserManagementContext,
+)
 from app.application.common.services.current_user import CurrentUserService
 from app.domain.entities.user import User
 from app.domain.enums.user_role import UserRole
@@ -41,14 +49,12 @@ class InactivateUserInteractor:
     def __init__(
         self,
         current_user_service: CurrentUserService,
-        authorization_service: AuthorizationService,
         user_command_gateway: UserCommandGateway,
         user_service: UserService,
         transaction_manager: TransactionManager,
         access_revoker: AccessRevoker,
     ):
         self._current_user_service = current_user_service
-        self._authorization_service = authorization_service
         self._user_command_gateway = user_command_gateway
         self._user_service = user_service
         self._transaction_manager = transaction_manager
@@ -61,9 +67,13 @@ async def __call__(self, request_data: InactivateUserRequest) -> None:
         )
 
         current_user = await self._current_user_service.get_current_user()
-        self._authorization_service.authorize_for_subordinate_role(
-            current_user.role,
-            target_role=UserRole.USER,
+
+        authorize(
+            CanManageRole(),
+            context=RoleManagementContext(
+                subject=current_user,
+                target_role=UserRole.USER,
+            ),
         )
 
         username = Username(request_data.username)
@@ -74,9 +84,12 @@ async def __call__(self, request_data: InactivateUserRequest) -> None:
         if user is None:
             raise UserNotFoundByUsernameError(username)
 
-        self._authorization_service.authorize_for_subordinate_role(
-            current_user.role,
-            target_role=user.role,
+        authorize(
+            CanManageSubordinate(),
+            context=UserManagementContext(
+                subject=current_user,
+                target=user,
+            ),
         )
 
         self._user_service.toggle_user_activation(user, is_active=False)

src/app/application/commands/reactivate_user.py

@@ -5,7 +5,15 @@
     TransactionManager,
 )
 from app.application.common.ports.user_command_gateway import UserCommandGateway
-from app.application.common.services.authorization import AuthorizationService
+from app.application.common.services.authorization.authorize import (
+    authorize,
+)
+from app.application.common.services.authorization.permissions import (
+    CanManageRole,
+    CanManageSubordinate,
+    RoleManagementContext,
+    UserManagementContext,
+)
 from app.application.common.services.current_user import CurrentUserService
 from app.domain.entities.user import User
 from app.domain.enums.user_role import UserRole
@@ -39,13 +47,11 @@ class ReactivateUserInteractor:
     def __init__(
         self,
         current_user_service: CurrentUserService,
-        authorization_service: AuthorizationService,
         user_command_gateway: UserCommandGateway,
         user_service: UserService,
         transaction_manager: TransactionManager,
     ):
         self._current_user_service = current_user_service
-        self._authorization_service = authorization_service
         self._user_command_gateway = user_command_gateway
         self._user_service = user_service
         self._transaction_manager = transaction_manager
@@ -57,9 +63,13 @@ async def __call__(self, request_data: ReactivateUserRequest) -> None:
         )
 
         current_user = await self._current_user_service.get_current_user()
-        self._authorization_service.authorize_for_subordinate_role(
-            current_user.role,
-            target_role=UserRole.USER,
+
+        authorize(
+            CanManageRole(),
+            context=RoleManagementContext(
+                subject=current_user,
+                target_role=UserRole.USER,
+            ),
         )
 
         username = Username(request_data.username)
@@ -70,9 +80,12 @@ async def __call__(self, request_data: ReactivateUserRequest) -> None:
         if user is None:
             raise UserNotFoundByUsernameError(username)
 
-        self._authorization_service.authorize_for_subordinate_role(
-            current_user.role,
-            target_role=user.role,
+        authorize(
+            CanManageSubordinate(),
+            context=UserManagementContext(
+                subject=current_user,
+                target=user,
+            ),
         )
 
         self._user_service.toggle_user_activation(user, is_active=True)

src/app/application/commands/revoke_admin.py

@@ -5,7 +5,11 @@
     TransactionManager,
 )
 from app.application.common.ports.user_command_gateway import UserCommandGateway
-from app.application.common.services.authorization import AuthorizationService
+from app.application.common.services.authorization.authorize import authorize
+from app.application.common.services.authorization.permissions import (
+    CanManageRole,
+    RoleManagementContext,
+)
 from app.application.common.services.current_user import CurrentUserService
 from app.domain.entities.user import User
 from app.domain.enums.user_role import UserRole
@@ -38,13 +42,11 @@ class RevokeAdminInteractor:
     def __init__(
         self,
         current_user_service: CurrentUserService,
-        authorization_service: AuthorizationService,
         user_command_gateway: UserCommandGateway,
         user_service: UserService,
         transaction_manager: TransactionManager,
     ):
         self._current_user_service = current_user_service
-        self._authorization_service = authorization_service
         self._user_command_gateway = user_command_gateway
         self._user_service = user_service
         self._transaction_manager = transaction_manager
@@ -56,9 +58,13 @@ async def __call__(self, request_data: RevokeAdminRequest) -> None:
         )
 
         current_user = await self._current_user_service.get_current_user()
-        self._authorization_service.authorize_for_subordinate_role(
-            current_user.role,
-            target_role=UserRole.ADMIN,
+
+        authorize(
+            CanManageRole(),
+            context=RoleManagementContext(
+                subject=current_user,
+                target_role=UserRole.ADMIN,
+            ),
         )
 
         username = Username(request_data.username)