add permission check for form productor and product

backend/src/auth/auth.py

@@ -1,20 +1,21 @@
-from typing import Annotated
-from fastapi import APIRouter, Security, HTTPException, Depends, Request, Cookie
-from fastapi.responses import RedirectResponse, Response
-from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
-from sqlmodel import Session, select
-import jwt
-from jwt import PyJWKClient
-
-from src.settings import AUTH_URL, TOKEN_URL, JWKS_URL, ISSUER, LOGOUT_URL, settings
-import src.users.service as service
-from src.database import get_session
-from src.models import UserCreate, User, UserPublic
-
 import secrets
-import requests
+from typing import Annotated
 from urllib.parse import urlencode
+
+import jwt
+import requests
 import src.messages as messages
+import src.users.service as service
+from fastapi import (APIRouter, Cookie, Depends, HTTPException, Request,
+                     Security)
+from fastapi.responses import RedirectResponse, Response
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from jwt import PyJWKClient
+from sqlmodel import Session, select
+from src.database import get_session
+from src.models import User, UserCreate, UserPublic
+from src.settings import (AUTH_URL, ISSUER, JWKS_URL, LOGOUT_URL, TOKEN_URL,
+                          settings)
 
 router = APIRouter(prefix='/auth')
 
@@ -98,7 +99,7 @@ def callback(code: str, session: Session = Depends(get_session)):
             'client_secret': settings.keycloak_client_secret,
             'refresh_token': token_data['refresh_token'],
         }
-        res = requests.post(LOGOUT_URL, data=data)
+        requests.post(LOGOUT_URL, data=data)
         resp = RedirectResponse(f'{settings.origins}?userNotAllowed=true')
         return resp
     roles = resource_access.get(settings.keycloak_client_id)
@@ -108,7 +109,7 @@ def callback(code: str, session: Session = Depends(get_session)):
             'client_secret': settings.keycloak_client_secret,
             'refresh_token': token_data['refresh_token'],
         }
-        res = requests.post(LOGOUT_URL, data=data)
+        requests.post(LOGOUT_URL, data=data)
         resp = RedirectResponse(f'{settings.origins}?userNotAllowed=true')
         return resp
 
@@ -160,12 +161,15 @@ def verify_token(token: str):
         )
         return decoded
     except jwt.ExpiredSignatureError:
-        raise HTTPException(status_code=401,
-                            detail=messages.Messages.tokenexipired)
+        raise HTTPException(
+            status_code=401,
+            detail=messages.Messages.tokenexipired
+        )
     except jwt.InvalidTokenError:
         raise HTTPException(
             status_code=401,
-            detail=messages.Messages.invalidtoken)
+            detail=messages.Messages.invalidtoken
+        )
 
 
 def get_current_user(
@@ -173,21 +177,30 @@ def get_current_user(
         session: Session = Depends(get_session)):
     access_token = request.cookies.get('access_token')
     if not access_token:
-        raise HTTPException(status_code=401,
-                            detail=messages.Messages.notauthenticated)
+        raise HTTPException(
+            status_code=401,
+            detail=messages.Messages.notauthenticated
+        )
     payload = verify_token(access_token)
     if not payload:
-        raise HTTPException(status_code=401, detail='aze')
+        raise HTTPException(
+            status_code=401, 
+            detail='aze'
+        )
     email = payload.get('email')
 
     if not email:
-        raise HTTPException(status_code=401,
-                            detail=messages.Messages.notauthenticated)
+        raise HTTPException(
+            status_code=401,
+            detail=messages.Messages.notauthenticated
+        )
 
     user = session.exec(select(User).where(User.email == email)).first()
     if not user:
-        raise HTTPException(status_code=401,
-                            detail=messages.Messages.not_found('user'))
+        raise HTTPException(
+            status_code=401,
+            detail=messages.Messages.not_found('user')
+        )
     return user
 
 
@@ -249,6 +262,6 @@ def me(user: UserPublic = Depends(get_current_user)):
             'name': user.name,
             'email': user.email,
             'id': user.id,
-            'roles': [role.name for role in user.roles]
+            'roles': user.roles
         }
     }

