Add PermissionChecker for routers (RBAC)

src/api/dependencies.py

@@ -5,6 +5,7 @@
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
 
+from src.core.security_policy import ROLE_PERMISSIONS, Action
 from src.domain.models.user import User as DomainUser
 from src.domain.repositories.user_repo import IUserRepository
 from src.infrastructure.security.jwt import decode_access_token
@@ -44,13 +45,17 @@ async def get_current_user(
     return user
 
 
-async def get_current_admin_user(
-    current_user: DomainUser = Depends(get_current_user),
-) -> DomainUser:
-    """Check that current user has an admin role."""
-    if current_user.role != "admin":
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Admin role needed.",
-        )
-    return current_user
+class PermissionChecker:
+
+    """Class for user's access validation (RBAC)."""
+
+    def __init__(self, required_action: Action):
+        self.required_action = required_action
+
+    def __call__(self, user: DomainUser = Depends(get_current_user)) -> None:
+        """Check user's access."""
+        if self.required_action not in ROLE_PERMISSIONS[user.role]:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"Access denied. Required permission: {self.required_action.value}"
+            )

src/api/routers/admin.py

@@ -4,29 +4,39 @@
 from fastapi import APIRouter, Depends, status
 from pydantic import UUID4
 
-from src.api.dependencies import get_current_admin_user
+from src.api.dependencies import PermissionChecker
 from src.api.schemas.admin import RoleCreate, RoleResponse, UserResponse, UserRoleUpdate
 from src.application.services.admin_service import AdminService
+from src.core.security_policy import Action
 
 router_admin = APIRouter(
-    dependencies=[Depends(get_current_admin_user)],
+    dependencies=[Depends(PermissionChecker(Action.ADMIN_ACCESS))],
     route_class=DishkaRoute
 )
 
 
-@router_admin.get("/users", response_model=List[UserResponse])
+@router_admin.get(
+    "/users",
+    response_model=List[UserResponse]
+)
 async def get_all_users(admin_service: FromDishka[AdminService]):
     """Get all users from database."""
     return await admin_service.get_all_users()
 
 
-@router_admin.get("/roles", response_model=List[RoleResponse])
+@router_admin.get(
+    "/roles",
+    response_model=List[RoleResponse]
+)
 async def get_all_roles(admin_service: FromDishka[AdminService]):
     """Get all roles from database."""
     return await admin_service.get_all_roles()
 
 
