Prod: harden program project access and expert filters

partner_programs/permissions.py

@@ -1,6 +1,12 @@
 from rest_framework.permissions import BasePermission
 
-from partner_programs.models import Application, PartnerProgram, Submission, Team, TeamMember
+from partner_programs.models import (
+    Application,
+    PartnerProgram,
+    Submission,
+    Team,
+    TeamMember,
+)
 
 
 def _is_authenticated(user) -> bool:
@@ -133,3 +139,17 @@ def has_permission(self, request, view):
                 return False
 
         return program.is_manager(user)
+
+
+class IsAdminOrManagerOrExpertOfProgram(IsAdminOrManagerOfProgram):
+    """Filter-schema read access; do not use for manager mutation endpoints."""
+
+    def has_permission(self, request, view):
+        if super().has_permission(request, view):
+            return True
+        if not _is_authenticated(request.user):
+            return False
+        program_id = view.kwargs.get("pk") or view.kwargs.get("program_id")
+        return PartnerProgram.objects.filter(
+            pk=program_id, experts__user_id=request.user.pk
+        ).exists()

partner_programs/tests/test_program_filter_access.py

@@ -0,0 +1,101 @@
+"""Expert schema reads must not widen manager APIs or change filter payloads."""
+
+from django.test import TestCase
+from rest_framework.test import APIClient
+
+from partner_programs.serializers import PartnerProgramFieldSerializer
+from partner_programs.tests.helpers import (
+    create_partner_program,
+    create_program_field,
+    create_program_member,
+    create_user,
+)
+from project_rates.tests.helpers import create_rate_expert
+
+
+class ProgramFilterAccessTests(TestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.program = create_partner_program()
+        cls.foreign_program = create_partner_program()
+        cls.field = create_program_field(
+            cls.program,
+            name="track",
+            field_type="select",
+            options=["A", "B"],
+            show_filter=True,
+        )
+        create_program_field(cls.program, name="private", show_filter=False)
+        create_program_field(cls.foreign_program, name="foreign", show_filter=True)
+        cls.manager = create_user(prefix="filter-manager")
+        cls.program.managers.add(cls.manager)
+        cls.expert = create_rate_expert(program=cls.program)
+        cls.foreign_expert = create_rate_expert(program=cls.foreign_program)
+        cls.foreign_manager = create_user(prefix="filter-foreign-manager")
+        cls.foreign_program.managers.add(cls.foreign_manager)
+        cls.participant = create_program_member(cls.program).user
+        cls.outsider = create_user(prefix="filter-outsider")
+        cls.staff = create_user(prefix="filter-staff", is_staff=True)
+        cls.superuser = create_user(prefix="filter-superuser", is_superuser=True)
+
+    def setUp(self):
+        self.client = APIClient()
+        self.url = f"/programs/{self.program.pk}/filters/"
+
+    def test_access_matrix_and_unchanged_response(self):
+        for user, expected in (
+            (None, 401),
+            (self.participant, 403),
+            (self.outsider, 403),
+            (self.foreign_expert, 403),
+            (self.foreign_manager, 403),
+            (self.expert, 200),
+            (self.manager, 200),
+            (self.staff, 200),
+            (self.superuser, 200),
+        ):
+            with self.subTest(user=user):
+                self.client.force_authenticate(user)
+                response = self.client.get(self.url)
+                self.assertEqual(response.status_code, expected, response.data)
+                if expected == 200:
+                    self.assertEqual(
+                        response.data, [PartnerProgramFieldSerializer(self.field).data]
+                    )
+
+    def test_schema_endpoint_stays_read_only(self):
+        for user in (self.expert, self.manager, self.staff, self.superuser):
+            self.client.force_authenticate(user)
+            for method in ("post", "patch", "put", "delete"):
+                with self.subTest(user=user.pk, method=method):
+                    response = getattr(self.client, method)(self.url, {}, format="json")
+                    self.assertEqual(response.status_code, 405, response.data)
+
+    def test_expert_does_not_gain_other_manager_endpoints(self):
+        self.client.force_authenticate(self.expert)
+        for suffix, method in (
+            ("projects/", "get"),
+            ("projects/filter/", "post"),
+            ("export-projects/", "get"),
+            ("manager-overview/", "get"),
+        ):
+            with self.subTest(suffix=suffix):
+                response = getattr(self.client, method)(
+                    f"/programs/{self.program.pk}/{suffix}"
+                )
+                self.assertEqual(response.status_code, 403, response.data)
+
+    def test_missing_program_preserves_production_statuses(self):
+        missing_id = self.foreign_program.pk + 1000
+        for user, expected in (
+            (None, 401),
+            (self.participant, 403),
+            (self.expert, 403),
+            (self.manager, 403),
+            (self.staff, 404),
+            (self.superuser, 404),
+        ):
+            with self.subTest(user=user):
+                self.client.force_authenticate(user)
+                response = self.client.get(f"/programs/{missing_id}/filters/")
+                self.assertEqual(response.status_code, expected, response.data)

partner_programs/views.py

@@ -26,6 +26,7 @@
 from partner_programs.pagination import PartnerProgramPagination
 from partner_programs.permissions import (
     IsAdminOrManagerOfProgram,
+    IsAdminOrManagerOrExpertOfProgram,
     IsProjectLeader,
 )
 from partner_programs.serializers import (
@@ -74,13 +75,10 @@ def get_queryset(self):
             qs = PartnerProgram.objects.none()
         else:
             now = timezone.now()
-            qs = (
-                base_qs.filter(
-                    partner_program_profiles__user=self.request.user,
-                    datetime_finished__gte=now,
-                )
-                .distinct()
-            )
+            qs = base_qs.filter(
+                partner_program_profiles__user=self.request.user,
+                datetime_finished__gte=now,
+            ).distinct()
 
         user = self.request.user
         if not user.is_authenticated:
@@ -179,7 +177,9 @@ def get(self, request, pk, *args, **kwargs):
                 "program_id": program.id,
                 "can_submit": program.is_project_submission_open(),
                 "submission_deadline": program.get_project_submission_deadline(),
-                "program_fields": PartnerProgramFieldSerializer(fields_qs, many=True).data,
+                "program_fields": PartnerProgramFieldSerializer(
+                    fields_qs, many=True
+                ).data,
             },
             status=status.HTTP_200_OK,
         )
