fix: prevent unauthorised access from group permissions (#5893)

Author: matthewelwell

api/conftest.py

@@ -265,8 +265,10 @@ def staff_client(staff_user):  # type: ignore[no-untyped-def]
 
 
 @pytest.fixture()
-def organisation(db, admin_user, staff_user):  # type: ignore[no-untyped-def]
-    org = Organisation.objects.create(name="Test Org")
+def organisation(
+    db: None, admin_user: FFAdminUser, staff_user: FFAdminUser
+) -> Organisation:
+    org: Organisation = Organisation.objects.create(name="Test Org")
     admin_user.add_organisation(org, role=OrganisationRole.ADMIN)
     staff_user.add_organisation(org, role=OrganisationRole.USER)
     return org

api/e2etests/e2e_seed_data.py

@@ -68,7 +68,7 @@ def seed_data() -> None:
         password=PASSWORD,
         username=settings.E2E_USER,
     )
-    org_admin.add_organisation(organisation, OrganisationRole.ADMIN)  # type: ignore[no-untyped-call]
+    org_admin.add_organisation(organisation, OrganisationRole.ADMIN)
     non_admin_user_with_org_permissions: FFAdminUser = FFAdminUser.objects.create_user(  # type: ignore[no-untyped-call]  # noqa: E501
         email=settings.E2E_NON_ADMIN_USER_WITH_ORG_PERMISSIONS,
         password=PASSWORD,
@@ -87,18 +87,10 @@ def seed_data() -> None:
         email=settings.E2E_NON_ADMIN_USER_WITH_A_ROLE,
         password=PASSWORD,
     )
-    non_admin_user_with_org_permissions.add_organisation(  # type: ignore[no-untyped-call]
-        organisation,
-    )
-    non_admin_user_with_project_permissions.add_organisation(  # type: ignore[no-untyped-call]
-        organisation,
-    )
-    non_admin_user_with_env_permissions.add_organisation(  # type: ignore[no-untyped-call]
-        organisation,
-    )
-    non_admin_user_with_a_role.add_organisation(  # type: ignore[no-untyped-call]
-        organisation,
-    )
+    non_admin_user_with_org_permissions.add_organisation(organisation)
+    non_admin_user_with_project_permissions.add_organisation(organisation)
+    non_admin_user_with_env_permissions.add_organisation(organisation)
+    non_admin_user_with_a_role.add_organisation(organisation)
 
     # Add permissions to the non-admin user with org permissions
     user_org_permission = UserOrganisationPermission.objects.create(

api/permissions/permission_service.py

@@ -86,16 +86,25 @@ def get_permitted_projects_for_user(
         tag_ids=tag_ids,
     )
 
-    organisation_filter = Q(
+    # The user has access to any projects belonging to organisations
+    # they are an admin of
+    admin_organisations_filter = Q(
         organisation__userorganisation__user=user,
         organisation__userorganisation__role=OrganisationRole.ADMIN.name,
     )
-    project_ids_from_organisation = Project.objects.filter(
-        organisation_filter
+    project_ids_from_admin_organisations = Project.objects.filter(
+        admin_organisations_filter
     ).values_list("id", flat=True)
 
-    project_ids = project_ids_from_base_filter | set(project_ids_from_organisation)
-    return Project.objects.filter(id__in=project_ids)
+    project_ids = project_ids_from_base_filter | set(
+        project_ids_from_admin_organisations
+    )
+    queryset = Project.objects.filter(id__in=project_ids)
+
+    # Final check to ensure that the user is a member of the organisation
+    queryset = queryset.filter(organisation__users=user)
+
+    return queryset
 
 
 def get_permitted_projects_for_master_api_key(
@@ -154,6 +163,9 @@ def get_permitted_environments_for_user(
     if prefetch_metadata:
         queryset = queryset.prefetch_related("metadata")
 
+    # Final check to ensure the user is a member of the organisation
+    queryset = queryset.filter(project__organisation__users=user)
+
     # Description is defered due to Oracle support where a
     # query can't have a where clause if description is in
     # the select parameters. This leads to an N+1 query for
@@ -199,7 +211,12 @@ def user_has_organisation_permission(
     )
     filter_ = base_filter & Q(id=organisation.id)
 
-    return Organisation.objects.filter(filter_).exists()  # type: ignore[no-any-return]
+    queryset = Organisation.objects.filter(filter_)
+
+    # Final check to verify that user belongs to organisation
+    queryset = queryset.filter(users=user)
+
+    return queryset.exists()  # type: ignore[no-any-return]
 
 
 def master_api_key_has_organisation_permission(
@@ -219,6 +236,14 @@ def _is_user_object_admin(
     ModelClass = type(object_)
     base_filter = get_base_permission_filter(user, ModelClass)  # type: ignore[arg-type]
     filter_ = base_filter & Q(id=object_.id)
+
+    if ModelClass is Project:
+        filter_ = filter_ & Q(organisation__users=user)
+    elif ModelClass is Environment:
+        filter_ = filter_ & Q(project__organisation__users=user)
+    else:  # pragma: no cover
+        raise ValueError(f"Unexpected object type {type(object_)}")
+
     return ModelClass.objects.filter(filter_).exists()
 
 

api/tests/unit/environments/identities/traits/test_traits_views.py

@@ -3,10 +3,8 @@
 
 from common.environments.permissions import (
     MANAGE_IDENTITIES,
-    VIEW_ENVIRONMENT,
     VIEW_IDENTITIES,
 )
-from common.projects.permissions import VIEW_PROJECT
 from django.test import override_settings
 from django.urls import reverse
 from rest_framework import status
@@ -26,6 +24,7 @@
 from organisations.models import Organisation
 from permissions.models import PermissionModel
 from projects.models import Project, UserProjectPermission
+from users.models import FFAdminUser
 
 
 def test_can_set_trait_for_an_identity(
@@ -970,27 +969,21 @@ def test_set_trait_for_an_identity_is_not_throttled_by_user_throttle(  # type: i
         assert res.status_code == status.HTTP_200_OK
 
 
-def test_user_with_manage_identities_permission_can_add_trait_for_identity(  # type: ignore[no-untyped-def]
-    environment, identity, django_user_model, api_client
-):
+def test_user_with_manage_identities_permission_can_add_trait_for_identity(
+    environment: Environment,
+    identity: Identity,
+    staff_user: FFAdminUser,
+    staff_client: APIClient,
+    user_environment_permission: UserEnvironmentPermission,
+    user_project_permission: UserProjectPermission,
+    view_environment_permission: PermissionModel,
+    manage_identities_permission: PermissionModel,
+    view_project_permission: PermissionModel,
+) -> None:
     # Given
-    user = django_user_model.objects.create(email="user@example.com")
-    api_client.force_authenticate(user)
-
-    view_environment_permission = PermissionModel.objects.get(key=VIEW_ENVIRONMENT)
-    manage_identities_permission = PermissionModel.objects.get(key=MANAGE_IDENTITIES)
-    view_project_permission = PermissionModel.objects.get(key=VIEW_PROJECT)
-
-    user_environment_permission = UserEnvironmentPermission.objects.create(
-        user=user, environment=environment
-    )
     user_environment_permission.permissions.add(
         view_environment_permission, manage_identities_permission
     )
-
-    user_project_permission = UserProjectPermission.objects.create(
-        user=user, project=environment.project
-    )
     user_project_permission.permissions.add(view_project_permission)
 
     url = reverse(
@@ -999,7 +992,7 @@ def test_user_with_manage_identities_permission_can_add_trait_for_identity(  # t
     )
 
     # When
-    response = api_client.post(
+    response = staff_client.post(
         url, data={"trait_key": "foo", "value_type": "unicode", "string_value": "foo"}
     )
 

api/tests/unit/environments/permissions/test_unit_environments_permissions.py

@@ -3,6 +3,7 @@
 from common.projects.permissions import (
     CREATE_ENVIRONMENT,
 )
+from pytest_mock import MockerFixture
 
 from environments.identities.models import Identity
 from environments.models import Environment
@@ -45,24 +46,28 @@ def test_environment_admin_permissions_has_permissions_returns_false_for_non_adm
     assert has_permission is False
 
 
-def test_environment_admin_permissions_has_permissions_returns_true_for_admin_user(  # type: ignore[no-untyped-def]
-    environment, django_user_model, mocker
+def test_environment_admin_permissions_has_permissions_returns_true_for_admin_user(
+    environment: Environment,
+    staff_user: FFAdminUser,
+    user_environment_permission: UserEnvironmentPermission,
+    mocker: MockerFixture,
 ) -> None:
     # Given
-    user = django_user_model.objects.create(username="test_user")
-    UserEnvironmentPermission.objects.create(
-        user=user, environment=environment, admin=True
-    )
     mocked_request = mocker.MagicMock()
-    mocked_request.user = user
+    mocked_request.user = staff_user
 
     mocked_view = mocker.MagicMock()
     mocked_view.kwargs = {"environment_api_key": environment.api_key}
 
+    user_environment_permission.admin = True
+    user_environment_permission.save()
+
     # When
     has_permission = environment_admin_permissions.has_permission(  # type: ignore[no-untyped-call]
         mocked_request, mocked_view
     )
+
+    # Then
     assert has_permission is True
 
 

api/tests/unit/environments/test_unit_environments_permissions.py

@@ -1,6 +1,11 @@
+from django.test import RequestFactory
+from pytest_mock import MockerFixture
+
+from environments.models import Environment
 from environments.permissions.models import UserEnvironmentPermission
 from environments.permissions.permissions import NestedEnvironmentPermissions
 from permissions.models import ENVIRONMENT_PERMISSION_TYPE, PermissionModel
+from users.models import FFAdminUser
 
 
 def test_nested_environment_permissions_has_permission_false_if_no_env_key(  # type: ignore[no-untyped-def]
@@ -19,28 +24,27 @@ def test_nested_environment_permissions_has_permission_false_if_no_env_key(  # t
     assert result is False
 
 
-def test_nested_environment_permissions_has_permission_true_if_action_in_map(  # type: ignore[no-untyped-def]
-    rf, mocker, db, environment, django_user_model
-):
+def test_nested_environment_permissions_has_permission_true_if_action_in_map(
+    rf: RequestFactory,
+    mocker: MockerFixture,
+    environment: Environment,
+    staff_user: FFAdminUser,
+    user_environment_permission: UserEnvironmentPermission,
+) -> None:
     # Given
     permission_key = "SOME_PERMISSION"
     permission = PermissionModel.objects.create(
         key=permission_key, type=ENVIRONMENT_PERMISSION_TYPE, description="foobar"
     )
+    user_environment_permission.permissions.add(permission)
 
     action = "retrieve"
     permissions = NestedEnvironmentPermissions(
         action_permission_map={action: permission.key}
     )
 
-    user = django_user_model.objects.create(email="test@example.com")
-    user_env_permission = UserEnvironmentPermission.objects.create(
-        user=user, environment=environment
-    )
-    user_env_permission.permissions.add(permission)
-
     request = rf.get("/")
-    request.user = user
+    request.user = staff_user
     view = mocker.MagicMock(
         action=action, kwargs={"environment_api_key": environment.api_key}
     )
@@ -52,19 +56,21 @@ def test_nested_environment_permissions_has_permission_true_if_action_in_map(  #
     assert has_permission is True
 
 
-def test_nested_environment_permissions_has_permission_if_create_and_user_is_admin(  # type: ignore[no-untyped-def]
-    rf, mocker, db, environment, django_user_model
-):
+def test_nested_environment_permissions_has_permission_if_create_and_user_is_admin(
+    rf: RequestFactory,
+    mocker: MockerFixture,
+    environment: Environment,
+    staff_user: FFAdminUser,
+    user_environment_permission: UserEnvironmentPermission,
+) -> None:
     # Given
     permissions = NestedEnvironmentPermissions()
 
-    user = django_user_model.objects.create(email="test@example.com")
-    UserEnvironmentPermission.objects.create(
-        user=user, environment=environment, admin=True
-    )
+    user_environment_permission.admin = True
+    user_environment_permission.save()
 
     request = rf.get("/")
-    request.user = user
+    request.user = staff_user
     view = mocker.MagicMock(
         action="create", kwargs={"environment_api_key": environment.api_key}
     )
@@ -76,9 +82,13 @@ def test_nested_environment_permissions_has_permission_if_create_and_user_is_adm
     assert has_permission is True
 
 
-def test_nested_environment_permissions_has_object_permission_true_if_action_in_map(  # type: ignore[no-untyped-def]
-    rf, mocker, django_user_model, environment
-):
+def test_nested_environment_permissions_has_object_permission_true_if_action_in_map(
+    rf: RequestFactory,
+    mocker: MockerFixture,
+    environment: Environment,
+    staff_user: FFAdminUser,
+    user_environment_permission: UserEnvironmentPermission,
+) -> None:
     # Given
     permission_key = "SOME_PERMISSION"
     permission = PermissionModel.objects.create(
@@ -90,14 +100,10 @@ def test_nested_environment_permissions_has_object_permission_true_if_action_in_
         action_permission_map={action: permission.key}
     )
 
-    user = django_user_model.objects.create(email="test@example.com")
-    user_env_permission = UserEnvironmentPermission.objects.create(
-        user=user, environment=environment
-    )
-    user_env_permission.permissions.add(permission)
+    user_environment_permission.permissions.add(permission)
 
     request = rf.get("/")
-    request.user = user
+    request.user = staff_user
     view = mocker.MagicMock(
         action=action, kwargs={"environment_api_key": environment.api_key}
     )
@@ -111,19 +117,21 @@ def test_nested_environment_permissions_has_object_permission_true_if_action_in_
     assert has_object_permission is True
 
 
-def test_nested_environment_permissions_has_object_permission_true_if_user_is_admin(  # type: ignore[no-untyped-def]
-    rf, mocker, django_user_model, environment
-):
+def test_nested_environment_permissions_has_object_permission_true_if_user_is_admin(
+    rf: RequestFactory,
+    mocker: MockerFixture,
+    environment: Environment,
+    staff_user: FFAdminUser,
+    user_environment_permission: UserEnvironmentPermission,
+) -> None:
     # Given
     permissions = NestedEnvironmentPermissions()
 
-    user = django_user_model.objects.create(email="test@example.com")
-    UserEnvironmentPermission.objects.create(
-        user=user, environment=environment, admin=True
-    )
+    user_environment_permission.admin = True
+    user_environment_permission.save()
 
     request = rf.get("/")
-    request.user = user
+    request.user = staff_user
     view = mocker.MagicMock(
         action="action", kwargs={"environment_api_key": environment.api_key}
     )

api/tests/unit/features/test_unit_features_views.py

@@ -1051,6 +1051,7 @@ def test_create_segment_override_staff(
     staff_user: FFAdminUser,
     staff_client: APIClient,
     manage_segment_overrides_permission: PermissionModel,
+    user_environment_permission: UserEnvironmentPermission,
 ) -> None:
     # Given
     url = reverse(
@@ -1066,9 +1067,6 @@ def test_create_segment_override_staff(
         "enabled": enabled,
         "feature_segment": {"segment": segment.id},
     }
-    user_environment_permission = UserEnvironmentPermission.objects.create(
-        user=staff_user, admin=False, environment=environment
-    )
     user_environment_permission.permissions.add(manage_segment_overrides_permission)
 
     response = staff_client.post(
@@ -1160,7 +1158,7 @@ def test_list_feature_states_from_simple_view_set(
     # add another organisation with a project, environment and feature (which should be
     # excluded)
     another_organisation = Organisation.objects.create(name="another_organisation")
-    admin_user.add_organisation(another_organisation)  # type: ignore[no-untyped-call]
+    admin_user.add_organisation(another_organisation)
     another_project = Project.objects.create(
         name="another_project", organisation=another_organisation
     )