fix: ensure SimpleFeatureStateViewSet uses correct permissions for se…

…gment overrides (#2990)

api/conftest.py

@@ -181,14 +181,14 @@ def environment(project):
 @pytest.fixture()
 def with_environment_permissions(
     environment: Environment, staff_user: FFAdminUser
-) -> typing.Callable[[list[str], int], None]:
+) -> typing.Callable[[list[str], int | None], UserEnvironmentPermission]:
     """
     Add environment permissions to the staff_user fixture.
     Defaults to associating to the environment fixture.
     """
 
     def _with_environment_permissions(
-        permission_keys: list[str], environment_id: typing.Optional[int] = None
+        permission_keys: list[str], environment_id: int | None = None
     ) -> UserEnvironmentPermission:
         environment_id = environment_id or environment.id
         uep, __ = UserEnvironmentPermission.objects.get_or_create(

api/features/permissions.py

@@ -2,6 +2,8 @@
 
 from django.shortcuts import get_object_or_404
 from rest_framework.permissions import IsAuthenticated
+from rest_framework.request import Request
+from rest_framework.viewsets import GenericViewSet
 
 from environments.models import Environment
 from environments.permissions.constants import MANAGE_SEGMENT_OVERRIDES
@@ -12,7 +14,7 @@
     UPDATE_FEATURE_STATE,
     VIEW_ENVIRONMENT,
 )