@@ -419,7 +419,7 @@ def post(self, request, pk, *args, **kwargs):
 
 
 class ProgramFiltersAPIView(APIView):
-    permission_classes = [IsAuthenticated, IsAdminOrManagerOfProgram]
+    permission_classes = [IsAuthenticated, IsAdminOrManagerOrExpertOfProgram]
 
     def get(self, request, pk):
         program = get_object_or_404(PartnerProgram, pk=pk)

projects/access.py

@@ -0,0 +1,37 @@
+"""Project-level involvement and program-scoped read access."""
+
+from django.db.models import Q
+
+from partner_programs.models import PartnerProgramProject
+from projects.models import Project
+
+
+def has_project_level_read_access(user, project: Project) -> bool:
+    """Existing project/admin involvement, independent of any program role."""
+    if not user or not user.is_authenticated:
+        return False
+    return (
+        user.is_staff
+        or user.is_superuser
+        or project.leader_id == user.pk
+        or project.collaborator_set.filter(user_id=user.pk).exists()
+        or project.invite_set.filter(user_id=user.pk).exists()
+    )
+
+
+def program_role_project_links(user, project: Project):
+    """Only links whose program the user manages or evaluates; no per-link SQL."""
+    if not user or not user.is_authenticated:
+        return PartnerProgramProject.objects.none()
+    return project.program_links.filter(
+        Q(partner_program__managers__pk=user.pk)
+        | Q(partner_program__experts__user_id=user.pk)
+    )
+
+
+def has_project_read_involvement(user, project: Project) -> bool:
+    """Read-only grant; callers must keep existing write permissions separate."""
+    return (
+        has_project_level_read_access(user, project)
+        or program_role_project_links(user, project).exists()
+    )

projects/permissions.py

@@ -1,16 +1,15 @@
 from datetime import datetime, timedelta
 
 from django.utils import timezone
-from django.db.models import Q
 from rest_framework.exceptions import PermissionDenied, ValidationError
 from rest_framework.permissions import SAFE_METHODS, BasePermission
 
 from partner_programs.models import (
     PartnerProgram,
-    PartnerProgramProject,
     PartnerProgramUserProfile,
 )
 from projects.models import Project
+from projects.access import has_project_read_involvement
 
 
 class ProjectVisibilityPermission(BasePermission):
@@ -31,7 +30,10 @@ def has_permission(self, request, view):
         if not project_id and view.__module__.startswith("projects."):
             project_id = view.kwargs.get("id") or view.kwargs.get("pk")
 
-        if not project_id and getattr(getattr(view, "queryset", None), "model", None) is Project:
+        if (
+            not project_id
+            and getattr(getattr(view, "queryset", None), "model", None) is Project
+        ):
             project_id = view.kwargs.get("pk")
 
         if not project_id:
