From 8e3cd2d89f1e42a6457f3e924b75d7cf8bba5e76 Mon Sep 17 00:00:00 2001 From: DarKylX Date: Wed, 9 Sep 2026 14:00:25 +0300 Subject: [PATCH] fix: scope production project reads and expert filters --- partner_programs/permissions.py | 22 +- .../tests/test_program_filter_access.py | 101 +++++++ partner_programs/views.py | 18 +- projects/access.py | 37 +++ projects/permissions.py | 51 ++-- projects/serializers.py | 30 +- .../test_program_project_detail_access.py | 281 ++++++++++++++++++ 7 files changed, 493 insertions(+), 47 deletions(-) create mode 100644 partner_programs/tests/test_program_filter_access.py create mode 100644 projects/access.py create mode 100644 projects/tests/test_program_project_detail_access.py diff --git a/partner_programs/permissions.py b/partner_programs/permissions.py index 7993b812..ed6ad807 100644 --- a/partner_programs/permissions.py +++ b/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() diff --git a/partner_programs/tests/test_program_filter_access.py b/partner_programs/tests/test_program_filter_access.py new file mode 100644 index 00000000..cd6b698d --- /dev/null +++ b/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) diff --git a/partner_programs/views.py b/partner_programs/views.py index 8961ddb1..8c2029af 100644 --- a/partner_programs/views.py +++ b/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) diff --git a/projects/access.py b/projects/access.py new file mode 100644 index 00000000..1728190c --- /dev/null +++ b/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() + ) diff --git a/projects/permissions.py b/projects/permissions.py index a4b7d790..60eeafa1 100644 --- a/projects/permissions.py +++ b/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 diff --git a/projects/serializers.py b/projects/serializers.py index 43c1f7b4..3f00c0f2 100644 --- a/projects/serializers.py +++ b/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): diff --git a/projects/tests/test_program_project_detail_access.py b/projects/tests/test_program_project_detail_access.py new file mode 100644 index 00000000..af76ce59 --- /dev/null +++ b/projects/tests/test_program_project_detail_access.py @@ -0,0 +1,281 @@ +"""Production project reads: scoped program data without new write grants.""" + +from django.contrib.contenttypes.models import ContentType +from django.core.cache import cache +from django.db import connection +from django.test import TestCase +from django.test.utils import CaptureQueriesContext +from rest_framework.test import APIClient, APIRequestFactory + +from invites.models import Invite +from partner_programs.models import PartnerProgramFieldValue +from partner_programs.tests.helpers import create_program_field +from project_rates.tests.helpers import create_rate_expert +from projects.permissions import HasInvolvementInProjectOrReadOnly +from projects.serializers import PartnerProgramProjectSerializer + +from .helpers import ( + add_program_member, + create_collaborator, + create_partner_program, + create_project, + create_user, + link_project_to_program, +) + + +class ProgramProjectDetailAccessTests(TestCase): + @classmethod + def setUpTestData(cls): + cls.program_a = create_partner_program(is_competitive=True) + cls.program_b = create_partner_program(is_competitive=True) + cls.foreign_program = create_partner_program() + cls.project = create_project(draft=True, is_public=False) + cls.link_a = link_project_to_program(cls.project, cls.program_a) + cls.field_a = create_program_field(cls.program_a, name="answer_a") + cls.value_a = PartnerProgramFieldValue.objects.create( + program_project=cls.link_a, field=cls.field_a, value_text="Answer A" + ) + cls.manager_a = create_user(prefix="manager-a") + cls.program_a.managers.add(cls.manager_a) + cls.expert_a = create_rate_expert(program=cls.program_a) + cls.manager_b = create_user(prefix="manager-b") + cls.program_b.managers.add(cls.manager_b) + cls.expert_b = create_rate_expert(program=cls.program_b) + cls.foreign_manager = create_user(prefix="foreign-manager") + cls.foreign_program.managers.add(cls.foreign_manager) + cls.foreign_expert = create_rate_expert(program=cls.foreign_program) + cls.collaborator = create_user(prefix="collaborator") + add_program_member(cls.program_a, cls.collaborator) + create_collaborator(cls.project, user=cls.collaborator) + cls.invited = create_user(prefix="invited") + Invite.objects.create(project=cls.project, user=cls.invited) + cls.staff = create_user(prefix="staff") + cls.staff.is_staff = True + cls.staff.save(update_fields=["is_staff"]) + cls.superuser = create_user(prefix="superuser") + cls.superuser.is_superuser = True + cls.superuser.save(update_fields=["is_superuser"]) + cls.outsider = create_user(prefix="outsider") + + def setUp(self): + self.client = APIClient() + self.url = f"/projects/{self.project.pk}/" + + def detail(self, user): + self.client.force_authenticate(user) + response = self.client.get(self.url) + self.assertEqual(response.status_code, 200, response.data) + return response.data + + def add_second_link(self): + link = link_project_to_program(self.project, self.program_b) + field = create_program_field(self.program_b, name="answer_b") + PartnerProgramFieldValue.objects.create( + program_project=link, field=field, value_text="Private answer B" + ) + link.submitted = True + link.save(update_fields=["submitted"]) + return link + + def test_linked_program_and_project_roles_can_read_private_draft(self): + for user in ( + self.manager_a, + self.expert_a, + self.project.leader, + self.collaborator, + self.invited, + self.staff, + self.superuser, + ): + with self.subTest(user=user.pk): + self.assertEqual(self.detail(user)["id"], self.project.pk) + + def test_linked_manager_and_expert_safe_methods(self): + for user in (self.manager_a, self.expert_a): + self.client.force_authenticate(user) + for method in ("get", "head", "options"): + with self.subTest(user=user.pk, method=method): + self.assertEqual( + getattr(self.client, method)(self.url).status_code, 200 + ) + + def test_private_draft_rejects_unrelated_readers(self): + for user in (self.outsider, self.foreign_manager, self.foreign_expert): + with self.subTest(user=user.pk): + self.client.force_authenticate(user) + self.assertEqual(self.client.get(self.url).status_code, 403) + self.client.force_authenticate(None) + self.assertEqual(self.client.get(self.url).status_code, 401) + + def test_public_flag_does_not_make_draft_readable_to_outsiders(self): + self.project.is_public = True + self.project.save(update_fields=["is_public"]) + for user, expected in ((None, 401), (self.outsider, 403)): + with self.subTest(user=user): + self.client.force_authenticate(user) + self.assertEqual(self.client.get(self.url).status_code, expected) + self.detail(self.manager_a) + self.detail(self.expert_a) + + def test_read_grants_do_not_allow_writes(self): + original_name = self.project.name + for draft, public in ((True, False), (False, False), (False, True)): + self.project.draft, self.project.is_public = draft, public + self.project.save(update_fields=["draft", "is_public"]) + for user in (self.manager_a, self.expert_a, self.staff, self.superuser): + self.client.force_authenticate(user) + for method in ("patch", "put", "delete"): + with self.subTest( + draft=draft, public=public, user=user.pk, method=method + ): + response = getattr(self.client, method)( + self.url, {"name": "Unauthorized change"}, format="json" + ) + self.assertEqual(response.status_code, 403, response.data) + self.project.refresh_from_db() + self.assertEqual(self.project.name, original_name) + self.assertTrue( + self.project.program_links.filter(pk=self.link_a.pk).exists() + ) + + def test_independent_project_write_permissions_are_preserved(self): + add_program_member(self.program_a, self.manager_a) + create_collaborator(self.project, user=self.manager_a) + Invite.objects.create(project=self.project, user=self.expert_a) + factory = APIRequestFactory() + permission = HasInvolvementInProjectOrReadOnly() + for draft in (True, False): + self.project.draft = draft + for user in (self.project.leader, self.manager_a, self.expert_a): + for method in ("patch", "put", "delete"): + with self.subTest(draft=draft, user=user.pk, method=method): + request = getattr(factory, method)(self.url) + request.user = user + self.assertEqual( + permission.has_object_permission(request, None, self.project), + draft or user.pk == self.project.leader_id, + ) + + def test_manager_with_independent_collaborator_role_can_patch_draft(self): + add_program_member(self.program_a, self.manager_a) + create_collaborator(self.project, user=self.manager_a) + self.client.force_authenticate(self.manager_a) + response = self.client.patch(self.url, {"name": "Allowed change"}, format="json") + self.assertEqual(response.status_code, 200, response.data) + self.project.refresh_from_db() + self.assertEqual(self.project.name, "Allowed change") + + def test_single_link_preserves_full_nested_contract(self): + expected = PartnerProgramProjectSerializer(self.link_a).data + self.assertEqual(self.detail(self.manager_a)["partner_program"], expected) + self.assertEqual(self.detail(self.project.leader)["partner_program"], expected) + + def test_no_links_returns_null(self): + self.link_a.delete() + self.assertIsNone(self.detail(self.project.leader)["partner_program"]) + + def test_program_roles_only_receive_their_link_and_fields(self): + link_b = self.add_second_link() + for user, link in ( + (self.manager_a, self.link_a), + (self.expert_a, self.link_a), + (self.manager_b, link_b), + (self.expert_b, link_b), + ): + with self.subTest(user=user.pk): + data = self.detail(user)["partner_program"] + self.assertEqual(data, PartnerProgramProjectSerializer(link).data) + self.assertEqual(data["program_id"], link.partner_program_id) + + def test_program_role_scope_also_applies_to_public_published_projects(self): + link_b = self.add_second_link() + self.project.draft, self.project.is_public = False, True + self.project.save(update_fields=["draft", "is_public"]) + for user in (self.manager_b, self.expert_b): + with self.subTest(user=user.pk): + self.assertEqual( + self.detail(user)["partner_program"], + PartnerProgramProjectSerializer(link_b).data, + ) + + def test_minimum_eligible_link_not_first_global_or_program_id(self): + link_b = self.add_second_link() + # Program A exists first, but its replacement link is created last. + self.link_a.delete() + replacement = link_project_to_program(self.project, self.program_a) + self.program_b.managers.add(self.manager_a) + self.expert_a.expert.programs.add(self.program_b) + for user in (self.manager_a, self.expert_a): + with self.subTest(user=user.pk): + self.assertEqual( + self.detail(user)["partner_program"]["program_link_id"], link_b.pk + ) + self.program_b.managers.remove(self.manager_a) + self.expert_a.expert.programs.remove(self.program_b) + for user in (self.manager_a, self.expert_a): + with self.subTest(user=user.pk, only_a=True): + self.assertEqual( + self.detail(user)["partner_program"]["program_link_id"], + replacement.pk, + ) + + def test_independent_project_roles_use_global_minimum_even_with_program_role(self): + self.add_second_link() + for user in ( + self.project.leader, + self.collaborator, + self.invited, + self.staff, + self.superuser, + ): + self.program_b.managers.add(user) + with self.subTest(user=user.pk): + self.assertEqual( + self.detail(user)["partner_program"], + PartnerProgramProjectSerializer(self.link_a).data, + ) + + def test_existing_public_field_exposure_is_preserved(self): + self.project.draft, self.project.is_public = False, True + self.project.save(update_fields=["draft", "is_public"]) + expected = PartnerProgramProjectSerializer(self.link_a).data + self.assertTrue(expected["program_fields"]) + self.assertTrue(expected["program_field_values"]) + for user in (None, self.outsider, self.foreign_manager, self.foreign_expert): + with self.subTest(user=user): + self.assertEqual(self.detail(user)["partner_program"], expected) + + def test_published_private_project_stays_private(self): + self.project.draft = False + self.project.save(update_fields=["draft"]) + self.detail(self.manager_a) + self.detail(self.expert_a) + self.client.force_authenticate(self.outsider) + self.assertEqual(self.client.get(self.url).status_code, 403) + + def query_counts(self, users): + counts = [] + for user in users: + self.detail(user) # Warm existing view records and content-type cache. + ContentType.objects.get_for_model(self.project) + cache.clear() + with CaptureQueriesContext(connection) as queries: + self.detail(user) + counts.append(len(queries)) + return counts + + def test_detail_query_count_is_constant_for_one_and_eight_links(self): + users = (self.project.leader, self.manager_a, self.expert_a) + single = self.query_counts(users) + for _ in range(7): + program = create_partner_program() + program.managers.add(self.manager_a) + self.expert_a.expert.programs.add(program) + link = link_project_to_program(self.project, program) + field = create_program_field(program) + PartnerProgramFieldValue.objects.create( + program_project=link, field=field, value_text="Other program answer" + ) + self.assertEqual(self.project.program_links.count(), 8) + self.assertEqual(self.query_counts(users), single)