-from features.models import Feature
+from features.models import Feature, FeatureState
 from projects.models import Project
 from projects.permissions import CREATE_FEATURE, DELETE_FEATURE
 from projects.permissions import (
@@ -77,11 +79,7 @@ def has_object_permission(self, request, view, obj):
 
 
 class FeatureStatePermissions(IsAuthenticated):
-    def has_permission(self, request, view):
-        action_permission_map = {
-            "list": VIEW_ENVIRONMENT,
-            "create": UPDATE_FEATURE_STATE,
-        }
+    def has_permission(self, request: Request, view: GenericViewSet) -> bool:
         if not super().has_permission(request, view):
             return False
 
@@ -98,7 +96,17 @@ def has_permission(self, request, view):
                 environment = Environment.objects.get(id=int(environment))
 
                 tag_ids = None
-                required_permission = action_permission_map.get(view.action)
+
+                if view.action == "list":
+                    required_permission = VIEW_ENVIRONMENT
+                elif (
+                    view.action == "create"
+                    and request.data.get("feature_segment") is not None
+                ):
+                    required_permission = MANAGE_SEGMENT_OVERRIDES
+                else:
+                    required_permission = UPDATE_FEATURE_STATE
+
                 if required_permission in TAG_SUPPORTED_ENVIRONMENT_PERMISSIONS:
                     feature_id = request.data.get("feature")
                     feature = Feature.objects.get(id=feature_id)
@@ -113,13 +121,19 @@ def has_permission(self, request, view):
         except (Environment.DoesNotExist, Feature.DoesNotExist):
             return False
 
-    def has_object_permission(self, request, view, obj):
+    def has_object_permission(
+        self, request: Request, view: GenericViewSet, obj: FeatureState
+    ) -> bool:
+        permission = (
+            MANAGE_SEGMENT_OVERRIDES if obj.feature_segment_id else UPDATE_FEATURE_STATE
+        )
+
         tag_ids = None
-        if UPDATE_FEATURE_STATE in TAG_SUPPORTED_ENVIRONMENT_PERMISSIONS:
+        if permission in TAG_SUPPORTED_ENVIRONMENT_PERMISSIONS:
             tag_ids = list(obj.feature.tags.values_list("id", flat=True))
 
         return request.user.has_environment_permission(
-            UPDATE_FEATURE_STATE, environment=obj.environment, tag_ids=tag_ids
+            permission, environment=obj.environment, tag_ids=tag_ids
         )
 
 

api/tests/unit/features/test_unit_features_views.py

@@ -1,16 +1,23 @@
 import json
 import uuid
+from typing import Callable
 
 import pytest
 from django.urls import reverse
 from django.utils import timezone
 from pytest_lazyfixture import lazy_fixture
 from rest_framework import status
+from rest_framework.test import APIClient
 
 from audit.constants import FEATURE_DELETED_MESSAGE
 from audit.models import AuditLog, RelatedObjectType
 from environments.identities.models import Identity
 from environments.models import Environment
+from environments.permissions.constants import (
+    MANAGE_SEGMENT_OVERRIDES,
+    UPDATE_FEATURE_STATE,
+)
+from environments.permissions.models import UserEnvironmentPermission
 from features.feature_types import MULTIVARIATE
 from features.models import Feature, FeatureSegment, FeatureState
 from features.multivariate.models import MultivariateFeatureOption
@@ -1264,3 +1271,161 @@ def test_cannot_update_feature_of_a_feature_state(
     assert (
         response.json()["feature"][0] == "Cannot change the feature of a feature state"
     )
+
+
+def test_create_segment_override__using_simple_feature_state_viewset__allows_manage_segment_overrides(
+    staff_client: APIClient,
+    with_environment_permissions: Callable[
+        [list[str], int | None], UserEnvironmentPermission
+    ],
+    environment: Environment,
+    feature: Feature,
+    segment: Segment,
+    feature_segment: FeatureSegment,
+) -> None:
+    # Given
+    with_environment_permissions([MANAGE_SEGMENT_OVERRIDES])
+
+    url = reverse("api-v1:features:featurestates-list")
+
+    data = {
+        "feature": feature.id,
+        "environment": environment.id,
+        "feature_segment": feature_segment.id,
+        "enabled": True,
+        "feature_state_value": {
+            "type": "unicode",
+            "string_value": "foo",
+        },
+    }
+
+    # When
+    response = staff_client.post(
+        url, data=json.dumps(data), content_type="application/json"
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_201_CREATED
+
+    assert FeatureState.objects.filter(
+        feature=feature, environment=environment, feature_segment=feature_segment
+    ).exists()
+
+
+def test_create_segment_override__using_simple_feature_state_viewset__denies_update_feature_state(
+    staff_client: APIClient,
+    with_environment_permissions: Callable[
+        [list[str], int | None], UserEnvironmentPermission
+    ],
+    environment: Environment,
+    feature: Feature,
+    segment: Segment,
+    feature_segment: FeatureSegment,
+) -> None:
+    # Given
+    with_environment_permissions([UPDATE_FEATURE_STATE])
+
+    url = reverse("api-v1:features:featurestates-list")
+
+    data = {
+        "feature": feature.id,
+        "environment": environment.id,
+        "feature_segment": feature_segment.id,
+        "enabled": True,
+        "feature_state_value": {
+            "type": "unicode",
+            "string_value": "foo",
+        },
+    }
+
+    # When
+    response = staff_client.post(
+        url, data=json.dumps(data), content_type="application/json"
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+def test_update_segment_override__using_simple_feature_state_viewset__allows_manage_segment_overrides(
+    staff_client: APIClient,
+    with_environment_permissions: Callable[
+        [list[str], int | None], UserEnvironmentPermission
+    ],
+    environment: Environment,
+    feature: Feature,
+    segment: Segment,
+    feature_segment: FeatureSegment,
+    segment_featurestate: FeatureState,
+) -> None:
+    # Given
+    with_environment_permissions([MANAGE_SEGMENT_OVERRIDES])
+
+    url = reverse(
+        "api-v1:features:featurestates-detail", args=[segment_featurestate.id]
+    )
+
+    data = {
+        "feature": feature.id,
+        "environment": environment.id,
+        "feature_segment": feature_segment.id,
+        "enabled": True,
+        "feature_state_value": {
+            "type": "unicode",
+            "string_value": "foo",
+        },
+    }
+
+    # When
+    response = staff_client.put(
+        url, data=json.dumps(data), content_type="application/json"
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+
+    assert FeatureState.objects.filter(
+        feature=feature,
+        environment=environment,
+        feature_segment=feature_segment,
+        enabled=True,
+        feature_state_value__string_value="foo",
+    ).exists()
+
+
+def test_update_segment_override__using_simple_feature_state_viewset__denies_update_feature_state(
+    staff_client: APIClient,
+    with_environment_permissions: Callable[
+        [list[str], int | None], UserEnvironmentPermission
+    ],
+    environment: Environment,
+    feature: Feature,
+    segment: Segment,
+    feature_segment: FeatureSegment,
+    segment_featurestate: FeatureState,
+) -> None:
+    # Given
+    with_environment_permissions([UPDATE_FEATURE_STATE])
+
+    url = reverse(
+        "api-v1:features:featurestates-detail", args=[segment_featurestate.id]
+    )
+
+    data = {
+        "feature": feature.id,
+        "environment": environment.id,
+        "feature_segment": feature_segment.id,
+        "enabled": True,
+        "feature_state_value": {
+            "type": "unicode",
+            "string_value": "foo",
+        },
+    }
+
+    # When
+    response = staff_client.put(
+        url, data=json.dumps(data), content_type="application/json"
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_403_FORBIDDEN