@@ -58,29 +60,7 @@ def has_object_permission(self, request, view, obj):
     def _can_view_project(self, request, project: Project) -> bool:
         if project.is_public:
             return True
-
-        user = getattr(request, "user", None)
-        if not user or not user.is_authenticated:
-            return False
-
-        if user.is_superuser or user.is_staff:
-            return True
-
-        if project.leader_id == user.id:
-            return True
-
-        if project.collaborator_set.filter(user_id=user.id).exists():
-            return True
-
-        if project.invite_set.filter(user_id=user.id).exists():
-            return True
-
-        return PartnerProgramProject.objects.filter(
-            project_id=project.id,
-        ).filter(
-            Q(partner_program__managers__id=user.id)
-            | Q(partner_program__experts__user_id=user.id)
-        ).exists()
+        return has_project_read_involvement(getattr(request, "user", None), project)
 
 
 class IsProjectLeaderOrReadOnlyForNonDrafts(BasePermission):
@@ -132,6 +112,9 @@ def has_permission(self, request, view) -> bool:
         return False
 
     def has_object_permission(self, request, view, obj):
+        if request.method in SAFE_METHODS:
+            return not obj.draft or has_project_read_involvement(request.user, obj)
+
         if not obj.draft:
             # not a draft, read-only for everyone, write only for leader
             if (request.method in SAFE_METHODS) or (request.user == obj.leader):
@@ -195,9 +178,7 @@ def _prepare_exception_detail(
             datetime_finished + timedelta(seconds=self._SECONDS_AFTER_CANT_EDIT)
         )
         days_until_resolution: int = (
-            int(self._SECONDS_AFTER_CANT_EDIT / 60 / 60 / 24)
-            - days_from_end_program
-            - 1
+            int(self._SECONDS_AFTER_CANT_EDIT / 60 / 60 / 24) - days_from_end_program - 1
         )
         return {
             "program_name": program_profile.partner_program.name,
@@ -222,7 +203,9 @@ def has_permission(self, request, view):
             return False
 
         project_pk = view.kwargs.get("project_pk")
-        project_id = project_pk or view.kwargs.get("project_id") or request.data.get("project")
+        project_id = (
+            project_pk or view.kwargs.get("project_id") or request.data.get("project")
+        )
         if not project_id:
             return False
 
@@ -259,10 +242,14 @@ def has_permission(self, request, view):
 
         submission_deadline = program.get_project_submission_deadline()
         if submission_deadline and submission_deadline < timezone.now():
-            raise ValidationError({"partner_program_id": "Срок подачи проектов в программу завершён."})
+            raise ValidationError(
+                {"partner_program_id": "Срок подачи проектов в программу завершён."}
+            )
 
         if program.datetime_finished < timezone.now():
-            raise ValidationError({"partner_program_id": "Нельзя выбрать завершённую программу."})
+            raise ValidationError(
+                {"partner_program_id": "Нельзя выбрать завершённую программу."}
+            )
 
         if program.is_manager(request.user):
             return True

projects/serializers.py

@@ -28,6 +28,7 @@
     Resource,
 )
 from projects.validators import validate_project
+from projects.access import has_project_level_read_access, program_role_project_links
 from vacancy.serializers import ProjectVacancyListSerializer
 
 User = get_user_model()
@@ -261,11 +262,30 @@ class ProjectDetailSerializer(serializers.ModelSerializer):
     trl = serializers.IntegerField(required=False, allow_null=True)
 
     def get_partner_program(self, project):
-        try:
-            link = project.program_links.select_related("partner_program").get()
-            return PartnerProgramProjectSerializer(link).data
-        except PartnerProgramProject.DoesNotExist:
-            return None
+        user = getattr(self.context.get("request"), "user", None)
+        link = None
+        if (
+            user
+            and user.is_authenticated
+            and not has_project_level_read_access(user, project)
+        ):
+            link = (
+                program_role_project_links(user, project)
+                .select_related("partner_program")
+                .order_by("pk")
+                .first()
+            )
+            if link is None and (project.draft or not project.is_public):
+                return None
+        # Independent project roles and existing public readers retain the legacy
+        # singular contract. Program-role readers serialize only an eligible link.
+        if link is None:
+            link = (
+                project.program_links.select_related("partner_program")
+                .order_by("pk")
+                .first()
+            )
+        return PartnerProgramProjectSerializer(link).data if link is not None else None
 
     @classmethod
     def get_partner_program_tags(cls, project):