backend/src/forms/forms.py

@@ -32,7 +32,10 @@ async def get_forms_filtered(
 
 
 @router.get('/{_id}', response_model=models.FormPublic)
-async def get_form(_id: int, session: Session = Depends(get_session)):
+async def get_form(
+    _id: int, 
+    session: Session = Depends(get_session)
+):
     result = service.get_one(session, _id)
     if result is None:
         raise HTTPException(
@@ -48,6 +51,11 @@ async def create_form(
     user: models.User = Depends(get_current_user),
     session: Session = Depends(get_session)
 ):
+    if not service.is_allowed(session, user, form=form):
+        raise HTTPException(
+            status_code=403,
+            detail=messages.Messages.not_allowed('forms', 'update')
+        )
     try:
         form = service.create_one(session, form)
     except exceptions.ProductorNotFoundError as error:
@@ -61,10 +69,16 @@ async def create_form(
 
 @router.put('/{_id}', response_model=models.FormPublic)
 async def update_form(
-    _id: int, form: models.FormUpdate,
+    _id: int, 
+    form: models.FormUpdate,
     user: models.User = Depends(get_current_user),
     session: Session = Depends(get_session)
 ):
+    if not service.is_allowed(session, user, _id=_id):
+        raise HTTPException(
+            status_code=403,
+            detail=messages.Messages.not_allowed('forms', 'update')
+        )
     try:
         result = service.update_one(session, _id, form)
     except exceptions.FormNotFoundError as error:
@@ -82,6 +96,11 @@ async def delete_form(
     user: models.User = Depends(get_current_user),
     session: Session = Depends(get_session)
 ):
+    if not service.is_allowed(session, user, _id=_id):
+        raise HTTPException(
+            status_code=403,
+            detail=messages.Messages.not_allowed('forms', 'delete')
+        )
     try:
         result = service.delete_one(session, _id)
     except exceptions.FormNotFoundError as error:

backend/src/forms/service.py

@@ -108,12 +108,25 @@ def delete_one(session: Session, _id: int) -> models.FormPublic:
     return result
 
 
-def is_allowed(session: Session, user: models.User, _id: int) -> bool:
+def is_allowed(
+    session: Session, 
+    user: models.User, 
+    _id: int = None, 
+    form: models.FormCreate = None
+) -> bool:
+    if not _id:
+        statement = (
+            select(models.Productor)
+            .where(models.Productor.id == form.productor_id)
+        )
+        productor = session.exec(statement).first()
+        return productor.type in [r.name for r in user.roles]
     statement = (
         select(models.Form)
         .join(
             models.Productor,
-            models.Form.productor_id == models.Productor.id)
+            models.Form.productor_id == models.Productor.id
+        )
         .where(models.Form.id == _id)
         .where(
             models.Productor.type.in_(

backend/src/productors/service.py

@@ -92,3 +92,19 @@ def delete_one(session: Session, id: int) -> models.ProductorPublic:
     session.delete(productor)
     session.commit()
     return result
+
+def is_allowed(
+    session: Session, 
+    user: models.User, 
+    _id: int, 
+    productor: models.ProductorCreate
+) -> bool:
+    if not _id:
+        return productor.type in [r.name for r in user.roles]
+    statement = (
+        select(models.Productor)
+        .where(models.Productor.id == _id)
+        .where(models.Productor.type.in_([r.name for r in user.roles]))
+        .distinct()
+    )
+    return len(session.exec(statement).all()) > 0

backend/src/products/service.py

@@ -85,3 +85,32 @@ def delete_one(session: Session, id: int) -> models.ProductPublic:
     session.delete(product)
     session.commit()
     return result
+
+def is_allowed(
+    session: Session, 
+    user: models.User, 
+    _id: int, 
+    product: models.ProductCreate
+) -> bool:
+    if not _id:
+        statement = (
+            select(models.Product)
+            .join(
+                models.Productor,
+                models.Product.productor_id == models.Productor.id
+            )
+            .where(models.Product.id == product.productor_id)
+        )
+        productor = session.exec(statement).first()
+        return productor.type in [r.name for r in user.roles]
+    statement = (
+        select(models.Product)
+        .join(
+            models.Productor,
+            models.Product.productor_id == models.Productor.id
+        )
+        .where(models.Product.id == _id)
+        .where(models.Productor.type.in_([r.name for r in user.roles]))
+        .distinct()
+    )
+    return len(session.exec(statement).all()) > 0

backend/src/users/service.py

@@ -56,7 +56,9 @@ def get_or_create_user(session: Session, user_create: models.UserCreate):
 
 
 def get_roles(session: Session):
-    statement = select(models.ContractType)
+    statement = (
+        select(models.ContractType)
+    )
     return session.exec(statement.order_by(models.ContractType.name)).all()
 
 
@@ -64,7 +66,9 @@ def create_one(session: Session, user: models.UserCreate) -> models.UserPublic:
     if user is None:
         raise exceptions.UserCreateError(
             messages.Messages.invalid_input(
-                'user', 'input cannot be None'))
+                'user', 'input cannot be None'
+            )
+        )
     new_user = models.User(
         name=user.name,
         email=user.email
@@ -81,17 +85,19 @@ def create_one(session: Session, user: models.UserCreate) -> models.UserPublic:
 
 def update_one(
         session: Session,
-        id: int,
+        _id: int,
         user: models.UserCreate) -> models.UserPublic:
     if user is None:
         raise exceptions.UserCreateError(
-            messages.s.invalid_input(
-                'user', 'input cannot be None'))
-    statement = select(models.User).where(models.User.id == id)
+            messages.Messages.invalid_input(
+                'user', 'input cannot be None'
+            )
+        )
+    statement = select(models.User).where(models.User.id == _id)
     result = session.exec(statement)
     new_user = result.first()
     if not new_user:
-        raise exceptions.UserNotFoundError(f'User {id} not found')
+        raise exceptions.UserNotFoundError(f'User {_id} not found')
     new_user.email = user.email
     new_user.name = user.name
 
@@ -103,12 +109,12 @@ def update_one(
     return new_user
 
 
-def delete_one(session: Session, id: int) -> models.UserPublic:
-    statement = select(models.User).where(models.User.id == id)
+def delete_one(session: Session, _id: int) -> models.UserPublic:
+    statement = select(models.User).where(models.User.id == _id)
     result = session.exec(statement)
     user = result.first()
     if not user:
-        raise exceptions.UserNotFoundError(f'User {id} not found')
+        raise exceptions.UserNotFoundError(f'User {_id} not found')
     result = models.UserPublic.model_validate(user)
     session.delete(user)
     session.commit()

backend/src/users/users.py

@@ -32,16 +32,18 @@ def get_roles(
     return service.get_roles(session)
 
 
-@router.get('/{id}', response_model=models.UserPublic)
-def get_users(
-    id: int,
+@router.get('/{_id}', response_model=models.UserPublic)
+def get_user(
+    _id: int,
     user: models.User = Depends(get_current_user),
     session: Session = Depends(get_session)
 ):
-    result = service.get_one(session, id)
+    result = service.get_one(session, _id)
     if result is None:
-        raise HTTPException(status_code=404,
-                            detail=messages.Messages.not_found('user'))
+        raise HTTPException(
+            status_code=404,
+            detail=messages.Messages.not_found('user')
+        )
     return result
 
 
@@ -54,22 +56,27 @@ def create_user(
     try:
         user = service.create_one(session, user)
     except exceptions.UserCreateError as error:
-        raise HTTPException(status_code=400, detail=str(error))
+        raise HTTPException(
+            status_code=400, 
+            detail=str(error)
+        ) from error
     return user
 
 
-@router.put('/{id}', response_model=models.UserPublic)
+@router.put('/{_id}', response_model=models.UserPublic)
 def update_user(
-    id: int,
+    _id: int,
     user: models.UserUpdate,
     logged_user: models.User = Depends(get_current_user),
     session: Session = Depends(get_session)
 ):
     try:
-        result = service.update_one(session, id, user)
+        result = service.update_one(session, _id, user)
     except exceptions.UserNotFoundError as error:
-        raise HTTPException(status_code=404,
-                            detail=messages.Messages.not_found('user'))
+        raise HTTPException(
+            status_code=404,
+            detail=messages.Messages.not_found('user')
+        ) from error
     return result
 
 
@@ -82,6 +89,8 @@ def delete_user(
     try:
         result = service.delete_one(session, id)
     except exceptions.UserNotFoundError as error:
-        raise HTTPException(status_code=404,
-                            detail=messages.Messages.not_found('user'))
+        raise HTTPException(
+            status_code=404,
+            detail=messages.Messages.not_found('user')
+        ) from error
     return result

backend/tests/routers/test_contracts.py

@@ -134,57 +134,6 @@ class TestContracts:
 
         app.dependency_overrides.clear()
 
-    def test_create_one(self, client, mocker, mock_session, mock_user):
-        contract_body = contract_factory.contract_body_factory(
-            products=[
-                contract_products_factory.contract_product_body_factory(
-                    product_id=1
-                ),
-                contract_products_factory.contract_product_body_factory(
-                    product_id=2
-                ),
-                contract_products_factory.contract_product_body_factory(
-                    product_id=3
-                )
-            ],
-            cheques=[{'name': '123123', 'value': '100'}]
-        )
-        contract_result = contract_factory.contract_factory(
-            products=[
-                contract_products_factory.contract_product_factory(
-                    product_id=1
-                ),
-                contract_products_factory.contract_product_factory(
-                    product_id=2
-                ),
-                contract_products_factory.contract_product_factory(
-                    product_id=3
-                )
-            ],
-            form=form_factory.form_factory(),
-            cheques=[models.Cheque(name='123123', value='100')]
-        )
-        mocker.patch.object(
-            service,
-            'create_one',
-            return_value=contract_result
-        )
-        mocker.patch.object(
-            service,
-            'add_contract_file',
-            return_value=True
-        )
-        mocker.patch(
-            'src.contracts.generate_contract.generate_html_contract')
-
-        response = client.post('/api/contracts', json=contract_body)
-        assert response.status_code == 200
-        contract_id = 'test_test_test type_hiver-2026'
-        assert response.headers[
-            'Content-Disposition'] == (
-                f'attachment; filename=contract_{contract_id}.pdf'
-        )
-
     def test_delete_one(self, client, mocker, mock_session, mock_user):
         contract_result = contract_factory.contract_public_factory()
 
@@ -213,7 +162,8 @@ class TestContracts:
             client,
             mocker,
             mock_session,
-            mock_user):
+            mock_user
+    ):
         contract_result = None
 
         mock = mocker.patch.object(
@@ -241,7 +191,8 @@ class TestContracts:
             client,
             mocker,
             mock_session,
-            mock_user):
+            mock_user
+    ):
         def unauthorized():
             raise HTTPException(status_code=401)
 

frontend/src/components/Productors/Modal/index.tsx

@@ -19,7 +19,7 @@ import {
     type ProductorInputs,
 } from "@/services/resources/productors";
 import { useMemo } from "react";
-import { useGetRoles } from "@/services/api";
+import { useAuth } from "@/services/auth/AuthProvider";
 
 export type ProductorModalProps = ModalBaseProps & {
     currentProductor?: Productor;
@@ -32,7 +32,7 @@ export function ProductorModal({
     currentProductor,
     handleSubmit,
 }: ProductorModalProps) {
-    const { data: allRoles } = useGetRoles();
+    const { loggedUser } = useAuth();
 
     const form = useForm<ProductorInputs>({
         initialValues: {
@@ -58,8 +58,8 @@ export function ProductorModal({
     });
 
     const roleSelect = useMemo(() => {
-        return allRoles?.map((role) => ({ value: String(role.name), label: role.name }));
-    }, [allRoles]);
+        return loggedUser?.user?.roles?.map((role) => ({ value: String(role.name), label: role.name }));
+    }, [loggedUser?.user?.roles]);
 
     return (
         <Modal opened={opened} onClose={onClose} title={t("create productor", { capfirst: true })}>