-@router_admin.put("/users/{user_id}/role", response_model=UserResponse)
+@router_admin.put(
+    "/users/{user_id}/role",
+    response_model=UserResponse
+)
 async def update_user_role(
     user_id: UUID4,
     role_update: UserRoleUpdate,
@@ -36,7 +46,10 @@ async def update_user_role(
     return await admin_service.update_user_role(user_id, role_update.role)
 
 
-@router_admin.post("/roles", status_code=status.HTTP_201_CREATED)
+@router_admin.post(
+    "/roles",
+    status_code=status.HTTP_201_CREATED
+)
 async def create_new_role(
     role_create: RoleCreate,
     admin_service: FromDishka[AdminService]

src/api/routers/chat.py

@@ -4,7 +4,7 @@
 from fastapi import APIRouter, Depends, status
 from pydantic import UUID4
 
-from src.api.dependencies import get_current_user
+from src.api.dependencies import PermissionChecker, get_current_user
 from src.api.schemas.chat import (
     ChatBase,
     ChatHistoryResponse,
@@ -13,6 +13,7 @@
     MessageResponse,
 )
 from src.application.services.chat_service import ChatService
+from src.core.security_policy import Action
 from src.domain.models.user import User
 
 router_chat = APIRouter(
@@ -25,6 +26,7 @@
     "/",
     response_model=ChatResponse,
     status_code=status.HTTP_201_CREATED,
+    dependencies=[Depends(PermissionChecker(Action.CHAT_CREATE))]
 )
 async def create_new_chat(
         chat_data: ChatBase,
@@ -38,7 +40,11 @@ async def create_new_chat(
     )
 
 
-@router_chat.get("/", response_model=List[ChatResponse])
+@router_chat.get(
+    "/",
+    response_model=List[ChatResponse],
+    dependencies=[Depends(PermissionChecker(Action.CHAT_READ))]
+)
 async def get_user_chats(
         service: FromDishka[ChatService],
         current_user: User = Depends(get_current_user)
@@ -47,7 +53,11 @@ async def get_user_chats(
     return await service.get_user_chats(current_user.id)
 
 
-@router_chat.get("/{chat_id}", response_model=ChatHistoryResponse)
+@router_chat.get(
+    "/{chat_id}",
+    response_model=ChatHistoryResponse,
+    dependencies=[Depends(PermissionChecker(Action.CHAT_READ))]
+)
 async def get_chat_history(
         chat_id: UUID4,
         service: FromDishka[ChatService],
@@ -63,6 +73,7 @@ async def get_chat_history(
 @router_chat.post(
     "/{chat_id}/message",
     response_model=MessageResponse,
+    dependencies=[Depends(PermissionChecker(Action.CHAT_WRITE))]
 )
 async def send(
         chat_id: UUID4,

src/api/routers/indexing.py

@@ -2,21 +2,25 @@
 from dishka.integrations.fastapi import DishkaRoute, FromDishka
 from fastapi import APIRouter, Depends, HTTPException, status
 
-from src.api.dependencies import get_current_admin_user
+from src.api.dependencies import PermissionChecker
 from src.api.schemas.repository import (
     IndexingJob,
     JobStatusUpdate,
     SyncRequest,
 )
 from src.application.services.index_service import IndexService
+from src.core.security_policy import Action
 
 router_indexing = APIRouter(
-    dependencies=[Depends(get_current_admin_user)],
     route_class=DishkaRoute
 )
 
 
-@router_indexing.post("/trigger", response_model=IndexingJob)
+@router_indexing.post(
+    "/trigger",
+    response_model=IndexingJob,
+    dependencies=[Depends(PermissionChecker(Action.INDEXING_TRIGGER))]
+)
 async def trigger_indexing(
     sync_request: SyncRequest,
     service: FromDishka[IndexService]
@@ -27,7 +31,10 @@ async def trigger_indexing(
     )
 
 
-@router_indexing.delete("/{job_id}")
+@router_indexing.delete(
+    "/{job_id}",
+    dependencies=[Depends(PermissionChecker(Action.INDEXING_DELETE))]
+)
 async def delete_indexing_job(
     job_id: str,
     service: FromDishka[IndexService]
@@ -39,7 +46,11 @@ async def delete_indexing_job(
     return await service.delete_indexind_job(job_id)
 
 
-@router_indexing.get("/status/{job_id}", response_model=IndexingJob)
+@router_indexing.get(
+    "/status/{job_id}",
+    response_model=IndexingJob,
+    dependencies=[Depends(PermissionChecker(Action.INDEXING_GET))]
+)
 async def get_indexing_status(
     job_id: str,
     service: FromDishka[IndexService]
@@ -48,7 +59,11 @@ async def get_indexing_status(
     return await service.get_indexing_status(job_id)
 
 
-@router_indexing.put("/status/{job_id}", response_model=IndexingJob)
+@router_indexing.put(
+    "/status/{job_id}",
+    response_model=IndexingJob,
+    dependencies=[Depends(PermissionChecker(Action.INDEXING_UPDATE))]
+)
 async def update_indexing_status(
     job_id: str,
     status_update: JobStatusUpdate,

src/api/routers/repository.py

@@ -3,20 +3,24 @@
 from dishka.integrations.fastapi import DishkaRoute, FromDishka
 from fastapi import APIRouter, Depends, status
 
-from src.api.dependencies import get_current_admin_user
+from src.api.dependencies import PermissionChecker
 from src.api.schemas.repository import (
     GitLabConfigCreate,
     Repository,
 )
 from src.application.services.index_service import IndexService
+from src.core.security_policy import Action
 
 router_repository = APIRouter(
-    dependencies=[Depends(get_current_admin_user)],
     route_class=DishkaRoute
 )
 
 
-@router_repository.post("/config", status_code=status.HTTP_202_ACCEPTED)
+@router_repository.post(
+    "/config",
+    status_code=status.HTTP_202_ACCEPTED,
+    dependencies=[Depends(PermissionChecker(Action.REPO_CONFIG))]
+)
 async def configure_gitlab(
     config_data: GitLabConfigCreate,
     service: FromDishka[IndexService]
@@ -28,7 +32,11 @@ async def configure_gitlab(
     )
 
 
-@router_repository.get("/list", response_model=List[Repository])
+@router_repository.get(
+    "/list",
+    response_model=List[Repository],
+    dependencies=[Depends(PermissionChecker(Action.REPO_READ))]
+)
 async def list_gitlab_repositories(
     service: FromDishka[IndexService]
 ):

src/api/schemas/repository.py

@@ -22,7 +22,7 @@ class Repository(BaseModel):
     id: UUID
     name: str
     path_with_namespace: str = Field(..., example="group/my-awesome-project")
-    web_url: HttpUrl
+    url: HttpUrl
 
 
 class SyncRequest(BaseModel):  # разделить на два разных типа - одиночный и multi?

src/core/security_policy.py

@@ -0,0 +1,45 @@
+from enum import Enum
+from typing import Dict, Set
+
+from src.domain.models.user import UserRole
+
+
+class Action(str, Enum):
+
+    """Data structure for actions in different services (for RBAC)."""
+
+    CHAT_READ = "chat:read"
+    CHAT_WRITE = "chat:write"
+    CHAT_CREATE = "chat:create"
+
+    REPO_READ = "repo:read"
+    REPO_CONFIG = "repo:config" # initial config
+
+    INDEXING_TRIGGER = "indexing:trigger"
+    INDEXING_DELETE = "indexing:delete"
+    INDEXING_UPDATE = "indexing:update"
+    INDEXING_GET = "indexing:get"
+
+    ADMIN_ACCESS = "admin:access" # for getting and changing values in db
+
+ROLE_PERMISSIONS: Dict[UserRole, Set[Action]] = {
+    UserRole.USER: {
+        Action.CHAT_READ,
+        Action.CHAT_WRITE,
+        Action.CHAT_CREATE,
+        Action.REPO_READ,
+        Action.INDEXING_GET
+    },
+    UserRole.ADMIN: {
+        Action.CHAT_READ,
+        Action.CHAT_WRITE,
+        Action.CHAT_CREATE,
+        Action.REPO_READ,
+        Action.REPO_CONFIG,
+        Action.INDEXING_TRIGGER,
+        Action.INDEXING_DELETE,
+        Action.INDEXING_UPDATE,
+        Action.INDEXING_GET,
+        Action.ADMIN_ACCESS
+    }
+}

tests/api/test_admin_router.py

@@ -8,7 +8,7 @@
 from fastapi import HTTPException, status
 from httpx import ASGITransport, AsyncClient
 
-from src.api.dependencies import get_current_admin_user
+from src.api.dependencies import get_current_user
 from src.application.services.admin_service import AdminService
 from src.domain.models.user import Role as DomainRole, User as DomainUser, UserRole
 
@@ -45,7 +45,7 @@ async def dishka_app(app_fixture, mock_admin_service):
 @pytest_asyncio.fixture(scope="function")
 async def ac(dishka_app):
     """Create AsyncClient for tests."""
-    dishka_app.dependency_overrides[get_current_admin_user] = lambda: DomainUser(
+    dishka_app.dependency_overrides[get_current_user] = lambda: DomainUser(
         id=uuid4(),
         username="admin",
         email="admin@test.com",

tests/api/test_indexing_router.py

@@ -8,7 +8,7 @@
 from dishka.integrations.fastapi import setup_dishka
 from httpx import ASGITransport, AsyncClient
 
-from src.api.dependencies import get_current_admin_user
+from src.api.dependencies import get_current_user
 from src.application.services.admin_service import AdminService
 from src.application.services.index_service import IndexService
 from src.domain.models.knowledge import JobStatus
@@ -54,7 +54,7 @@ async def dishka_app(app_fixture, mock_index_service):
 @pytest_asyncio.fixture(scope="function")
 async def ac(dishka_app):
     """Create AsyncClient for tests."""
-    dishka_app.dependency_overrides[get_current_admin_user] = lambda: DomainUser(
+    dishka_app.dependency_overrides[get_current_user] = lambda: DomainUser(
         id=uuid4(),
         username="admin",
         email="admin@test.com",

tests/api/test_repository_router.py

@@ -7,7 +7,7 @@
 from dishka.integrations.fastapi import setup_dishka
 from httpx import ASGITransport, AsyncClient
 
-from src.api.dependencies import get_current_admin_user
+from src.api.dependencies import get_current_user
 from src.application.services.index_service import IndexService
 from src.domain.models.user import User as DomainUser
 
@@ -44,7 +44,7 @@ async def dishka_app(app_fixture, mock_index_service):
 @pytest_asyncio.fixture(scope="function")
 async def ac(dishka_app):
     """Create AsyncClient for tests."""
-    dishka_app.dependency_overrides[get_current_admin_user] = lambda: DomainUser(
+    dishka_app.dependency_overrides[get_current_user] = lambda: DomainUser(
         id=uuid4(),
         username="admin",
         email="admin@test.com",
@@ -92,13 +92,13 @@ async def test_list_gitlab_repositories_success(ac, mock_index_service):
             "id": str(uuid4()),
             "name": "Repo 1",
             "path_with_namespace": "group/repo1",
-            "web_url": "https://gitlab.com/group/repo1"
+            "url": "https://gitlab.com/group/repo1"
         },
         {
             "id": str(uuid4()),
             "name": "Repo 2",
             "path_with_namespace": "group/repo2",
-            "web_url": "https://gitlab.com/group/repo2"
+            "url": "https://gitlab.com/group/repo2"
         }
     ]
     mock_index_service.list_repositories.return_value = mock_repos