diff --git a/api/changelog.d/saml-missing-usertype-read-only.fixed.md b/api/changelog.d/saml-missing-usertype-read-only.fixed.md
new file mode 100644
index 0000000000..940ae579bc
--- /dev/null
+++ b/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
diff --git a/api/src/backend/api/tests/test_views.py b/api/src/backend/api/tests/test_views.py
index 3dd96d162b..32fd37a59e 100644
--- a/api/src/backend/api/tests/test_views.py
+++ b/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,
diff --git a/api/src/backend/api/v1/views.py b/api/src/backend/api/v1/views.py
index 075dd36829..f239512234 100644
--- a/api/src/backend/api/v1/views.py
+++ b/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 = (
diff --git a/docs/user-guide/tutorials/prowler-app-sso-google-workspace.mdx b/docs/user-guide/tutorials/prowler-app-sso-google-workspace.mdx
index ce0e5edc79..4c4169dff1 100644
--- a/docs/user-guide/tutorials/prowler-app-sso-google-workspace.mdx
+++ b/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"
@@ -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`**
+
+
+
+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.
diff --git a/docs/user-guide/tutorials/prowler-app-sso.mdx b/docs/user-guide/tutorials/prowler-app-sso.mdx
index 4eafc6846b..2083e15e29 100644
--- a/docs/user-guide/tutorials/prowler-app-sso.mdx
+++ b/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`**
+
+
+
+ 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).
+
**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`**
+
+
+
+ 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.