fix(api): assign fallback role to SAML users (#12223)

api/changelog.d/saml-missing-usertype-read-only.fixed.md

@@ -0,0 +1 @@
+SAML users without a `userType` attribute and without an existing role in the SAML tenant now receive a least-privilege `read_only` fallback role; a numeric suffix is used when that name belongs to a role with different permissions

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

@@ -3,9 +3,11 @@ import io
 import json
 import os
 import tempfile
+from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, date, datetime, timedelta
 from decimal import Decimal
 from pathlib import Path
+from threading import Event, Lock
 from types import SimpleNamespace
 from unittest.mock import ANY, MagicMock, Mock, patch
 from urllib.parse import parse_qs, urlparse
@@ -74,7 +76,7 @@ from conftest import (
     today_after_n_days,
 )
 from django.conf import settings
-from django.db import connection
+from django.db import close_old_connections, connection
 from django.db.models import Count
 from django.http import JsonResponse
 from django.test import RequestFactory
@@ -14843,19 +14845,94 @@ class TestTenantFinishACSView:
         # Verify no new role was created
         assert Role.objects.using(MainRouter.admin_db).count() == roles_before
 
-    def test_dispatch_assigns_no_role_to_new_user_when_usertype_missing(
+    @pytest.mark.parametrize(
+        (
+            "existing_role_attributes",
+            "existing_suffixes",
+            "expected_role_name",
+            "expected_role_created",
+        ),
+        [
+            (None, (), "read_only", True),
+            ({"unlimited_visibility": True}, (), "read_only", False),
+            (
+                {"manage_users": True, "unlimited_visibility": True},
+                (
+                    ("read_only_0", {"unlimited_visibility": True}),
+                    ("read_only_1", {"unlimited_visibility": True}),
+                ),
+                "read_only_0",
+                False,
+            ),
+            (
+                {"manage_users": True, "unlimited_visibility": True},
+                (
+                    (
+                        "read_only_0",
+                        {"manage_users": True, "unlimited_visibility": True},
+                    ),
+                    ("read_only_1", {"unlimited_visibility": True}),
+                ),
+                "read_only_1",
+                False,
+            ),
+            ({"unlimited_visibility": False}, (), "read_only_0", True),
+        ],
+        ids=[
+            "creates-role",
+            "reuses-safe-role",
+            "reuses-first-safe-suffixed-role",
+            "skips-unsafe-suffixed-role",
+            "avoids-restricted-visibility",
+        ],
+    )
+    def test_dispatch_assigns_read_only_role_when_usertype_missing(
         self,
         create_test_user,
         tenants_fixture,
         saml_setup,
         settings,
         monkeypatch,
+        existing_role_attributes,
+        existing_suffixes,
+        expected_role_name,
+        expected_role_created,
     ):
-        """Test that a user without roles gets none assigned when userType is missing"""
+        """Test safe fallback role assignment when userType is missing"""
         monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
         user = create_test_user
         tenant = tenants_fixture[0]
-        roles_before = Role.objects.using(MainRouter.admin_db).count()
+        other_tenant = tenants_fixture[1]
+
+        other_tenant_role = Role.objects.using(MainRouter.admin_db).create(
+            name="read_only",
+            tenant=other_tenant,
+            unlimited_visibility=True,
+        )
+        other_tenant_relationship = UserRoleRelationship.objects.using(
+            MainRouter.admin_db
+        ).create(
+            user=user,
+            role=other_tenant_role,
+            tenant=other_tenant,
+        )
+
+        existing_role = None
+        if existing_role_attributes is not None:
+            existing_role = Role.objects.using(MainRouter.admin_db).create(
+                name="read_only",
+                tenant=tenant,
+                **existing_role_attributes,
+            )
+        for role_name, role_attributes in existing_suffixes:
+            Role.objects.using(MainRouter.admin_db).create(
+                name=role_name,
+                tenant=tenant,
+                **role_attributes,
+            )
+        roles_before = (
+            Role.objects.using(MainRouter.admin_db).filter(tenant=tenant).count()
+        )
 
         social_account = SocialAccount(
             user=user,
@@ -14908,12 +14985,44 @@ class TestTenantFinishACSView:
 
         assert response.status_code == 302
 
-        # Verify no role was created or assigned
-        assert Role.objects.using(MainRouter.admin_db).count() == roles_before
-        assert not (
+        # Verify the fallback role was created or reused with read-only access
+        expected_role_count = roles_before + expected_role_created
+        assert (
+            Role.objects.using(MainRouter.admin_db).filter(tenant=tenant).count()
+            == expected_role_count
+        )
+        role = Role.objects.using(MainRouter.admin_db).get(
+            name=expected_role_name, tenant=tenant
+        )
+        if existing_role is not None and expected_role_name == "read_only":
+            assert role == existing_role
+        assert not role.manage_users
+        assert not role.manage_account
+        assert not role.manage_billing
+        assert not role.manage_providers
+        assert not role.manage_integrations
+        assert not role.manage_scans
+        assert role.unlimited_visibility
+        assert (
+            UserRoleRelationship.objects.using(MainRouter.admin_db)
+            .filter(user=user, role=role, tenant_id=tenant.id)
+            .exists()
+        )
+        assert (
+            UserRoleRelationship.objects.using(MainRouter.admin_db)
+            .filter(
+                id=other_tenant_relationship.id,
+                user=user,
+                role=other_tenant_role,
+                tenant_id=other_tenant.id,
+            )
+            .exists()
+        )
+        assert (
             UserRoleRelationship.objects.using(MainRouter.admin_db)
             .filter(user=user, tenant_id=tenant.id)
-            .exists()
+            .count()
+            == 1
         )
 
         # Membership is still created so the user belongs to the tenant
@@ -14923,6 +15032,131 @@ class TestTenantFinishACSView:
             .exists()
         )
 
+    @pytest.mark.django_db(transaction=True)
+    def test_dispatch_serializes_concurrent_fallback_role_assignment(
+        self,
+        create_test_user,
+        tenants_fixture,
+        saml_setup,
+        monkeypatch,
+    ):
+        """Test concurrent callbacks assign only one fallback role"""
+        monkeypatch.setenv("SAML_SSO_CALLBACK_URL", "http://localhost/sso-complete")
+        user = create_test_user
+        tenant = tenants_fixture[0]
+
+        Role.objects.using(MainRouter.admin_db).create(
+            name="read_only",
+            tenant=tenant,
+            manage_users=True,
+            unlimited_visibility=True,
+        )
+
+        social_account = SocialAccount(
+            user=user,
+            provider="saml",
+            extra_data={
+                "firstName": ["John"],
+                "lastName": ["Doe"],
+                "organization": ["testing_company"],
+            },
+        )
+        # Without the user lock, both callbacks reach this query before either
+        # creates a fallback. With the lock, the first callback times out here
+        # while the second waits for the transaction to finish.
+        second_role_check_reached = Event()
+        concurrent_role_checks_detected = Event()
+        role_check_count_lock = Lock()
+        role_check_count = 0
+        original_role_check = TenantFinishACSView._user_has_tenant_role
+
+        def synchronize_role_checks(user_id, tenant_id):
+            nonlocal role_check_count
+            with role_check_count_lock:
+                role_check_count += 1
+                is_first_role_check = role_check_count == 1
+                if role_check_count == 2:
+                    second_role_check_reached.set()
+            if is_first_role_check and second_role_check_reached.wait(timeout=1):
+                concurrent_role_checks_detected.set()
+            return original_role_check(user_id, tenant_id)
+
+        def dispatch_callback():
+            close_old_connections()
+            try:
+                thread_user = User.objects.using(MainRouter.admin_db).get(pk=user.pk)
+                request = RequestFactory().get(
+                    reverse(
+                        "saml_finish_acs",
+                        kwargs={"organization_slug": saml_setup["domain"]},
+                    )
+                )
+                request.user = thread_user
+                request.session = {}
+                response = TenantFinishACSView.as_view()(
+                    request, organization_slug=saml_setup["domain"]
+                )
+                return response
+            finally:
+                close_old_connections()
+
+        with (
+            patch(
+                "allauth.socialaccount.providers.saml.views.get_app_or_404"
+            ) as mock_get_app_or_404,
+            patch(
+                "allauth.socialaccount.models.SocialApp.objects.get"
+            ) as mock_socialapp_get,
+            patch(
+                "allauth.socialaccount.models.SocialAccount.objects.get"
+            ) as mock_sa_get,
+            patch("api.models.SAMLDomainIndex.objects.get") as mock_saml_domain_get,
+            patch("api.models.SAMLConfiguration.objects.get") as mock_saml_config_get,
+            patch("api.models.User.objects.get") as mock_user_get,
+            patch.object(
+                TenantFinishACSView,
+                "_user_has_tenant_role",
+                side_effect=synchronize_role_checks,
+            ),
+        ):
+            mock_get_app_or_404.return_value = MagicMock(
+                provider="saml",
+                client_id=saml_setup["domain"],
+                name="Test App",
+                settings={},
+            )
+            mock_sa_get.return_value = social_account
+            mock_socialapp_get.return_value = MagicMock(provider_id="saml")
+            mock_saml_domain_get.return_value = SimpleNamespace(tenant_id=tenant.id)
+            mock_saml_config_get.return_value = SimpleNamespace(
+                email_domain=saml_setup["domain"], tenant=tenant
+            )
+            mock_user_get.side_effect = lambda *_args, **_kwargs: User.objects.using(
+                MainRouter.admin_db
+            ).get(pk=user.pk)
+
+            with ThreadPoolExecutor(max_workers=2) as executor:
+                responses = list(executor.map(lambda _: dispatch_callback(), range(2)))
+
+        assert role_check_count == 2
+        assert not concurrent_role_checks_detected.is_set()
+        for response in responses:
+            assert response.status_code == status.HTTP_302_FOUND
+            parsed_redirect = urlparse(response.url)
+            assert parsed_redirect.path == "/sso-complete"
+            assert set(parse_qs(parsed_redirect.query)) == {"id"}
+        relationships = UserRoleRelationship.objects.using(MainRouter.admin_db).filter(
+            user=user, tenant_id=tenant.id
+        )
+        assert relationships.count() == 1
+        assert relationships.get().role.name == "read_only_0"
+        assert (
+            Role.objects.using(MainRouter.admin_db)
+            .filter(tenant=tenant, name__startswith="read_only_")
+            .count()
+            == 1
+        )
+
     def test_dispatch_skips_role_mapping_when_last_manage_account_user_maps_to_new_role(
         self,
         create_test_user,

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

@@ -814,6 +814,21 @@ class TenantFinishACSView(FinishACSView):
             User.objects.using(MainRouter.admin_db).filter(id=saml_user_id).delete()
             request.session.pop("saml_user_created", None)
 
+    @staticmethod
+    def _user_has_tenant_role(user_id, tenant_id):
+        return (
+            UserRoleRelationship.objects.using(MainRouter.admin_db)
+            .filter(user_id=user_id, tenant_id=tenant_id)
+            .exists()
+        )
+
+    @staticmethod
+    def _is_read_only_fallback_role(role):
+        return (
+            not any(getattr(role, permission) for permission in Role.PERMISSION_FIELDS)
+            and role.unlimited_visibility
+        )
+
     def dispatch(self, request, organization_slug):
         try:
             super().dispatch(request, organization_slug)
@@ -879,11 +894,56 @@ class TenantFinishACSView(FinishACSView):
             user.name = "N/A"
         user.save()
 
-        # Only remap roles when the IdP provides a userType attribute.
-        # Without it, the user's current roles are left untouched.
+        # Only remap existing roles when the IdP provides a userType attribute.
+        # Without it, preserve current roles or assign a read-only fallback.
         role_name = (
             extra.get("userType", [""])[0].strip() if extra.get("userType") else ""
         )
+        if not role_name:
+            with rls_transaction(str(tenant.id), using=MainRouter.admin_db):
+                with transaction.atomic(using=MainRouter.admin_db):
+                    # Serialize concurrent ACS callbacks for the same user.
+                    (
+                        User.objects.using(MainRouter.admin_db)
+                        .select_for_update()
+                        .only("id")
+                        .get(pk=user_id)
+                    )
+                    user_has_roles = self._user_has_tenant_role(user_id, tenant.id)
+                    if not user_has_roles:
+                        read_only_defaults = dict.fromkeys(
+                            Role.PERMISSION_FIELDS, False
+                        )
+                        read_only_defaults["unlimited_visibility"] = True
+                        role, role_created = Role.objects.using(
+                            MainRouter.admin_db
+                        ).get_or_create(
+                            name="read_only",
+                            tenant=tenant,
+                            defaults=read_only_defaults,
+                        )
+                        role_is_read_only = self._is_read_only_fallback_role(role)
+                        if not role_created and not role_is_read_only:
+                            suffix = 0
+                            while not role_created and not role_is_read_only:
+                                role, role_created = Role.objects.using(
+                                    MainRouter.admin_db
+                                ).get_or_create(
+                                    name=f"read_only_{suffix}",
+                                    tenant=tenant,
+                                    defaults=read_only_defaults,
+                                )
+                                role_is_read_only = self._is_read_only_fallback_role(
+                                    role
+                                )
+                                suffix += 1
+                        UserRoleRelationship.objects.using(
+                            MainRouter.admin_db
+                        ).get_or_create(
+                            user=user,
+                            role=role,
+                            defaults={"tenant": tenant},
+                        )
         if role_name:
             with transaction.atomic(using=MainRouter.admin_db):
                 role = (

docs/user-guide/tutorials/prowler-app-sso-google-workspace.mdx

@@ -2,6 +2,7 @@
 title: 'SAML SSO: Google Workspace'
 ---
 
+import { VersionBadge } from "/snippets/version-badge.mdx"
 import { AppliesTo } from "/snippets/applies-to.mdx"
 
 <AppliesTo />
@@ -113,7 +114,12 @@ The `userType` attribute controls which Prowler role is assigned to the user:
 
 - If `userType` matches an existing Prowler role name, the user receives that role automatically.
 - If `userType` does not match any existing role, Prowler Cloud creates a new role with that name **with read-only access** (visibility over all providers, no management permissions). A Prowler administrator can adjust its permissions afterward through the [RBAC Management](/user-guide/tutorials/prowler-app-rbac) tab.
-- If `userType` is not set, the user's existing roles are left unchanged.
+
+**Fallback Role Without `userType`**
+
+<VersionBadge version="5.37.0" />
+
+If `userType` is not set, the user's existing roles are left unchanged. Users without an existing role in that tenant receive a least-privilege `read_only` fallback role until a Prowler administrator assigns another role. If `read_only` already belongs to a role with different permissions, Prowler Cloud checks suffixed names in order, starting with `read_only_0`. It reuses the first role with the fallback permissions or creates the first available name.
 
 The `userType` value is **case-sensitive** - for example, `Backend` and `backend` are treated as different roles.
 

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

@@ -91,9 +91,15 @@ Choose a Method:
         |----------------|---------------------------------------------------------------------------------------------------------|----------|
         | `firstName`    | The user's first name.                                                                                  | Yes      |
         | `lastName`     | The user's last name.                                                                                   | Yes      |
-        | `userType`     | Determines which Prowler role the user receives (e.g., `admin`, `auditor`). If a role with that name already exists, the user receives it automatically; if it does not exist, Prowler Cloud creates a new role with that name with read-only access (visibility over all providers, no management permissions). If `userType` is not defined, the user's existing roles are left unchanged. Role permissions can be edited in the [RBAC Management tab](/user-guide/tutorials/prowler-app-rbac). | No       |
+        | `userType`     | Determines which Prowler role the user receives (e.g., `admin`, `auditor`). If a role with that name already exists, the user receives it automatically; if it does not exist, Prowler Cloud creates a new role with that name with read-only access (visibility over all providers, no management permissions). A Prowler administrator can adjust its permissions through the [RBAC Management tab](/user-guide/tutorials/prowler-app-rbac). If `userType` is not defined, Prowler Cloud applies the fallback behavior described below. | No       |
         | `organization` | The user's company name.                                                                                | No       |
 
+        **Fallback Role Without `userType`**
+
+        <VersionBadge version="5.37.0" />
+
+        If `userType` is not defined, the user's existing roles are left unchanged. Users without an existing role in that tenant receive a least-privilege `read_only` fallback role. If `read_only` already belongs to a role with different permissions, Prowler Cloud checks suffixed names in order, starting with `read_only_0`. It reuses the first role with the fallback permissions or creates the first available name. A Prowler administrator can then assign the appropriate role through the [RBAC Management tab](/user-guide/tutorials/prowler-app-rbac).
+
         <Info>
         **IdP Attribute Mapping**
 
@@ -163,9 +169,14 @@ Choose a Method:
 
             * If a role with the specified name already exists in Prowler Cloud, the user automatically receives that role.
             * If the role does not exist, Prowler Cloud creates a new role with that exact name with read-only access: the user can see all providers and their findings but cannot manage anything. A Prowler administrator (a user whose role includes the "Manage Account" permission) can adjust its permissions afterward through the [RBAC Management tab](/user-guide/tutorials/prowler-app-rbac).
-            * If `userType` is not defined in the user's Okta profile, the user's existing roles in Prowler Cloud are left unchanged.
             * `userType` must contain a single value. If the IdP sends multiple values, Prowler Cloud uses only the first value and does not assign multiple roles.
 
+            **Fallback Role Without `userType`**
+
+            <VersionBadge version="5.37.0" />
+
+            If `userType` is not defined in the user's Okta profile, the user's existing roles in Prowler Cloud are left unchanged. Users without an existing role in that tenant receive a least-privilege `read_only` fallback role until a Prowler administrator assigns another role. If `read_only` already belongs to a role with different permissions, Prowler Cloud checks suffixed names in order, starting with `read_only_0`. It reuses the first role with the fallback permissions or creates the first available name.
+
             **Example:** To assign the `IT` role to a user, set the `userType` value to `IT` in Okta. If a role named `IT` already exists in Prowler Cloud, the user receives it automatically upon login. If it does not exist, Prowler Cloud creates a new role called `IT` with read-only access, and a Prowler administrator can adjust its permissions as needed.
 
             </Warning>