fix(api): restrict user profile updates to self (#11792)

api/CHANGELOG.md

@@ -8,6 +8,10 @@ All notable changes to the **Prowler API** are documented in this file.
 
 - Attack Paths: Scan rows now have database defaults for `is_migrated` and `sink_backend` so `scan-perform-scheduled` inserts survive deploy skew [(#11826)](https://github.com/prowler-cloud/prowler/pull/11826)
 
+### 🔐 Security
+
+- User profile updates now allow users to update their own account while requiring user-management permissions to update other users in the same tenant [(#11792)](https://github.com/prowler-cloud/prowler/pull/11792)
+
 ---
 
 ## [1.33.0] (Prowler v5.32.0)
@@ -27,9 +31,6 @@ All notable changes to the **Prowler API** are documented in this file.
 - Attack Paths: Provider graph cleanup now deletes Neo4j and Neptune relationships in directed batches before deleting nodes [(#11755)](https://github.com/prowler-cloud/prowler/pull/11755)
 - `scan-perform` no longer reports an error when a provider is deleted during a running scan [(#11696)](https://github.com/prowler-cloud/prowler/pull/11696)
 
-
----
-
 ## [1.32.1] (Prowler v5.31.1)
 
 ### 🐞 Fixed

api/src/backend/api/tests/test_rbac.py

@@ -103,20 +103,84 @@ class TestUserViewSet:
         assert response.json()["data"]["attributes"]["name"] == "Updated Name"
 
     def test_partial_update_user_with_no_permissions(
-        self, authenticated_client_no_permissions_rbac, create_test_user
+        self, authenticated_client_no_permissions_rbac, create_test_user_rbac_limited
     ):
         updated_data = {
             "data": {
                 "type": "users",
+                "id": str(create_test_user_rbac_limited.id),
                 "attributes": {"name": "Updated Name"},
             }
         }
         response = authenticated_client_no_permissions_rbac.patch(
-            reverse("user-detail", kwargs={"pk": create_test_user.id}),
+            reverse("user-detail", kwargs={"pk": create_test_user_rbac_limited.id}),
             data=updated_data,
-            format="vnd.api+json",
+            content_type="application/vnd.api+json",
         )
-        assert response.status_code == status.HTTP_403_FORBIDDEN
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["data"]["attributes"]["name"] == "Updated Name"
+
+    def test_partial_update_other_user_with_no_permissions_denied(
+        self, authenticated_client_no_permissions_rbac, tenants_fixture
+    ):
+        original_email = "target-rbac-update@example.com"
+        original_password = "OriginalPassword123@"
+        target_user = User.objects.create_user(
+            name="target_rbac_update",
+            email=original_email,
+            password=original_password,
+        )
+        Membership.objects.create(user=target_user, tenant=tenants_fixture[0])
+        updated_data = {
+            "data": {
+                "type": "users",
+                "id": str(target_user.id),
+                "attributes": {
+                    "email": "updated-target-rbac@example.com",
+                    "password": "UpdatedPassword123@",
+                },
+            }
+        }
+
+        response = authenticated_client_no_permissions_rbac.patch(
+            reverse("user-detail", kwargs={"pk": target_user.id}),
+            data=updated_data,
+            content_type="application/vnd.api+json",
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        target_user.refresh_from_db()
+        assert target_user.email == original_email
+        assert target_user.check_password(original_password)
+
+    def test_partial_update_other_user_with_manage_users_allowed(
+        self, authenticated_client_rbac_manage_users_only
+    ):
+        user = authenticated_client_rbac_manage_users_only.user
+        tenant = Membership.objects.filter(user=user).first().tenant
+        target_user = User.objects.create_user(
+            name="target_manage_users_update",
+            email="target-manage-users-update@example.com",
+            password="Password123@",
+        )
+        Membership.objects.create(user=target_user, tenant=tenant)
+        updated_data = {
+            "data": {
+                "type": "users",
+                "id": str(target_user.id),
+                "attributes": {"name": "Updated Target Name"},
+            }
+        }
+
+        response = authenticated_client_rbac_manage_users_only.patch(
+            reverse("user-detail", kwargs={"pk": target_user.id}),
+            data=updated_data,
+            content_type="application/vnd.api+json",
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        target_user.refresh_from_db()
+        assert target_user.name == "Updated Target Name"
 
     def test_delete_user_with_all_permissions(
         self, authenticated_client_rbac, create_test_user_rbac
@@ -540,9 +604,7 @@ class TestLimitedVisibility:
     TEST_PASSWORD = "Thisisapassword123@"
 
     @pytest.fixture
-    def limited_admin_user(
+    def limited_admin_user(self, django_db_blocker, tenants_fixture, providers_fixture):
-        self, django_db_setup, django_db_blocker, tenants_fixture, providers_fixture
-    ):
         with django_db_blocker.unblock():
             tenant = tenants_fixture[0]
             provider = providers_fixture[0]
@@ -626,10 +688,10 @@ class TestLimitedVisibility:
             response.json()["data"]["relationships"]["providers"]["meta"]["count"] == 1
         )
 
+    @pytest.mark.usefixtures("scan_summaries_fixture")
     def test_overviews_providers(
         self,
         authenticated_client_rbac_limited,
-        scan_summaries_fixture,
         providers_fixture,
     ):
         # By default, the associated provider is the one which has the overview data
@@ -648,6 +710,7 @@ class TestLimitedVisibility:
         assert response.status_code == status.HTTP_200_OK
         assert len(response.json()["data"]) == 0
 
+    @pytest.mark.usefixtures("scan_summaries_fixture")
     @pytest.mark.parametrize(
         "endpoint_name",
         [
@@ -659,7 +722,6 @@ class TestLimitedVisibility:
         self,
         endpoint_name,
         authenticated_client_rbac_limited,
-        scan_summaries_fixture,
         providers_fixture,
     ):
         # By default, the associated provider is the one which has the overview data
@@ -684,10 +746,10 @@ class TestLimitedVisibility:
         data = response.json()["data"]["attributes"].values()
         assert all(value == 0 for value in data)
 
+    @pytest.mark.usefixtures("scan_summaries_fixture")
     def test_overviews_services(
         self,
         authenticated_client_rbac_limited,
-        scan_summaries_fixture,
         providers_fixture,
     ):
         # By default, the associated provider is the one which has the overview data

api/src/backend/api/tests/test_views.py

@@ -244,6 +244,63 @@ class TestUserViewSet:
         create_test_user.refresh_from_db()
         assert create_test_user.company_name == new_company_name
 
+    def test_users_partial_update_same_tenant_other_user_password_denied(
+        self, authenticated_client_no_permissions_rbac, tenants_fixture
+    ):
+        original_password = "OriginalPassword123@"
+        new_password = "UpdatedPassword123@"
+        target_user = User.objects.create_user(
+            password=original_password,
+            email="target-password-update@example.com",
+        )
+        Membership.objects.create(user=target_user, tenant=tenants_fixture[0])
+        payload = {
+            "data": {
+                "type": "users",
+                "id": str(target_user.id),
+                "attributes": {"password": new_password},
+            },
+        }
+
+        response = authenticated_client_no_permissions_rbac.patch(
+            reverse("user-detail", kwargs={"pk": target_user.id}),
+            data=payload,
+            content_type="application/vnd.api+json",
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        target_user.refresh_from_db()
+        assert target_user.check_password(original_password)
+        assert not target_user.check_password(new_password)
+
+    def test_users_partial_update_same_tenant_other_user_email_denied(
+        self, authenticated_client_no_permissions_rbac, tenants_fixture
+    ):
+        original_email = "target-email-update@example.com"
+        new_email = "updated-target-email@example.com"
+        target_user = User.objects.create_user(
+            password="OriginalPassword123@",
+            email=original_email,
+        )
+        Membership.objects.create(user=target_user, tenant=tenants_fixture[0])
+        payload = {
+            "data": {
+                "type": "users",
+                "id": str(target_user.id),
+                "attributes": {"email": new_email},
+            },
+        }
+
+        response = authenticated_client_no_permissions_rbac.patch(
+            reverse("user-detail", kwargs={"pk": target_user.id}),
+            data=payload,
+            content_type="application/vnd.api+json",
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        target_user.refresh_from_db()
+        assert target_user.email == original_email
+
     def test_users_partial_update_invalid_content_type(
         self, authenticated_client, create_test_user
     ):
@@ -1491,13 +1548,13 @@ class TestProviderViewSet:
             ("provider_groups", ["provider-groups"]),
         ],
     )
+    @pytest.mark.usefixtures("create_provider_group_relationship")
     def test_providers_list_include(
         self,
         include_values,
         expected_resources,
         authenticated_client,
         providers_fixture,
-        create_provider_group_relationship,
     ):
         response = authenticated_client.get(
             reverse("provider-list"), {"include": include_values}
@@ -3542,7 +3599,7 @@ class TestScanViewSet:
         assert response.status_code == status.HTTP_404_NOT_FOUND
 
     @pytest.mark.parametrize(
-        "scan_json_payload, expected_scanner_args",
+        "scan_json_payload, _expected_scanner_args",
         [
             # Case 1: No scanner_args in payload (should use provider's scanner_args)
             (
@@ -3591,7 +3648,7 @@ class TestScanViewSet:
         mock_task_get,
         authenticated_client,
         scan_json_payload,
-        expected_scanner_args,
+        _expected_scanner_args,
         providers_fixture,
         tasks_fixture,
     ):
@@ -4070,7 +4127,7 @@ class TestScanViewSet:
 
         monkeypatch.setattr(
             "api.v1.views.env",
-            type("env", (), {"str": lambda self, *args, **kwargs: "test-bucket"})(),
+            type("env", (), {"str": lambda self, *_args, **_kwargs: "test-bucket"})(),
         )
 
         presigned_url = (
@@ -4176,7 +4233,7 @@ class TestScanViewSet:
 
         monkeypatch.setattr(
             "api.v1.views.TaskSerializer",
-            lambda *args, **kwargs: type("S", (), {"data": dummy}),
+            lambda *_args, **_kwargs: type("S", (), {"data": dummy}),
         )
 
         framework = get_compliance_frameworks(scan.provider.provider)[0]
@@ -4234,7 +4291,7 @@ class TestScanViewSet:
 
         monkeypatch.setattr(
             "api.v1.views.env",
-            type("env", (), {"str": lambda self, *args, **kwargs: "test-bucket"})(),
+            type("env", (), {"str": lambda self, *_args, **_kwargs: "test-bucket"})(),
         )
 
         match_key = "path/compliance/mitre_attack_aws.csv"
@@ -4245,6 +4302,7 @@ class TestScanViewSet:
 
         class FakeS3Client:
             def list_objects_v2(self, Bucket, Prefix):
+                del Prefix
                 return {"Contents": [{"Key": match_key}]}
 
             def generate_presigned_url(self, ClientMethod, Params, ExpiresIn):
@@ -4276,7 +4334,7 @@ class TestScanViewSet:
 
         monkeypatch.setattr(
             "api.v1.views.env",
-            type("env", (), {"str": lambda self, *args, **kwargs: "test-bucket"})(),
+            type("env", (), {"str": lambda self, *_args, **_kwargs: "test-bucket"})(),
         )
 
         old_key = "path/compliance/prowler-output-aws-20240101000000_cis_1.4_aws.csv"
@@ -4284,6 +4342,7 @@ class TestScanViewSet:
 
         class FakeS3Client:
             def list_objects_v2(self, Bucket, Prefix):
+                del Prefix
                 return {
                     "Contents": [
                         {
@@ -4357,11 +4416,12 @@ class TestScanViewSet:
 
         monkeypatch.setattr(
             "api.v1.views.env",
-            type("env", (), {"str": lambda self, *args, **kwargs: "test-bucket"})(),
+            type("env", (), {"str": lambda self, *_args, **_kwargs: "test-bucket"})(),
         )
 
         class FakeS3Client:
             def list_objects_v2(self, Bucket, Prefix):
+                del Prefix
                 return {"Contents": []}
 
             def get_object(self, Bucket, Key):
@@ -4547,7 +4607,7 @@ class TestScanViewSet:
             inserted_at=base + timedelta(hours=1)
         )
 
-        mock_task_serializer.side_effect = lambda instance, *a, **k: SimpleNamespace(
+        mock_task_serializer.side_effect = lambda instance, *_a, **_k: SimpleNamespace(
             data={"id": str(instance.id), "state": StateChoices.EXECUTING}
         )
 
@@ -6279,9 +6339,8 @@ class TestResourceViewSet:
         )
         assert response.status_code == status.HTTP_404_NOT_FOUND
 
-    def test_resources_metadata_retrieve(
+    @pytest.mark.usefixtures("backfill_scan_metadata_fixture")
-        self, authenticated_client, resources_fixture, backfill_scan_metadata_fixture
+    def test_resources_metadata_retrieve(self, authenticated_client, resources_fixture):
-    ):
         resource_1, *_ = resources_fixture
         response = authenticated_client.get(
             reverse("resource-metadata"),
@@ -6301,8 +6360,9 @@ class TestResourceViewSet:
         assert set(data["data"]["attributes"]["types"]) == expected_resource_types
         assert set(data["data"]["attributes"]["groups"]) == expected_groups
 
+    @pytest.mark.usefixtures("backfill_scan_metadata_fixture")
     def test_resources_metadata_resource_filter_retrieve(
-        self, authenticated_client, resources_fixture, backfill_scan_metadata_fixture
+        self, authenticated_client, resources_fixture
     ):
         resource_1, *_ = resources_fixture
         response = authenticated_client.get(
@@ -7796,9 +7856,8 @@ class TestFindingViewSet:
         )
         assert response.status_code == status.HTTP_404_NOT_FOUND
 
-    def test_findings_metadata_retrieve(
+    @pytest.mark.usefixtures("backfill_scan_metadata_fixture")
-        self, authenticated_client, findings_fixture, backfill_scan_metadata_fixture
+    def test_findings_metadata_retrieve(self, authenticated_client, findings_fixture):
-    ):
         finding_1, *_ = findings_fixture
         response = authenticated_client.get(
             reverse("finding-metadata"),
@@ -7821,8 +7880,9 @@ class TestFindingViewSet:
         )
         # assert data["data"]["attributes"]["tags"] == expected_tags
 
+    @pytest.mark.usefixtures("backfill_scan_metadata_fixture")
     def test_findings_metadata_resource_filter_retrieve(
-        self, authenticated_client, findings_fixture, backfill_scan_metadata_fixture
+        self, authenticated_client, findings_fixture
     ):
         finding_1, *_ = findings_fixture
         response = authenticated_client.get(
@@ -8008,9 +8068,8 @@ class TestFindingViewSet:
         attributes = response.json()["data"]["attributes"]
         assert set(attributes["categories"]) == {"gen-ai", "security"}
 
-    def test_findings_metadata_latest_categories(
+    @pytest.mark.usefixtures("latest_scan_finding_with_categories")
-        self, authenticated_client, latest_scan_finding_with_categories
+    def test_findings_metadata_latest_categories(self, authenticated_client):
-    ):
         response = authenticated_client.get(
             reverse("finding-metadata_latest"),
         )
@@ -8018,9 +8077,8 @@ class TestFindingViewSet:
         attributes = response.json()["data"]["attributes"]
         assert set(attributes["categories"]) == {"gen-ai", "iam"}
 
-    def test_findings_metadata_latest_groups(
+    @pytest.mark.usefixtures("latest_scan_finding_with_categories")
-        self, authenticated_client, latest_scan_finding_with_categories
+    def test_findings_metadata_latest_groups(self, authenticated_client):
-    ):
         response = authenticated_client.get(
             reverse("finding-metadata_latest"),
         )
@@ -10723,9 +10781,8 @@ class TestOverviewViewSet:
         response = authenticated_client.put(reverse("overview-list"))
         assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
 
-    def test_overview_providers_list(
+    @pytest.mark.usefixtures("scan_summaries_fixture")
-        self, authenticated_client, scan_summaries_fixture, resources_fixture
+    def test_overview_providers_list(self, authenticated_client, resources_fixture):
-    ):
         response = authenticated_client.get(reverse("overview-providers"))
         assert response.status_code == status.HTTP_200_OK
         assert len(response.json()["data"]) == 1
@@ -10736,10 +10793,10 @@ class TestOverviewViewSet:
         # Aggregated resources include all AWS providers present in the tenant
         assert response.json()["data"][0]["attributes"]["resources"]["total"] == 3
 
+    @pytest.mark.usefixtures("scan_summaries_fixture")
     def test_overview_providers_aggregates_same_provider_type(
         self,
         authenticated_client,
-        scan_summaries_fixture,
         resources_fixture,
         providers_fixture,
         tenants_fixture,
@@ -10790,10 +10847,10 @@ class TestOverviewViewSet:
         assert attributes["findings"]["muted"] == 7
         assert attributes["resources"]["total"] == 4
 
+    @pytest.mark.usefixtures("scan_summaries_fixture")
     def test_overview_providers_count(
         self,
         authenticated_client,
-        scan_summaries_fixture,
         resources_fixture,
         providers_fixture,
         tenants_fixture,
@@ -11259,15 +11316,15 @@ class TestOverviewViewSet:
         assert data[0]["id"] == str(snapshot1.id)
         assert data[0]["attributes"]["overall_score"] == "55.55"
 
-    def test_overview_services_list_no_required_filters(
+    @pytest.mark.usefixtures("scan_summaries_fixture")
-        self, authenticated_client, scan_summaries_fixture
+    def test_overview_services_list_no_required_filters(self, authenticated_client):
-    ):
         response = authenticated_client.get(reverse("overview-services"))
         assert response.status_code == status.HTTP_200_OK
         # Should return services from latest scans
         assert len(response.json()["data"]) == 2
 
-    def test_overview_regions_list(self, authenticated_client, scan_summaries_fixture):
+    @pytest.mark.usefixtures("scan_summaries_fixture")
+    def test_overview_regions_list(self, authenticated_client):
         response = authenticated_client.get(
             reverse("overview-regions"), {"filter[inserted_at]": TODAY}
         )
@@ -11293,7 +11350,8 @@ class TestOverviewViewSet:
         assert regions["aws:region2"]["fail"] == 1
         assert regions["aws:region2"]["muted"] == 3
 
-    def test_overview_services_list(self, authenticated_client, scan_summaries_fixture):
+    @pytest.mark.usefixtures("scan_summaries_fixture")
+    def test_overview_services_list(self, authenticated_client):
         response = authenticated_client.get(
             reverse("overview-services"), {"filter[inserted_at]": TODAY}
         )
@@ -11941,9 +11999,8 @@ class TestOverviewViewSet:
         assert results_by_type["internet-exposed"]["total_findings"] == 10
         assert results_by_type["internet-exposed"]["failed_findings"] == 5
 
-    def test_overview_services_region_filter(
+    @pytest.mark.usefixtures("scan_summaries_fixture")
-        self, authenticated_client, scan_summaries_fixture
+    def test_overview_services_region_filter(self, authenticated_client):
-    ):
         response = authenticated_client.get(
             reverse("overview-services"),
             {"filter[region]": "region1"},
@@ -12011,7 +12068,7 @@ class TestOverviewViewSet:
         assert "gcp-service" not in service_ids
 
     @pytest.mark.parametrize(
-        "status_filter,field_to_check",
+        "status_filter,_field_to_check",
         [
             ("FAIL", "fail"),
             ("PASS", "_pass"),
@@ -12023,7 +12080,7 @@ class TestOverviewViewSet:
         tenants_fixture,
         providers_fixture,
         status_filter,
-        field_to_check,
+        _field_to_check,
     ):
         tenant = tenants_fixture[0]
         provider = providers_fixture[0]
@@ -12663,8 +12720,9 @@ class TestOverviewViewSet:
         assert data[0]["attributes"]["new_failed_findings"] == 5
         assert data[0]["attributes"]["resources_count"] == 10
 
+    @pytest.mark.usefixtures("tenant_compliance_summary_fixture")
     def test_compliance_watchlist_no_filters_uses_tenant_summary(
-        self, authenticated_client, tenant_compliance_summary_fixture
+        self, authenticated_client
     ):
         response = authenticated_client.get(reverse("overview-compliance-watchlist"))
         assert response.status_code == status.HTTP_200_OK
@@ -12684,10 +12742,10 @@ class TestOverviewViewSet:
         assert by_id["gdpr_aws"]["requirements_failed"] == 0
         assert by_id["gdpr_aws"]["total_requirements"] == 7
 
+    @pytest.mark.usefixtures("provider_compliance_scores_fixture")
     def test_compliance_watchlist_with_provider_filter_uses_provider_scores(
         self,
         authenticated_client,
-        provider_compliance_scores_fixture,
         providers_fixture,
     ):
         provider1 = providers_fixture[0]
@@ -12704,9 +12762,8 @@ class TestOverviewViewSet:
         assert by_id["aws_cis_2.0"]["requirements_manual"] == 1
         assert by_id["aws_cis_2.0"]["total_requirements"] == 3
 
-    def test_compliance_watchlist_fail_dominant_logic(
+    @pytest.mark.usefixtures("provider_compliance_scores_fixture")
-        self, authenticated_client, provider_compliance_scores_fixture
+    def test_compliance_watchlist_fail_dominant_logic(self, authenticated_client):
-    ):
         response = authenticated_client.get(
             f"{reverse('overview-compliance-watchlist')}?filter[provider_type]=aws"
         )
@@ -12721,10 +12778,10 @@ class TestOverviewViewSet:
         assert aws_cis["requirements_manual"] == 1
         assert aws_cis["total_requirements"] == 3
 
+    @pytest.mark.usefixtures("provider_compliance_scores_fixture")
     def test_compliance_watchlist_provider_id_in_filter(
         self,
         authenticated_client,
-        provider_compliance_scores_fixture,
         providers_fixture,
     ):
         provider1, provider2, *_ = providers_fixture
@@ -12737,10 +12794,10 @@ class TestOverviewViewSet:
         data = response.json()["data"]
         assert len(data) >= 1
 
+    @pytest.mark.usefixtures("provider_compliance_scores_fixture")
     def test_compliance_watchlist_provider_groups_filter(
         self,
         authenticated_client,
-        provider_compliance_scores_fixture,
         providers_fixture,
         provider_groups_fixture,
         tenants_fixture,
@@ -18292,10 +18349,10 @@ class TestFindingGroupViewSet:
         ],
         ids=["summary_path", "finding_level_path"],
     )
+    @pytest.mark.usefixtures("finding_groups_title_variants_fixture")
     def test_check_title_icontains_includes_all_title_variants(
         self,
         authenticated_client,
-        finding_groups_title_variants_fixture,
         extra_filters,
     ):
         """

api/src/backend/api/v1/views.py

@@ -948,8 +948,8 @@ class UserViewSet(BaseUserViewset):
         """
         Returns the required permissions based on the request method.
         """
-        if self.action == "me":
+        if self.action in ["me", "partial_update"]:
-            # No permissions required for me request
+            # No permissions required for me and partial_update requests
             self.required_permissions = []
         else:
             # Require permission for the rest of the requests
@@ -1003,6 +1003,24 @@ class UserViewSet(BaseUserViewset):
             status=status.HTTP_200_OK,
         )
 
+    def partial_update(self, request, *args, **kwargs):
+        user = self.get_object()
+        if user.id != self.request.user.id:
+            role = get_role(self.request.user, self.request.tenant_id)
+            if not getattr(role, Permissions.MANAGE_USERS.value, False):
+                raise ValidationError(
+                    "Only users with manage users permission can update other users."
+                )
+
+        serializer = self.get_serializer(user, data=request.data, partial=True)
+        serializer.is_valid(raise_exception=True)
+        self.perform_update(serializer)
+
+        if getattr(user, "_prefetched_objects_cache", None):
+            user._prefetched_objects_cache = {}
+
+        return Response(serializer.data)
+
     def destroy(self, request, *args, **kwargs):
         if kwargs["pk"] != str(self.request.user.id):
             raise ValidationError("Only the current user can be deleted.")

docs/user-guide/tutorials/prowler-app-rbac.mdx

@@ -40,6 +40,11 @@ Follow these steps to edit a user of your account:
 
     <img src="/images/prowler-app/rbac/user_edit_details.png" alt="Edit User Details" width="700" />
 
+<Note>
+Users can edit their own account details. Editing another user's account details requires the **Invite and Manage Users** or **admin** permission.
+
+</Note>
+
 #### Removing a User
 
 Follow these steps to remove a user of your account: