Merge branch 'master' into PROWLER-1391-provider-contract-dynamic-discovery

.pre-commit-config.yaml

@@ -138,12 +138,3 @@ repos:
         entry: bash -c 'vulture --exclude "contrib,.venv,api/src/backend/api/tests/,api/src/backend/conftest.py,api/src/backend/tasks/tests/,skills/" --min-confidence 100 .'
         language: system
         files: '.*\.py'
-
-      - id: ui-checks
-        name: UI - Husky Pre-commit
-        description: "Run UI pre-commit checks (Claude Code validation + healthcheck)"
-        entry: bash -c 'cd ui && .husky/pre-commit'
-        language: system
-        files: '^ui/.*\.(ts|tsx|js|jsx|json|css)$'
-        pass_filenames: false
-        verbose: true

api/CHANGELOG.md

@@ -2,6 +2,14 @@
 
 All notable changes to the **Prowler API** are documented in this file.
 
+## [1.26.0] (Prowler UNRELEASED)
+
+### 🔄 Changed
+
+- Allows tenant owners to expel users from their organizations  [(#10787)](https://github.com/prowler-cloud/prowler/pull/10787)
+
+---
+
 ## [1.25.3] (Prowler v5.24.3)
 
 ### 🐞 Fixed
@@ -20,7 +28,6 @@ All notable changes to the **Prowler API** are documented in this file.
 
 - `/finding-groups/latest/<check_id>/resources` now selects the latest completed scan per provider by `-completed_at` (then `-inserted_at`) instead of `-inserted_at`, matching the `/finding-groups/latest` summary path and the daily-summary upsert so overlapping scans no longer produce diverging `delta`/`new_count` between the two endpoints [(#10802)](https://github.com/prowler-cloud/prowler/pull/10802)
 
----
 
 ## [1.25.1] (Prowler v5.24.1)
 
@@ -34,6 +41,7 @@ All notable changes to the **Prowler API** are documented in this file.
 - Attack Paths: Missing `tenant_id` filter while getting related findings after scan completes [(#10722)](https://github.com/prowler-cloud/prowler/pull/10722)
 - Finding group counters `pass_count`, `fail_count` and `manual_count` now exclude muted findings [(#10753)](https://github.com/prowler-cloud/prowler/pull/10753)
 - Silent data loss in `ResourceFindingMapping` bulk insert that left findings orphaned when `INSERT ... ON CONFLICT DO NOTHING` dropped rows without raising; added explicit `unique_fields` [(#10724)](https://github.com/prowler-cloud/prowler/pull/10724)
+- `DELETE /tenants/{tenant_pk}/memberships/{id}` now deletes the expelled user's account when the removed membership was their last one, and blacklists every outstanding refresh token for that user so their existing sessions can no longer mint new access tokens [(#10787)](https://github.com/prowler-cloud/prowler/pull/10787)
 
 ---
 

api/poetry.lock

@@ -7147,14 +7147,14 @@ urllib3 = ">=1.26.0"
 
 [[package]]
 name = "pygments"
-version = "2.19.2"
+version = "2.20.0"
 description = "Pygments is a syntax highlighting package written in Python."
 optional = false
-python-versions = ">=3.8"
+python-versions = ">=3.9"
 groups = ["main", "dev"]
 files = [
-    {file = "pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b"},
-    {file = "pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887"},
+    {file = "pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176"},
+    {file = "pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f"},
 ]
 
 [package.extras]

api/src/backend/api/filters.py

@@ -330,6 +330,7 @@ class MembershipFilter(FilterSet):
         model = Membership
         fields = {
             "tenant": ["exact"],
+            "user": ["exact"],
             "role": ["exact"],
             "date_joined": ["date", "gte", "lte"],
         }

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

@@ -32,6 +32,11 @@ from django_celery_results.models import TaskResult
 from rest_framework import status
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.response import Response
+from rest_framework_simplejwt.token_blacklist.models import (
+    BlacklistedToken,
+    OutstandingToken,
+)
+from rest_framework_simplejwt.tokens import RefreshToken
 
 from api.attack_paths import (
     AttackPathsQueryDefinition,
@@ -47,6 +52,7 @@ from api.models import (
     Finding,
     Integration,
     Invitation,
+    InvitationRoleRelationship,
     LighthouseProviderConfiguration,
     LighthouseProviderModels,
     LighthouseTenantConfiguration,
@@ -746,6 +752,39 @@ class TestTenantViewSet:
         # Test user + 2 extra users for tenant 2
         assert len(response.json()["data"]) == 3
 
+    def test_tenants_list_memberships_filter_by_user(
+        self, authenticated_client, tenants_fixture, extra_users
+    ):
+        _, tenant2, _ = tenants_fixture
+        _, user3_membership = extra_users
+        user3, membership3 = user3_membership
+
+        response = authenticated_client.get(
+            reverse("tenant-membership-list", kwargs={"tenant_pk": tenant2.id}),
+            {"filter[user]": str(user3.id)},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 1
+        assert data[0]["id"] == str(membership3.id)
+
+    def test_tenants_list_memberships_filter_by_user_no_match(
+        self, authenticated_client, tenants_fixture, extra_users
+    ):
+        _, tenant2, _ = tenants_fixture
+        unrelated_user = User.objects.create_user(
+            name="unrelated",
+            password=TEST_PASSWORD,
+            email="unrelated@gmail.com",
+        )
+
+        response = authenticated_client.get(
+            reverse("tenant-membership-list", kwargs={"tenant_pk": tenant2.id}),
+            {"filter[user]": str(unrelated_user.id)},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["data"] == []
+
     def test_tenants_list_memberships_as_member(
         self, authenticated_client, tenants_fixture, extra_users
     ):
@@ -803,6 +842,7 @@ class TestTenantViewSet:
     ):
         _, tenant2, _ = tenants_fixture
         user_membership = Membership.objects.get(tenant=tenant2, user__email=TEST_USER)
+        user_id = user_membership.user_id
         response = authenticated_client.delete(
             reverse(
                 "tenant-membership-detail",
@@ -811,6 +851,127 @@ class TestTenantViewSet:
         )
         assert response.status_code == status.HTTP_403_FORBIDDEN
         assert Membership.objects.filter(id=user_membership.id).exists()
+        assert User.objects.filter(id=user_id).exists()
+
+    def test_expel_user_deletes_account_if_last_membership(
+        self, authenticated_client, tenants_fixture, extra_users
+    ):
+        # TEST_USER is OWNER of tenant2; user3 is MEMBER only in tenant2
+        _, tenant2, _ = tenants_fixture
+        _, user3_membership = extra_users
+        user3, membership3 = user3_membership
+
+        assert Membership.objects.filter(user=user3).count() == 1
+
+        response = authenticated_client.delete(
+            reverse(
+                "tenant-membership-detail",
+                kwargs={"tenant_pk": tenant2.id, "pk": membership3.id},
+            )
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert not Membership.objects.filter(id=membership3.id).exists()
+        assert not User.objects.filter(id=user3.id).exists()
+
+    def test_expel_user_blacklists_refresh_tokens(
+        self, authenticated_client, tenants_fixture, extra_users
+    ):
+        _, tenant2, _ = tenants_fixture
+        _, user3_membership = extra_users
+        user3, membership3 = user3_membership
+
+        # Issue two refresh tokens to simulate active sessions
+        RefreshToken.for_user(user3)
+        RefreshToken.for_user(user3)
+        outstanding_ids = list(
+            OutstandingToken.objects.filter(user=user3).values_list("id", flat=True)
+        )
+        assert len(outstanding_ids) == 2
+        assert not BlacklistedToken.objects.filter(
+            token_id__in=outstanding_ids
+        ).exists()
+
+        response = authenticated_client.delete(
+            reverse(
+                "tenant-membership-detail",
+                kwargs={"tenant_pk": tenant2.id, "pk": membership3.id},
+            )
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert (
+            BlacklistedToken.objects.filter(token_id__in=outstanding_ids).count() == 2
+        )
+
+    def test_expel_user_blacklists_refresh_tokens_is_idempotent(
+        self, authenticated_client, tenants_fixture, extra_users
+    ):
+        # Regression test for the bulk blacklisting path: if one of the
+        # user's refresh tokens is already blacklisted when the expel
+        # endpoint runs, the remaining tokens must still be blacklisted
+        # and the already-blacklisted one must not be duplicated.
+        tenant1, tenant2, _ = tenants_fixture
+        _, user3_membership = extra_users
+        user3, membership3 = user3_membership
+
+        # Keep the user alive after the expel so the assertions below can
+        # still query OutstandingToken by user_id.
+        Membership.objects.create(
+            user=user3,
+            tenant=tenant1,
+            role=Membership.RoleChoices.MEMBER,
+        )
+
+        RefreshToken.for_user(user3)
+        RefreshToken.for_user(user3)
+        outstanding_ids = list(
+            OutstandingToken.objects.filter(user=user3).values_list("id", flat=True)
+        )
+        assert len(outstanding_ids) == 2
+
+        # Pre-blacklist one of the two tokens to simulate a prior revocation.
+        BlacklistedToken.objects.create(token_id=outstanding_ids[0])
+        assert (
+            BlacklistedToken.objects.filter(token_id__in=outstanding_ids).count() == 1
+        )
+
+        response = authenticated_client.delete(
+            reverse(
+                "tenant-membership-detail",
+                kwargs={"tenant_pk": tenant2.id, "pk": membership3.id},
+            )
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        blacklisted = BlacklistedToken.objects.filter(token_id__in=outstanding_ids)
+        assert blacklisted.count() == 2
+        assert set(blacklisted.values_list("token_id", flat=True)) == set(
+            outstanding_ids
+        )
+
+    def test_expel_user_keeps_account_if_has_other_memberships(
+        self, authenticated_client, tenants_fixture, extra_users
+    ):
+        tenant1, tenant2, _ = tenants_fixture
+        _, user3_membership = extra_users
+        user3, membership3 = user3_membership
+
+        # Give user3 an additional membership in tenant1 so they are not orphaned
+        other_membership = Membership.objects.create(
+            user=user3,
+            tenant=tenant1,
+            role=Membership.RoleChoices.MEMBER,
+        )
+
+        response = authenticated_client.delete(
+            reverse(
+                "tenant-membership-detail",
+                kwargs={"tenant_pk": tenant2.id, "pk": membership3.id},
+            )
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert not Membership.objects.filter(id=membership3.id).exists()
+        assert User.objects.filter(id=user3.id).exists()
+        assert Membership.objects.filter(id=other_membership.id).exists()
 
     def test_tenants_delete_another_membership_as_owner(
         self, authenticated_client, tenants_fixture, extra_users
@@ -882,6 +1043,128 @@ class TestTenantViewSet:
         assert response.status_code == status.HTTP_404_NOT_FOUND
         assert Membership.objects.filter(id=other_membership.id).exists()
 
+    def test_delete_membership_cleans_up_orphaned_role_grants(
+        self, authenticated_client, tenants_fixture
+    ):
+        """Test that deleting a membership removes UserRoleRelationship records
+        for that tenant while preserving grants in other tenants."""
+        tenant1, tenant2, _ = tenants_fixture
+
+        # Create a user with memberships in both tenants
+        user = User.objects.create_user(
+            name="Multi-tenant User",
+            password=TEST_PASSWORD,
+            email="multitenant@test.com",
+        )
+
+        # Create memberships in both tenants
+        Membership.objects.create(
+            user=user, tenant=tenant1, role=Membership.RoleChoices.MEMBER
+        )
+        membership2 = Membership.objects.create(
+            user=user, tenant=tenant2, role=Membership.RoleChoices.MEMBER
+        )
+
+        # Create roles in both tenants
+        role1 = Role.objects.create(
+            name="Test Role 1", tenant=tenant1, manage_providers=True
+        )
+        role2 = Role.objects.create(
+            name="Test Role 2", tenant=tenant2, manage_scans=True
+        )
+
+        # Create user role relationships for both tenants
+        UserRoleRelationship.objects.create(user=user, role=role1, tenant=tenant1)
+        UserRoleRelationship.objects.create(user=user, role=role2, tenant=tenant2)
+
+        # Verify initial state
+        assert UserRoleRelationship.objects.filter(user=user, tenant=tenant1).exists()
+        assert UserRoleRelationship.objects.filter(user=user, tenant=tenant2).exists()
+        assert Role.objects.filter(id=role1.id).exists()
+        assert Role.objects.filter(id=role2.id).exists()
+
+        # Delete membership from tenant2 (authenticated user is owner of tenant2)
+        response = authenticated_client.delete(
+            reverse(
+                "tenant-membership-detail",
+                kwargs={"tenant_pk": tenant2.id, "pk": membership2.id},
+            )
+        )
+
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        # Verify the membership was deleted
+        assert not Membership.objects.filter(id=membership2.id).exists()
+
+        # Verify UserRoleRelationship for tenant2 was deleted
+        assert not UserRoleRelationship.objects.filter(
+            user=user, tenant=tenant2
+        ).exists()
+
+        # Verify UserRoleRelationship for tenant1 is preserved
+        assert UserRoleRelationship.objects.filter(user=user, tenant=tenant1).exists()
+
+        # Verify orphaned role2 was deleted (no more user or invitation relationships)
+        assert not Role.objects.filter(id=role2.id).exists()
+
+        # Verify role1 is preserved (still has user relationship)
+        assert Role.objects.filter(id=role1.id).exists()
+
+        # Verify the user still exists (has other memberships)
+        assert User.objects.filter(id=user.id).exists()
+
+    def test_delete_membership_preserves_role_with_invitation_relationship(
+        self, authenticated_client, tenants_fixture
+    ):
+        """Test that roles are not deleted if they have invitation relationships."""
+        _, tenant2, _ = tenants_fixture
+
+        # Create a user with membership
+        user = User.objects.create_user(
+            name="Test User", password=TEST_PASSWORD, email="testuser@test.com"
+        )
+        membership = Membership.objects.create(
+            user=user, tenant=tenant2, role=Membership.RoleChoices.MEMBER
+        )
+
+        # Create a role and user relationship
+        role = Role.objects.create(
+            name="Shared Role", tenant=tenant2, manage_providers=True
+        )
+        UserRoleRelationship.objects.create(user=user, role=role, tenant=tenant2)
+
+        # Create an invitation with the same role
+        invitation = Invitation.objects.create(email="pending@test.com", tenant=tenant2)
+        InvitationRoleRelationship.objects.create(
+            invitation=invitation, role=role, tenant=tenant2
+        )
+
+        # Verify initial state
+        assert UserRoleRelationship.objects.filter(user=user, role=role).exists()
+        assert InvitationRoleRelationship.objects.filter(
+            invitation=invitation, role=role
+        ).exists()
+        assert Role.objects.filter(id=role.id).exists()
+
+        # Delete the membership
+        response = authenticated_client.delete(
+            reverse(
+                "tenant-membership-detail",
+                kwargs={"tenant_pk": tenant2.id, "pk": membership.id},
+            )
+        )
+
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        # Verify UserRoleRelationship was deleted
+        assert not UserRoleRelationship.objects.filter(user=user, role=role).exists()
+
+        # Verify role is preserved because invitation relationship exists
+        assert Role.objects.filter(id=role.id).exists()
+        assert InvitationRoleRelationship.objects.filter(
+            invitation=invitation, role=role
+        ).exists()
+
     def test_tenants_list_no_permissions(
         self, authenticated_client_no_permissions_rbac, tenants_fixture
     ):

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

@@ -83,6 +83,10 @@ from rest_framework.permissions import SAFE_METHODS
 from rest_framework_json_api import filters as jsonapi_filters
 from rest_framework_json_api.views import RelationshipView, Response
 from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
+from rest_framework_simplejwt.token_blacklist.models import (
+    BlacklistedToken,
+    OutstandingToken,
+)
 from tasks.beat import schedule_provider_scan
 from tasks.jobs.attack_paths import db_utils as attack_paths_db_utils
 from tasks.jobs.export import get_s3_client
@@ -169,6 +173,7 @@ from api.models import (
     FindingGroupDailySummary,
     Integration,
     Invitation,
+    InvitationRoleRelationship,
     LighthouseConfiguration,
     LighthouseProviderConfiguration,
     LighthouseProviderModels,
@@ -1330,9 +1335,11 @@ class MembershipViewSet(BaseTenantViewset):
     ),
     destroy=extend_schema(
         summary="Delete tenant memberships",
-        description="Delete the membership details of users in a tenant. You need to be one of the owners to delete a "
-        "membership that is not yours. If you are the last owner of a tenant, you cannot delete your own "
-        "membership.",
+        description="Delete a user's membership from a tenant. This action: (1) removes the membership, "
+        "(2) revokes all refresh tokens for the expelled user, (3) removes their role grants for this tenant, "
+        "(4) cleans up orphaned roles, and (5) deletes the user account if this was their last membership. "
+        "You must be a tenant owner to delete another user's membership. The last owner of a tenant cannot "
+        "delete their own membership.",
         tags=["Tenant"],
     ),
 )
@@ -1341,6 +1348,7 @@ class TenantMembersViewSet(BaseTenantViewset):
     http_method_names = ["get", "delete"]
     serializer_class = MembershipSerializer
     queryset = Membership.objects.none()
+    filterset_class = MembershipFilter
     # Authorization is handled by get_requesting_membership (owner/member checks),
     # not by RBAC, since the target tenant differs from the JWT tenant.
     required_permissions = []
@@ -1398,7 +1406,84 @@ class TenantMembersViewSet(BaseTenantViewset):
                     "You do not have permission to delete this membership."
                 )
 
-        membership_to_delete.delete()
+        user_to_check_id = membership_to_delete.user_id
+        tenant_id = membership_to_delete.tenant_id
+        # All writes run on the admin connection so that the uncommitted
+        # membership delete is visible to the subsequent "other memberships"
+        # check. Splitting the delete and the check across the default
+        # (prowler_user, RLS) and admin connections caused the admin side to
+        # miss the just-deleted row and leave the User row orphaned.
+        with transaction.atomic(using=MainRouter.admin_db):
+            Membership.objects.using(MainRouter.admin_db).filter(
+                id=membership_to_delete.id
+            ).delete()
+
+            # Remove role grants for this user in this tenant to prevent
+            # orphaned permissions that could allow access after expulsion
+            deleted_role_relationships = UserRoleRelationship.objects.using(
+                MainRouter.admin_db
+            ).filter(user_id=user_to_check_id, tenant_id=tenant_id)
+
+            # Collect role IDs that might become orphaned after deletion
+            role_ids_to_check = list(
+                deleted_role_relationships.values_list("role_id", flat=True)
+            )
+
+            # Delete the user role relationships for this tenant
+            deleted_role_relationships.delete()
+
+            # Clean up orphaned roles that have no remaining user or invitation relationships
+            if role_ids_to_check:
+                for role_id in role_ids_to_check:
+                    has_user_relationships = (
+                        UserRoleRelationship.objects.using(MainRouter.admin_db)
+                        .filter(role_id=role_id)
+                        .exists()
+                    )
+
+                    has_invitation_relationships = (
+                        InvitationRoleRelationship.objects.using(MainRouter.admin_db)
+                        .filter(role_id=role_id)
+                        .exists()
+                    )
+
+                    if not has_user_relationships and not has_invitation_relationships:
+                        Role.objects.using(MainRouter.admin_db).filter(
+                            id=role_id
+                        ).delete()
+
+            # Revoke any refresh tokens the expelled user still holds so they
+            # cannot mint fresh access tokens. This must happen before the
+            # User row is deleted, because OutstandingToken.user is
+            # on_delete=SET_NULL in djangorestframework-simplejwt 5.5.1
+            # (see rest_framework_simplejwt/token_blacklist/models.py): once
+            # the user row is gone, user_id becomes NULL and we can no longer
+            # look up that user's outstanding tokens. Access tokens already
+            # issued remain valid until SIMPLE_JWT["ACCESS_TOKEN_LIFETIME"]
+            # expires.
+            outstanding_token_ids = list(
+                OutstandingToken.objects.using(MainRouter.admin_db)
+                .filter(user_id=user_to_check_id)
+                .values_list("id", flat=True)
+            )
+            if outstanding_token_ids:
+                BlacklistedToken.objects.using(MainRouter.admin_db).bulk_create(
+                    [
+                        BlacklistedToken(token_id=token_id)
+                        for token_id in outstanding_token_ids
+                    ],
+                    ignore_conflicts=True,
+                )
+
+            has_other_memberships = (
+                Membership.objects.using(MainRouter.admin_db)
+                .filter(user_id=user_to_check_id)
+                .exists()
+            )
+            if not has_other_memberships:
+                User.objects.using(MainRouter.admin_db).filter(
+                    id=user_to_check_id
+                ).delete()
         return Response(status=status.HTTP_204_NO_CONTENT)
 
 

docs/developer-guide/outputs.mdx

@@ -8,7 +8,7 @@ Prowler supports multiple output formats, allowing users to tailor findings pres
 
 - Output Organization in Prowler
 
-    Prowler outputs are managed within the `/lib/outputs` directory. Each format—such as JSON, CSV, HTML—is implemented as a Python class.
+    Prowler outputs are managed within the `/lib/outputs` directory. Each format—such as JSON, CSV, HTML, SARIF—is implemented as a Python class.
 
 - Outputs are generated based on scan findings, which are stored as structured dictionaries containing details such as:
 

docs/getting-started/basic-usage/prowler-cli.mdx

@@ -25,7 +25,12 @@ If you prefer the former verbose output, use: `--verbose`. This allows seeing mo
 
 ## Report Generation
 
-By default, Prowler generates CSV, JSON-OCSF, and HTML reports. To generate a JSON-ASFF report (used by AWS Security Hub), specify `-M` or `--output-modes`:
+By default, Prowler generates CSV, JSON-OCSF, and HTML reports. Additional provider-specific formats are available:
+
+* **JSON-ASFF** (AWS only): Used by AWS Security Hub
+* **SARIF** (IaC only): Used by GitHub Code Scanning
+
+To specify output formats, use the `-M` or `--output-modes` flag:
 
 ```console
 prowler <provider> -M csv json-asff json-ocsf html

docs/user-guide/cli/tutorials/reporting.mdx

@@ -61,6 +61,7 @@ Prowler natively supports the following reporting output formats:
 - JSON-OCSF
 - JSON-ASFF (AWS only)
 - HTML
+- SARIF (IaC only)
 
 Hereunder is the structure for each of the supported report formats by Prowler:
 
@@ -368,6 +369,29 @@ Each finding is a `json` object within a list.
 The following image is an example of the HTML output:
 
 <img src="/images/cli/reporting/html-output.png" />
+
+### SARIF (IaC Only)
+
+import { VersionBadge } from "/snippets/version-badge.mdx"
+
+<VersionBadge version="5.25.0" />
+
+The SARIF (Static Analysis Results Interchange Format) output generates a [SARIF 2.1.0](https://docs.oasis-open.org/sarif/sarif/v2.1.0/sarif-v2.1.0.html) document compatible with GitHub Code Scanning and other SARIF-compatible tools. This format is exclusively available for the IaC provider, as it is designed for static analysis results that reference specific files and line numbers.
+
+```console
+prowler iac --scan-repository-url https://github.com/user/repo -M sarif
+```
+
+<Note>
+The SARIF output format is only available when using the `iac` provider. Attempting to use it with other providers results in an error.
+</Note>
+
+The SARIF output includes:
+
+* **Rules:** Each unique check ID produces a rule entry with severity, description, remediation, and a markdown help panel.
+* **Results:** Only failed (non-muted) findings are included, with file paths and line numbers for precise annotation.
+* **Severity mapping:** Prowler severities map to SARIF levels (`critical`/`high` → `error`, `medium` → `warning`, `low`/`informational` → `note`).
+
 ## V4 Deprecations
 
 Some deprecations have been made to unify formats and improve outputs.

docs/user-guide/providers/iac/getting-started-iac.mdx

@@ -29,7 +29,7 @@ Prowler IaC provider scans the following Infrastructure as Code configurations f
 - For remote repository scans, authentication can be provided via [git URL](https://git-scm.com/docs/git-clone#_git_urls), CLI flags or environment variables.
   - Check the [IaC Authentication](/user-guide/providers/iac/authentication) page for more details.
 - Mutelist logic ([filtering](https://trivy.dev/latest/docs/configuration/filtering/)) is handled by Trivy, not Prowler.
-- Results are output in the same formats as other Prowler providers (CSV, JSON, HTML, etc.).
+- Results are output in the same formats as other Prowler providers (CSV, JSON-OCSF, HTML), plus [SARIF](/user-guide/cli/tutorials/reporting#sarif-iac-only) for GitHub Code Scanning integration.
 
 ## Prowler Cloud
 
@@ -140,8 +140,20 @@ prowler iac --scan-path ./my-iac-directory --exclude-path ./my-iac-directory/tes
 
 ### Output
 
-Use the standard Prowler output options, for example:
+Use the standard Prowler output options. The IaC provider also supports [SARIF](/user-guide/cli/tutorials/reporting#sarif-iac-only) output for GitHub Code Scanning integration:
 
 ```sh
-prowler iac --scan-path ./iac --output-formats csv json html
+prowler iac --scan-path ./iac --output-formats csv json-ocsf html
 ```
+
+#### SARIF Output
+
+<VersionBadge version="5.25.0" />
+
+To generate SARIF output for integration with SARIF-compatible tools:
+
+```sh
+prowler iac --scan-repository-url https://github.com/user/repo -M sarif
+```
+
+See the [SARIF reporting documentation](/user-guide/cli/tutorials/reporting#sarif-iac-only) for details on the format and severity mapping.

docs/user-guide/tutorials/prowler-app-multi-tenant.mdx

@@ -140,6 +140,34 @@ Invitations expire after 7 days. If an invitation has expired, contact the organ
 
 </Note>
 
+## Expelling a User From an Organization
+
+Organization owners can expel a member from the organization. Expelling removes the membership immediately, revoking access to all providers, scans, and findings scoped to that organization. Owners expelling themselves are blocked if they are the last remaining owner of the organization.
+
+To expel a user:
+
+1. Navigate to the **Users** page.
+
+2. Locate the user to remove and open the row actions menu.
+
+3. Select **Expel user**.
+
+    <img src="/images/prowler-app/multi-tenant/expel-user-organization.png" alt="Users table row action menu showing the 'Expel user' destructive option" width="700" />
+
+
+4. Confirm the action in the dialog. The membership is removed immediately and the expelled user loses access to the organization.
+
+    <img src="/images/prowler-app/multi-tenant/expel-user-organization-modal.png" alt="Confirmation dialog asking to expel the selected user from the current organization" width="700" />
+
+
+<Warning>
+Expelling a user revokes any refresh tokens the account holds, but access tokens already issued remain valid until they expire. The default access token lifetime is 30 minutes, so an expelled user may retain access to the organization for up to that window before being fully locked out.
+</Warning>
+
+<Warning>
+If the expelled organization was the user's **only** organization, the account is permanently deleted along with the membership. All personal profile data associated with that account is removed and cannot be recovered. To preserve the account, confirm that the user belongs to another organization before expelling.
+</Warning>
+
 ## Permissions Reference
 
 | Action | Required Conditions |
@@ -149,3 +177,4 @@ Invitations expire after 7 days. If an invitation has expired, contact the organ
 | Switch organizations | Any authenticated user |
 | Edit organization name | Organization owner with **Manage Account** permission |
 | Delete an organization | Organization owner with **Manage Account** permission; must belong to more than one organization |
+| Expel a user from an organization | Organization owner (no additional permission required); last remaining owner cannot expel themselves |

mcp_server/uv.lock

@@ -879,11 +879,11 @@ wheels = [
 
 [[package]]
 name = "pygments"
-version = "2.19.2"
+version = "2.20.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/b0/77/a5b8c569bf593b0140bde72ea885a803b82086995367bf2037de0159d924/pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887", size = 4968631, upload-time = "2025-06-21T13:39:12.283Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" },
+    { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" },
 ]
 
 [[package]]

prowler/CHANGELOG.md

@@ -7,6 +7,10 @@ All notable changes to the **Prowler SDK** are documented in this file.
 ### 🚀 Added
 
 - Support for external/custom providers, checks, and compliance frameworks without modifying core code [(#10700)](https://github.com/prowler-cloud/prowler/pull/10700)
+- SARIF output format for the IaC provider, enabling GitHub Code Scanning integration via `--output-formats sarif` [(#10626)](https://github.com/prowler-cloud/prowler/pull/10626)
+
+---
+
 ## [5.24.3] (Prowler v5.24.3)
 
 ### 🐞 Fixed

prowler/__main__.py

@@ -18,6 +18,7 @@ from prowler.config.config import (
     json_asff_file_suffix,
     json_ocsf_file_suffix,
     orange_color,
+    sarif_file_suffix,
 )
 from prowler.lib.banner import print_banner
 from prowler.lib.check.check import (
@@ -122,6 +123,7 @@ from prowler.lib.outputs.html.html import HTML
 from prowler.lib.outputs.ocsf.ingestion import send_ocsf_to_api
 from prowler.lib.outputs.ocsf.ocsf import OCSF
 from prowler.lib.outputs.outputs import extract_findings_statistics, report
+from prowler.lib.outputs.sarif.sarif import SARIF
 from prowler.lib.outputs.slack.slack import Slack
 from prowler.lib.outputs.summary_table import display_summary_table
 from prowler.providers.alibabacloud.models import AlibabaCloudOutputOptions
@@ -555,6 +557,13 @@ def prowler():
                 html_output.batch_write_data_to_file(
                     provider=global_provider, stats=stats
                 )
+            if mode == "sarif":
+                sarif_output = SARIF(
+                    findings=finding_outputs,
+                    file_path=f"{filename}{sarif_file_suffix}",
+                )
+                generated_outputs["regular"].append(sarif_output)
+                sarif_output.batch_write_data_to_file()
 
     if getattr(args, "push_to_cloud", False):
         if not ocsf_output or not getattr(ocsf_output, "file_path", None):

prowler/config/config.py

@@ -147,6 +147,7 @@ json_file_suffix = ".json"
 json_asff_file_suffix = ".asff.json"
 json_ocsf_file_suffix = ".ocsf.json"
 html_file_suffix = ".html"
+sarif_file_suffix = ".sarif"
 default_config_file_path = (
     f"{pathlib.Path(os.path.dirname(os.path.realpath(__file__)))}/config.yaml"
 )
@@ -157,7 +158,7 @@ default_redteam_config_file_path = (
     f"{pathlib.Path(os.path.dirname(os.path.realpath(__file__)))}/llm_config.yaml"
 )
 encoding_format_utf_8 = "utf-8"
-available_output_formats = ["csv", "json-asff", "json-ocsf", "html"]
+available_output_formats = ["csv", "json-asff", "json-ocsf", "html", "sarif"]
 
 # Prowler Cloud API settings
 cloud_api_base_url = os.getenv("PROWLER_CLOUD_API_BASE_URL", "https://api.prowler.com")

prowler/lib/check/models.py

@@ -1095,15 +1095,10 @@ class CheckReportIAC(Check_Report):
 
         self.resource = finding
         self.resource_name = file_path
-        self.resource_line_range = (
-            (
-                str(finding.get("CauseMetadata", {}).get("StartLine", ""))
-                + ":"
-                + str(finding.get("CauseMetadata", {}).get("EndLine", ""))
-            )
-            if finding.get("CauseMetadata", {}).get("StartLine", "")
-            else ""
-        )
+        cause = finding.get("CauseMetadata", {})
+        start = cause.get("StartLine") or finding.get("StartLine")
+        end = cause.get("EndLine") or finding.get("EndLine")
+        self.resource_line_range = f"{start}:{end}" if start else ""
 
 
 @dataclass

prowler/lib/cli/parser.py

@@ -18,6 +18,7 @@ from prowler.providers.common.arguments import (
     init_providers_parser,
     validate_asff_usage,
     validate_provider_arguments,
+    validate_sarif_usage,
 )
 from prowler.providers.common.provider import Provider
 
@@ -192,6 +193,12 @@ Detailed documentation at https://docs.prowler.com
         if not asff_is_valid:
             self.parser.error(asff_error)
 
+        sarif_is_valid, sarif_error = validate_sarif_usage(
+            args.provider, getattr(args, "output_formats", None)
+        )
+        if not sarif_is_valid:
+            self.parser.error(sarif_error)
+
         return args
 
     def __set_default_provider__(self, args: list) -> list:

prowler/lib/outputs/compliance/universal/ocsf_compliance.py

@@ -0,0 +1,395 @@
+import os
+from datetime import datetime
+from typing import List
+
+from py_ocsf_models.events.base_event import SeverityID
+from py_ocsf_models.events.base_event import StatusID as EventStatusID
+from py_ocsf_models.events.findings.compliance_finding import ComplianceFinding
+from py_ocsf_models.events.findings.compliance_finding_type_id import (
+    ComplianceFindingTypeID,
+)
+from py_ocsf_models.events.findings.finding import ActivityID, FindingInformation
+from py_ocsf_models.objects.check import Check
+from py_ocsf_models.objects.compliance import Compliance
+from py_ocsf_models.objects.compliance_status import StatusID as ComplianceStatusID
+from py_ocsf_models.objects.group import Group
+from py_ocsf_models.objects.metadata import Metadata
+from py_ocsf_models.objects.product import Product
+from py_ocsf_models.objects.resource_details import ResourceDetails
+
+from prowler.config.config import prowler_version
+from prowler.lib.check.compliance_models import ComplianceFramework
+from prowler.lib.logger import logger
+from prowler.lib.outputs.finding import Finding
+from prowler.lib.outputs.ocsf.ocsf import OCSF
+from prowler.lib.outputs.utils import unroll_dict_to_list
+from prowler.lib.utils.utils import open_file
+
+PROWLER_TO_COMPLIANCE_STATUS = {
+    "PASS": ComplianceStatusID.Pass,
+    "FAIL": ComplianceStatusID.Fail,
+    "MANUAL": ComplianceStatusID.Unknown,
+}
+
+
+def _to_snake_case(name: str) -> str:
+    """Convert a PascalCase or camelCase string to snake_case."""
+    import re
+
+    # Insert underscore before uppercase letters preceded by lowercase
+    s = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", name)
+    # Insert underscore between consecutive uppercase and following lowercase
+    s = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", s)
+    return s.lower()
+
+
+def _build_requirement_attrs(requirement, framework) -> dict:
+    """Build a dict with requirement attributes for the unmapped section.
+
+    Keys are normalized to snake_case for OCSF consistency.
+    Only includes attributes whose AttributeMetadata has output_formats.ocsf=True.
+    When no metadata is declared, all attributes are included.
+    """
+    attrs = requirement.attributes
+    if not attrs:
+        return {}
+
+    # Build set of keys allowed for OCSF output
+    metadata = framework.attributes_metadata
+    if metadata:
+        ocsf_keys = {m.key for m in metadata if m.output_formats.ocsf}
+    else:
+        ocsf_keys = None  # No metadata → include all
+
+    result = {}
+    for key, value in attrs.items():
+        if ocsf_keys is not None and key not in ocsf_keys:
+            continue
+        result[_to_snake_case(key)] = value
+    return result
+
+
+class OCSFComplianceOutput:
+    """Produces OCSF ComplianceFinding (class_uid 2003) events from
+    universal compliance framework data.
+
+    Each finding × requirement combination produces one ComplianceFinding event
+    with structured Compliance and Check objects.
+    """
+
+    def __init__(
+        self,
+        findings: list,
+        framework: ComplianceFramework,
+        file_path: str = None,
+        from_cli: bool = True,
+        provider: str = None,
+    ) -> None:
+        self._data = []
+        self._file_descriptor = None
+        self.file_path = file_path
+        self._from_cli = from_cli
+        self._provider = provider
+        self.close_file = False
+
+        if findings:
+            compliance_name = (
+                framework.framework + "-" + framework.version
+                if framework.version
+                else framework.framework
+            )
+            self._transform(findings, framework, compliance_name)
+            if not self._file_descriptor and file_path:
+                self._create_file_descriptor(file_path)
+
+    @property
+    def data(self):
+        return self._data
+
+    def _transform(
+        self,
+        findings: List[Finding],
+        framework: ComplianceFramework,
+        compliance_name: str,
+    ) -> None:
+        # Build check -> requirements map
+        check_req_map = {}
+        for req in framework.requirements:
+            checks = req.checks
+            if self._provider:
+                all_checks = checks.get(self._provider.lower(), [])
+            else:
+                all_checks = []
+                for check_list in checks.values():
+                    all_checks.extend(check_list)
+            for check_id in all_checks:
+                check_req_map.setdefault(check_id, []).append(req)
+
+        for finding in findings:
+            if finding.check_id in check_req_map:
+                for req in check_req_map[finding.check_id]:
+                    cf = self._build_compliance_finding(
+                        finding, framework, req, compliance_name
+                    )
+                    if cf:
+                        self._data.append(cf)
+
+        # Manual requirements (no checks or empty for current provider)
+        for req in framework.requirements:
+            checks = req.checks
+            if self._provider:
+                has_checks = bool(checks.get(self._provider.lower(), []))
+            else:
+                has_checks = any(checks.values())
+
+            if not has_checks:
+                cf = self._build_manual_compliance_finding(
+                    framework, req, compliance_name
+                )
+                if cf:
+                    self._data.append(cf)
+
+    def _build_unmapped(self, finding, requirement, framework) -> dict:
+        """Build the unmapped dict with cloud info and requirement attributes."""
+        unmapped = {}
+
+        # Cloud info (from finding, when available)
+        if finding and getattr(finding, "provider", None) != "kubernetes":
+            unmapped["cloud"] = {
+                "provider": finding.provider,
+                "region": finding.region,
+                "account": {
+                    "uid": finding.account_uid,
+                    "name": finding.account_name,
+                },
+                "org": {
+                    "uid": finding.account_organization_uid,
+                    "name": finding.account_organization_name,
+                },
+            }
+
+        # Requirement attributes
+        req_attrs = _build_requirement_attrs(requirement, framework)
+        if req_attrs:
+            unmapped["requirement_attributes"] = req_attrs
+
+        return unmapped or None
+
+    def _build_compliance_finding(
+        self,
+        finding: Finding,
+        framework: ComplianceFramework,
+        requirement,
+        compliance_name: str,
+    ) -> ComplianceFinding:
+        try:
+            compliance_status = PROWLER_TO_COMPLIANCE_STATUS.get(
+                finding.status, ComplianceStatusID.Unknown
+            )
+            check_status = PROWLER_TO_COMPLIANCE_STATUS.get(
+                finding.status, ComplianceStatusID.Unknown
+            )
+
+            finding_severity = getattr(
+                SeverityID,
+                finding.metadata.Severity.capitalize(),
+                SeverityID.Unknown,
+            )
+            event_status = OCSF.get_finding_status_id(finding.muted)
+
+            time_value = (
+                int(finding.timestamp.timestamp())
+                if isinstance(finding.timestamp, datetime)
+                else finding.timestamp
+            )
+
+            cf = ComplianceFinding(
+                activity_id=ActivityID.Create.value,
+                activity_name=ActivityID.Create.name,
+                compliance=Compliance(
+                    standards=[compliance_name],
+                    requirements=[requirement.id],
+                    control=requirement.description,
+                    status_id=compliance_status,
+                    checks=[
+                        Check(
+                            uid=finding.check_id,
+                            name=finding.metadata.CheckTitle,
+                            desc=finding.metadata.Description,
+                            status=finding.status,
+                            status_id=check_status,
+                        )
+                    ],
+                ),
+                finding_info=FindingInformation(
+                    uid=f"{finding.uid}-{requirement.id}",
+                    title=requirement.id,
+                    desc=requirement.description,
+                    created_time=time_value,
+                    created_time_dt=(
+                        finding.timestamp
+                        if isinstance(finding.timestamp, datetime)
+                        else None
+                    ),
+                ),
+                message=finding.status_extended,
+                metadata=Metadata(
+                    event_code=finding.check_id,
+                    product=Product(
+                        uid="prowler",
+                        name="Prowler",
+                        vendor_name="Prowler",
+                        version=finding.prowler_version,
+                    ),
+                    profiles=(
+                        ["cloud", "datetime"]
+                        if finding.provider != "kubernetes"
+                        else ["container", "datetime"]
+                    ),
+                    tenant_uid=finding.account_organization_uid,
+                ),
+                resources=[
+                    ResourceDetails(
+                        labels=unroll_dict_to_list(finding.resource_tags),
+                        name=finding.resource_name,
+                        uid=finding.resource_uid,
+                        group=Group(name=finding.metadata.ServiceName),
+                        type=finding.metadata.ResourceType,
+                        cloud_partition=(
+                            finding.partition
+                            if finding.provider != "kubernetes"
+                            else None
+                        ),
+                        region=(
+                            finding.region if finding.provider != "kubernetes" else None
+                        ),
+                        namespace=(
+                            finding.region.replace("namespace: ", "")
+                            if finding.provider == "kubernetes"
+                            else None
+                        ),
+                        data={
+                            "details": finding.resource_details,
+                            "metadata": finding.resource_metadata,
+                        },
+                    )
+                ],
+                severity_id=finding_severity.value,
+                severity=finding_severity.name,
+                status_id=event_status.value,
+                status=event_status.name,
+                status_code=finding.status,
+                status_detail=finding.status_extended,
+                time=time_value,
+                time_dt=(
+                    finding.timestamp
+                    if isinstance(finding.timestamp, datetime)
+                    else None
+                ),
+                type_uid=ComplianceFindingTypeID.Create,
+                type_name=f"Compliance Finding: {ComplianceFindingTypeID.Create.name}",
+                unmapped=self._build_unmapped(finding, requirement, framework),
+            )
+
+            return cf
+        except Exception as e:
+            logger.debug(f"Skipping OCSF compliance finding for {requirement.id}: {e}")
+            return None
+
+    def _build_manual_compliance_finding(
+        self,
+        framework: ComplianceFramework,
+        requirement,
+        compliance_name: str,
+    ) -> ComplianceFinding:
+        try:
+            from prowler.config.config import timestamp as config_timestamp
+
+            time_value = int(config_timestamp.timestamp())
+
+            return ComplianceFinding(
+                activity_id=ActivityID.Create.value,
+                activity_name=ActivityID.Create.name,
+                compliance=Compliance(
+                    standards=[compliance_name],
+                    requirements=[requirement.id],
+                    control=requirement.description,
+                    status_id=ComplianceStatusID.Unknown,
+                ),
+                finding_info=FindingInformation(
+                    uid=f"manual-{requirement.id}",
+                    title=requirement.id,
+                    desc=requirement.description,
+                    created_time=time_value,
+                ),
+                message="Manual check",
+                metadata=Metadata(
+                    event_code="manual",
+                    product=Product(
+                        uid="prowler",
+                        name="Prowler",
+                        vendor_name="Prowler",
+                        version=prowler_version,
+                    ),
+                ),
+                severity_id=SeverityID.Informational.value,
+                severity=SeverityID.Informational.name,
+                status_id=EventStatusID.New.value,
+                status=EventStatusID.New.name,
+                status_code="MANUAL",
+                status_detail="Manual check",
+                time=time_value,
+                type_uid=ComplianceFindingTypeID.Create,
+                type_name=f"Compliance Finding: {ComplianceFindingTypeID.Create.name}",
+                unmapped=self._build_unmapped(None, requirement, framework),
+            )
+        except Exception as e:
+            logger.debug(
+                f"Skipping manual OCSF compliance finding for {requirement.id}: {e}"
+            )
+            return None
+
+    def _create_file_descriptor(self, file_path: str) -> None:
+        try:
+            self._file_descriptor = open_file(file_path, "a")
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def batch_write_data_to_file(self) -> None:
+        """Write ComplianceFinding events to a JSON array file."""
+        try:
+            if (
+                getattr(self, "_file_descriptor", None)
+                and not self._file_descriptor.closed
+                and self._data
+            ):
+                if self._file_descriptor.tell() == 0:
+                    self._file_descriptor.write("[")
+                for finding in self._data:
+                    try:
+                        if hasattr(finding, "model_dump_json"):
+                            json_output = finding.model_dump_json(
+                                exclude_none=True, indent=4
+                            )
+                        else:
+                            json_output = finding.json(exclude_none=True, indent=4)
+                        self._file_descriptor.write(json_output)
+                        self._file_descriptor.write(",")
+                    except Exception as error:
+                        logger.error(
+                            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                        )
+                if self.close_file or self._from_cli:
+                    if self._file_descriptor.tell() != 1:
+                        self._file_descriptor.seek(
+                            self._file_descriptor.tell() - 1, os.SEEK_SET
+                        )
+                    self._file_descriptor.truncate()
+                    self._file_descriptor.write("]")
+                    self._file_descriptor.close()
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )

prowler/lib/outputs/compliance/universal/universal_table.py

@@ -0,0 +1,490 @@
+from colorama import Fore, Style
+from tabulate import tabulate
+
+from prowler.config.config import orange_color
+from prowler.lib.check.compliance_models import ComplianceFramework
+
+
+def get_universal_table(
+    findings: list,
+    bulk_checks_metadata: dict,
+    compliance_framework_name: str,
+    output_filename: str,
+    output_directory: str,
+    compliance_overview: bool,
+    framework: ComplianceFramework = None,
+    provider: str = None,
+    output_formats: list = None,
+) -> None:
+    """Render a compliance console table driven by TableConfig.
+
+    Supports 3 modes:
+    - Grouped: group_by only (generic, C5, CSA, ISO, KISA)
+    - Split: group_by + split_by (CIS Level 1/2, ENS alto/medio/bajo/opcional)
+    - Scored: group_by + scoring (ThreatScore weighted risk %)
+
+    When ``provider`` is given and ``checks`` is a multi-provider dict,
+    only the checks for that provider are matched against findings.
+    """
+    if framework is None or not framework.outputs or not framework.outputs.table_config:
+        return
+
+    tc = framework.outputs.table_config
+    labels = tc.labels or _default_labels()
+
+    group_by = tc.group_by
+    split_by = tc.split_by
+    scoring = tc.scoring
+
+    if scoring:
+        _render_scored(
+            findings,
+            bulk_checks_metadata,
+            compliance_framework_name,
+            output_filename,
+            output_directory,
+            compliance_overview,
+            framework,
+            group_by,
+            scoring,
+            labels,
+            provider,
+            output_formats=output_formats,
+        )
+    elif split_by:
+        _render_split(
+            findings,
+            bulk_checks_metadata,
+            compliance_framework_name,
+            output_filename,
+            output_directory,
+            compliance_overview,
+            framework,
+            group_by,
+            split_by,
+            labels,
+            provider,
+            output_formats=output_formats,
+        )
+    else:
+        _render_grouped(
+            findings,
+            bulk_checks_metadata,
+            compliance_framework_name,
+            output_filename,
+            output_directory,
+            compliance_overview,
+            framework,
+            group_by,
+            labels,
+            provider,
+            output_formats=output_formats,
+        )
+
+
+def _default_labels():
+    """Return a simple namespace with default labels."""
+    from prowler.lib.check.compliance_models import TableLabels
+
+    return TableLabels()
+
+
+def _build_requirement_check_map(framework, provider=None):
+    """Build a map of check_id -> list of requirements for fast lookup.
+
+    When *provider* is given, only the checks for that provider are included.
+    """
+    check_map = {}
+    for req in framework.requirements:
+        checks = req.checks
+        if provider:
+            all_checks = checks.get(provider.lower(), [])
+        else:
+            all_checks = []
+            for check_list in checks.values():
+                all_checks.extend(check_list)
+        for check_id in all_checks:
+            if check_id not in check_map:
+                check_map[check_id] = []
+            check_map[check_id].append(req)
+    return check_map
+
+
+def _get_group_key(req, group_by):
+    """Extract the group key from a requirement."""
+    if group_by == "_Tactics":
+        return req.tactics or []
+    return [req.attributes.get(group_by, "Unknown")]
+
+
+def _print_overview(pass_count, fail_count, muted_count, framework_name, labels):
+    """Print the overview pass/fail/muted summary."""
+    total = len(fail_count) + len(pass_count) + len(muted_count)
+    if total < 2:
+        return False
+
+    title = (
+        labels.title
+        or f"Compliance Status of {Fore.YELLOW}{framework_name.upper()}{Style.RESET_ALL} Framework:"
+    )
+    print(f"\n{title}")
+
+    fail_pct = round(len(fail_count) / total * 100, 2)
+    pass_pct = round(len(pass_count) / total * 100, 2)
+    muted_pct = round(len(muted_count) / total * 100, 2)
+
+    fail_label = labels.fail_label
+    pass_label = labels.pass_label
+
+    overview_table = [
+        [
+            f"{Fore.RED}{fail_pct}% ({len(fail_count)}) {fail_label}{Style.RESET_ALL}",
+            f"{Fore.GREEN}{pass_pct}% ({len(pass_count)}) {pass_label}{Style.RESET_ALL}",
+            f"{orange_color}{muted_pct}% ({len(muted_count)}) MUTED{Style.RESET_ALL}",
+        ]
+    ]
+    print(tabulate(overview_table, tablefmt="rounded_grid"))
+    return True
+
+
+def _render_grouped(
+    findings,
+    bulk_checks_metadata,
+    compliance_framework_name,
+    output_filename,
+    output_directory,
+    compliance_overview,
+    framework,
+    group_by,
+    labels,
+    provider=None,
+    output_formats=None,
+):
+    """Grouped mode: one row per group with pass/fail counts."""
+    check_map = _build_requirement_check_map(framework, provider)
+    groups = {}
+    pass_count = []
+    fail_count = []
+    muted_count = []
+
+    for index, finding in enumerate(findings):
+        check_id = finding.check_metadata.CheckID
+        if check_id not in check_map:
+            continue
+
+        for req in check_map[check_id]:
+            for group_key in _get_group_key(req, group_by):
+                if group_key not in groups:
+                    groups[group_key] = {"FAIL": 0, "PASS": 0, "Muted": 0}
+
+                if finding.muted:
+                    if index not in muted_count:
+                        muted_count.append(index)
+                        groups[group_key]["Muted"] += 1
+                else:
+                    if finding.status == "FAIL" and index not in fail_count:
+                        fail_count.append(index)
+                        groups[group_key]["FAIL"] += 1
+                    elif finding.status == "PASS" and index not in pass_count:
+                        pass_count.append(index)
+                        groups[group_key]["PASS"] += 1
+
+    if not _print_overview(
+        pass_count, fail_count, muted_count, compliance_framework_name, labels
+    ):
+        return
+
+    if not compliance_overview:
+        provider_header = labels.provider_header
+        group_header = labels.group_header or group_by
+        table = {
+            provider_header: [],
+            group_header: [],
+            labels.status_header: [],
+            "Muted": [],
+        }
+        for group_key in sorted(groups):
+            table[provider_header].append(
+                framework.provider or (provider.upper() if provider else "")
+            )
+            table[group_header].append(group_key)
+            if groups[group_key]["FAIL"] > 0:
+                table[labels.status_header].append(
+                    f"{Fore.RED}{labels.fail_label}({groups[group_key]['FAIL']}){Style.RESET_ALL}"
+                )
+            else:
+                table[labels.status_header].append(
+                    f"{Fore.GREEN}{labels.pass_label}({groups[group_key]['PASS']}){Style.RESET_ALL}"
+                )
+            table["Muted"].append(
+                f"{orange_color}{groups[group_key]['Muted']}{Style.RESET_ALL}"
+            )
+
+        results_title = (
+            labels.results_title
+            or f"Framework {Fore.YELLOW}{compliance_framework_name.upper()}{Style.RESET_ALL} Results:"
+        )
+        print(f"\n{results_title}")
+        print(tabulate(table, headers="keys", tablefmt="rounded_grid"))
+        footer = labels.footer_note or "* Only sections containing results appear."
+        print(f"{Style.BRIGHT}{footer}{Style.RESET_ALL}")
+        print(f"\nDetailed results of {compliance_framework_name.upper()} are in:")
+        print(
+            f" - CSV: {output_directory}/compliance/{output_filename}_{compliance_framework_name}.csv"
+        )
+        if "json-ocsf" in (output_formats or []):
+            print(
+                f" - OCSF: {output_directory}/compliance/{output_filename}_{compliance_framework_name}.ocsf.json"
+            )
+        print()
+
+
+def _render_split(
+    findings,
+    bulk_checks_metadata,
+    compliance_framework_name,
+    output_filename,
+    output_directory,
+    compliance_overview,
+    framework,
+    group_by,
+    split_by,
+    labels,
+    provider=None,
+    output_formats=None,
+):
+    """Split mode: one row per group with columns for each split value (e.g. Level 1/Level 2)."""
+    check_map = _build_requirement_check_map(framework, provider)
+    split_field = split_by.field
+    split_values = split_by.values
+    groups = {}
+    pass_count = []
+    fail_count = []
+    muted_count = []
+
+    for index, finding in enumerate(findings):
+        check_id = finding.check_metadata.CheckID
+        if check_id not in check_map:
+            continue
+
+        for req in check_map[check_id]:
+            for group_key in _get_group_key(req, group_by):
+                if group_key not in groups:
+                    groups[group_key] = {
+                        sv: {"FAIL": 0, "PASS": 0} for sv in split_values
+                    }
+                    groups[group_key]["Muted"] = 0
+
+                split_val = req.attributes.get(split_field, "")
+
+                if finding.muted:
+                    if index not in muted_count:
+                        muted_count.append(index)
+                        groups[group_key]["Muted"] += 1
+                else:
+                    if finding.status == "FAIL" and index not in fail_count:
+                        fail_count.append(index)
+                    elif finding.status == "PASS" and index not in pass_count:
+                        pass_count.append(index)
+
+                    for sv in split_values:
+                        if sv in str(split_val):
+                            if not finding.muted:
+                                if finding.status == "FAIL":
+                                    groups[group_key][sv]["FAIL"] += 1
+                                else:
+                                    groups[group_key][sv]["PASS"] += 1
+
+    if not _print_overview(
+        pass_count, fail_count, muted_count, compliance_framework_name, labels
+    ):
+        return
+
+    if not compliance_overview:
+        provider_header = labels.provider_header
+        group_header = labels.group_header or group_by
+        table = {provider_header: [], group_header: []}
+        for sv in split_values:
+            table[sv] = []
+        table["Muted"] = []
+
+        for group_key in sorted(groups):
+            table[provider_header].append(
+                framework.provider or (provider.upper() if provider else "")
+            )
+            table[group_header].append(group_key)
+            for sv in split_values:
+                if groups[group_key][sv]["FAIL"] > 0:
+                    table[sv].append(
+                        f"{Fore.RED}{labels.fail_label}({groups[group_key][sv]['FAIL']}){Style.RESET_ALL}"
+                    )
+                else:
+                    table[sv].append(
+                        f"{Fore.GREEN}{labels.pass_label}({groups[group_key][sv]['PASS']}){Style.RESET_ALL}"
+                    )
+            table["Muted"].append(
+                f"{orange_color}{groups[group_key]['Muted']}{Style.RESET_ALL}"
+            )
+
+        results_title = (
+            labels.results_title
+            or f"Framework {Fore.YELLOW}{compliance_framework_name.upper()}{Style.RESET_ALL} Results:"
+        )
+        print(f"\n{results_title}")
+        print(tabulate(table, headers="keys", tablefmt="rounded_grid"))
+        footer = labels.footer_note or "* Only sections containing results appear."
+        print(f"{Style.BRIGHT}{footer}{Style.RESET_ALL}")
+        print(f"\nDetailed results of {compliance_framework_name.upper()} are in:")
+        print(
+            f" - CSV: {output_directory}/compliance/{output_filename}_{compliance_framework_name}.csv"
+        )
+        if "json-ocsf" in (output_formats or []):
+            print(
+                f" - OCSF: {output_directory}/compliance/{output_filename}_{compliance_framework_name}.ocsf.json"
+            )
+        print()
+
+
+def _render_scored(
+    findings,
+    bulk_checks_metadata,
+    compliance_framework_name,
+    output_filename,
+    output_directory,
+    compliance_overview,
+    framework,
+    group_by,
+    scoring,
+    labels,
+    provider=None,
+    output_formats=None,
+):
+    """Scored mode: weighted risk scoring per group (e.g. ThreatScore)."""
+    check_map = _build_requirement_check_map(framework, provider)
+    risk_field = scoring.risk_field
+    weight_field = scoring.weight_field
+    groups = {}
+    pass_count = []
+    fail_count = []
+    muted_count = []
+
+    score_per_group = {}
+    max_score_per_group = {}
+    counted_per_group = {}
+    generic_score = 0
+    max_generic_score = 0
+    counted_generic = []
+
+    for index, finding in enumerate(findings):
+        check_id = finding.check_metadata.CheckID
+        if check_id not in check_map:
+            continue
+
+        for req in check_map[check_id]:
+            for group_key in _get_group_key(req, group_by):
+                attrs = req.attributes
+                risk = attrs.get(risk_field, 0)
+                weight = attrs.get(weight_field, 0)
+
+                if group_key not in groups:
+                    groups[group_key] = {"FAIL": 0, "PASS": 0, "Muted": 0}
+                    score_per_group[group_key] = 0
+                    max_score_per_group[group_key] = 0
+                    counted_per_group[group_key] = []
+
+                if index not in counted_per_group[group_key] and not finding.muted:
+                    if finding.status == "PASS":
+                        score_per_group[group_key] += risk * weight
+                    max_score_per_group[group_key] += risk * weight
+                    counted_per_group[group_key].append(index)
+
+                if finding.muted:
+                    if index not in muted_count:
+                        muted_count.append(index)
+                        groups[group_key]["Muted"] += 1
+                else:
+                    if finding.status == "FAIL" and index not in fail_count:
+                        fail_count.append(index)
+                        groups[group_key]["FAIL"] += 1
+                    elif finding.status == "PASS" and index not in pass_count:
+                        pass_count.append(index)
+                        groups[group_key]["PASS"] += 1
+
+                if index not in counted_generic and not finding.muted:
+                    if finding.status == "PASS":
+                        generic_score += risk * weight
+                    max_generic_score += risk * weight
+                    counted_generic.append(index)
+
+    if not _print_overview(
+        pass_count, fail_count, muted_count, compliance_framework_name, labels
+    ):
+        return
+
+    if not compliance_overview:
+        provider_header = labels.provider_header
+        group_header = labels.group_header or group_by
+        table = {
+            provider_header: [],
+            group_header: [],
+            labels.status_header: [],
+            "Score": [],
+            "Muted": [],
+        }
+
+        for group_key in sorted(groups):
+            table[provider_header].append(
+                framework.provider or (provider.upper() if provider else "")
+            )
+            table[group_header].append(group_key)
+            if max_score_per_group[group_key] == 0:
+                group_score = 100.0
+                score_color = Fore.GREEN
+            else:
+                group_score = (
+                    score_per_group[group_key] / max_score_per_group[group_key]
+                ) * 100
+                score_color = Fore.RED
+            table["Score"].append(
+                f"{Style.BRIGHT}{score_color}{group_score:.2f}%{Style.RESET_ALL}"
+            )
+            if groups[group_key]["FAIL"] > 0:
+                table[labels.status_header].append(
+                    f"{Fore.RED}{labels.fail_label}({groups[group_key]['FAIL']}){Style.RESET_ALL}"
+                )
+            else:
+                table[labels.status_header].append(
+                    f"{Fore.GREEN}{labels.pass_label}({groups[group_key]['PASS']}){Style.RESET_ALL}"
+                )
+            table["Muted"].append(
+                f"{orange_color}{groups[group_key]['Muted']}{Style.RESET_ALL}"
+            )
+
+        if max_generic_score == 0:
+            generic_threat_score = 100.0
+        else:
+            generic_threat_score = generic_score / max_generic_score * 100
+
+        results_title = (
+            labels.results_title
+            or f"Framework {Fore.YELLOW}{compliance_framework_name.upper()}{Style.RESET_ALL} Results:"
+        )
+        print(f"\n{results_title}")
+        print(f"\nGeneric Threat Score: {generic_threat_score:.2f}%")
+        print(tabulate(table, headers="keys", tablefmt="rounded_grid"))
+        footer = labels.footer_note or (
+            f"{Style.BRIGHT}\n=== Threat Score Guide ===\n"
+            f"The lower the score, the higher the risk.{Style.RESET_ALL}\n"
+            f"{Style.BRIGHT}(Only sections containing results appear, the score is calculated as the sum of the "
+            f"level of risk * weight of the passed findings divided by the sum of the risk * weight of all the findings){Style.RESET_ALL}"
+        )
+        print(footer)
+        print(f"\nDetailed results of {compliance_framework_name.upper()} are in:")
+        print(
+            f" - CSV: {output_directory}/compliance/{output_filename}_{compliance_framework_name}.csv"
+        )
+        if "json-ocsf" in (output_formats or []):
+            print(
+                f" - OCSF: {output_directory}/compliance/{output_filename}_{compliance_framework_name}.ocsf.json"
+            )
+        print()

prowler/lib/outputs/finding.py

@@ -354,6 +354,9 @@ class Finding(BaseModel):
                     check_output, "resource_line_range", ""
                 )
                 output_data["framework"] = check_output.check_metadata.ServiceName
+                output_data["raw"] = {
+                    "resource_line_range": output_data.get("resource_line_range", ""),
+                }
 
             elif provider.type == "llm":
                 output_data["auth_method"] = provider.auth_method

prowler/lib/outputs/sarif/sarif.py

@@ -0,0 +1,191 @@
+from json import dump
+from typing import Optional
+
+from prowler.config.config import prowler_version
+from prowler.lib.logger import logger
+from prowler.lib.outputs.finding import Finding
+from prowler.lib.outputs.output import Output
+
+SARIF_SCHEMA_URL = "https://json.schemastore.org/sarif-2.1.0.json"
+SARIF_VERSION = "2.1.0"
+
+SEVERITY_TO_SARIF_LEVEL = {
+    "critical": "error",
+    "high": "error",
+    "medium": "warning",
+    "low": "note",
+    "informational": "note",
+}
+
+SEVERITY_TO_SECURITY_SEVERITY = {
+    "critical": "9.0",
+    "high": "7.0",
+    "medium": "4.0",
+    "low": "2.0",
+    "informational": "0.0",
+}
+
+
+class SARIF(Output):
+    """Generates SARIF 2.1.0 output compatible with GitHub Code Scanning."""
+
+    def transform(self, findings: list[Finding]) -> None:
+        """Transform findings into a SARIF 2.1.0 document.
+
+        Only FAIL findings that are not muted are included. Each unique
+        check ID produces one rule entry; multiple findings for the same
+        check share the rule via ruleIndex.
+
+        Args:
+            findings: List of Finding objects to transform.
+        """
+        rules = {}
+        rule_indices = {}
+        results = []
+
+        for finding in findings:
+            if finding.status != "FAIL" or finding.muted:
+                continue
+
+            check_id = finding.metadata.CheckID
+            severity = finding.metadata.Severity.lower()
+
+            if check_id not in rules:
+                rule_indices[check_id] = len(rules)
+                rule = {
+                    "id": check_id,
+                    "name": finding.metadata.CheckTitle,
+                    "shortDescription": {"text": finding.metadata.CheckTitle},
+                    "fullDescription": {
+                        "text": finding.metadata.Description or check_id
+                    },
+                    "help": {
+                        "text": finding.metadata.Remediation.Recommendation.Text
+                        or finding.metadata.Description
+                        or check_id,
+                        "markdown": self._build_help_markdown(finding, severity),
+                    },
+                    "defaultConfiguration": {
+                        "level": SEVERITY_TO_SARIF_LEVEL.get(severity, "note"),
+                    },
+                    "properties": {
+                        "tags": [
+                            "security",
+                            f"prowler/{finding.metadata.Provider}",
+                            f"severity/{severity}",
+                        ],
+                        "security-severity": SEVERITY_TO_SECURITY_SEVERITY.get(
+                            severity, "0.0"
+                        ),
+                    },
+                }
+                if finding.metadata.RelatedUrl:
+                    rule["helpUri"] = finding.metadata.RelatedUrl
+                rules[check_id] = rule
+
+            rule_index = rule_indices[check_id]
+            result = {
+                "ruleId": check_id,
+                "ruleIndex": rule_index,
+                "level": SEVERITY_TO_SARIF_LEVEL.get(severity, "note"),
+                "message": {
+                    "text": finding.status_extended or finding.metadata.CheckTitle
+                },
+            }
+
+            location = self._build_location(finding)
+            if location is not None:
+                result["locations"] = [location]
+
+            results.append(result)
+
+        sarif_document = {
+            "$schema": SARIF_SCHEMA_URL,
+            "version": SARIF_VERSION,
+            "runs": [
+                {
+                    "tool": {
+                        "driver": {
+                            "name": "Prowler",
+                            "version": prowler_version,
+                            "informationUri": "https://prowler.com",
+                            "rules": list(rules.values()),
+                        },
+                    },
+                    "results": results,
+                },
+            ],
+        }
+
+        self._data = [sarif_document]
+
+    def batch_write_data_to_file(self) -> None:
+        """Write the SARIF document to the output file as JSON."""
+        try:
+            if (
+                getattr(self, "_file_descriptor", None)
+                and not self._file_descriptor.closed
+                and self._data
+            ):
+                dump(self._data[0], self._file_descriptor, indent=2)
+                if self.close_file or self._from_cli:
+                    self._file_descriptor.close()
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    @staticmethod
+    def _build_help_markdown(finding: Finding, severity: str) -> str:
+        """Build a markdown help string for a SARIF rule."""
+        remediation = (
+            finding.metadata.Remediation.Recommendation.Text
+            or finding.metadata.Description
+            or finding.metadata.CheckID
+        )
+        lines = [
+            f"**{finding.metadata.CheckTitle}**\n",
+            "| Severity | Remediation |",
+            "| --- | --- |",
+            f"| {severity.upper()} | {remediation} |",
+        ]
+        if finding.metadata.RelatedUrl:
+            lines.append(f"\n[More info]({finding.metadata.RelatedUrl})")
+        return "\n".join(lines)
+
+    @staticmethod
+    def _build_location(finding: Finding) -> Optional[dict]:
+        """Build a SARIF physicalLocation from a Finding.
+
+        Uses resource_name as the artifact URI and resource_line_range
+        (stored in finding.raw for IaC findings) for line range info.
+
+        Returns:
+            A SARIF location dict, or None if resource_name is empty.
+        """
+        if not finding.resource_name:
+            return None
+
+        location = {
+            "physicalLocation": {
+                "artifactLocation": {
+                    "uri": finding.resource_name,
+                },
+            },
+        }
+
+        line_range = finding.raw.get("resource_line_range", "")
+        if line_range and ":" in line_range:
+            parts = line_range.split(":")
+            try:
+                start_line = int(parts[0])
+                end_line = int(parts[1])
+                if start_line >= 1 and end_line >= 1:
+                    location["physicalLocation"]["region"] = {
+                        "startLine": start_line,
+                        "endLine": end_line,
+                    }
+            except (ValueError, IndexError):
+                pass  # Malformed line range — skip region, keep location
+
+        return location

prowler/lib/outputs/summary_table.py

@@ -9,6 +9,7 @@ from prowler.config.config import (
     json_asff_file_suffix,
     json_ocsf_file_suffix,
     orange_color,
+    sarif_file_suffix,
 )
 from prowler.lib.logger import logger
 from prowler.providers.github.models import GithubAppIdentityInfo, GithubIdentityInfo
@@ -210,6 +211,10 @@ def display_summary_table(
                 print(
                     f" - HTML: {output_directory}/{output_filename}{html_file_suffix}"
                 )
+            if "sarif" in output_options.output_modes:
+                print(
+                    f" - SARIF: {output_directory}/{output_filename}{sarif_file_suffix}"
+                )
 
         else:
             print(

prowler/providers/common/arguments.py

@@ -80,3 +80,19 @@ def validate_asff_usage(
         False,
         f"json-asff output format is only available for the aws provider, but {provider} was selected",
     )
+
+
+def validate_sarif_usage(
+    provider: Optional[str], output_formats: Optional[Sequence[str]]
+) -> tuple[bool, str]:
+    """Ensure sarif output is only requested for the IaC provider."""
+    if not output_formats or "sarif" not in output_formats:
+        return (True, "")
+
+    if provider == "iac":
+        return (True, "")
+
+    return (
+        False,
+        f"sarif output format is only available for the iac provider, but {provider} was selected",
+    )

tests/lib/outputs/compliance/universal/ocsf_compliance_test.py

@@ -0,0 +1,475 @@
+import json
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+from py_ocsf_models.events.findings.compliance_finding import ComplianceFinding
+from py_ocsf_models.events.findings.compliance_finding_type_id import (
+    ComplianceFindingTypeID,
+)
+from py_ocsf_models.objects.compliance_status import StatusID as ComplianceStatusID
+
+from prowler.lib.check.compliance_models import (
+    AttributeMetadata,
+    ComplianceFramework,
+    OutputFormats,
+    OutputsConfig,
+    TableConfig,
+    UniversalComplianceRequirement,
+)
+from prowler.lib.outputs.compliance.universal.ocsf_compliance import (
+    OCSFComplianceOutput,
+)
+
+
+def _make_finding(check_id, status="PASS", provider="aws"):
+    """Create a mock Finding with all fields needed by OCSFComplianceOutput."""
+    finding = SimpleNamespace()
+    finding.provider = provider
+    finding.account_uid = "123456789012"
+    finding.account_name = "test-account"
+    finding.account_email = ""
+    finding.account_organization_uid = "org-123"
+    finding.account_organization_name = "test-org"
+    finding.account_tags = {"env": "test"}
+    finding.region = "us-east-1"
+    finding.status = status
+    finding.status_extended = f"{check_id} is {status}"
+    finding.resource_uid = f"arn:aws:iam::123456789012:{check_id}"
+    finding.resource_name = check_id
+    finding.resource_details = "some details"
+    finding.resource_metadata = {}
+    finding.resource_tags = {"Name": "test"}
+    finding.partition = "aws"
+    finding.muted = False
+    finding.check_id = check_id
+    finding.uid = "test-finding-uid"
+    finding.timestamp = datetime(2025, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+    finding.prowler_version = "5.0.0"
+    finding.compliance = {}
+    finding.metadata = SimpleNamespace(
+        Provider=provider,
+        CheckID=check_id,
+        CheckTitle=f"Title for {check_id}",
+        CheckType=["test-type"],
+        Description=f"Description for {check_id}",
+        Severity="medium",
+        ServiceName="iam",
+        ResourceType="aws-iam-role",
+        Risk="test-risk",
+        RelatedUrl="https://example.com",
+        Remediation=SimpleNamespace(
+            Recommendation=SimpleNamespace(Text="Fix it", Url="https://fix.com"),
+        ),
+        DependsOn=[],
+        RelatedTo=[],
+        Categories=["test"],
+        Notes="",
+        AdditionalURLs=[],
+    )
+    return finding
+
+
+def _make_framework(requirements, attrs_metadata=None):
+    return ComplianceFramework(
+        framework="TestFW",
+        name="Test Framework",
+        provider="AWS",
+        version="1.0",
+        description="Test framework",
+        requirements=requirements,
+        attributes_metadata=attrs_metadata,
+        outputs=OutputsConfig(table_config=TableConfig(group_by="Section")),
+    )
+
+
+def _simple_requirement(req_id="REQ-1", checks=None):
+    if checks is None:
+        checks_dict = {"aws": ["check_a"]}
+    elif isinstance(checks, dict):
+        checks_dict = checks
+    else:
+        checks_dict = {"aws": list(checks)} if checks else {}
+    return UniversalComplianceRequirement(
+        id=req_id,
+        description=f"Description for {req_id}",
+        attributes={},
+        checks=checks_dict,
+    )
+
+
+class TestOCSFComplianceOutput:
+    def test_transform_basic(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert len(output.data) == 1
+        assert isinstance(output.data[0], ComplianceFinding)
+
+    def test_class_uid(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert output.data[0].class_uid == 2003
+
+    def test_type_uid(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert output.data[0].type_uid == ComplianceFindingTypeID.Create
+        assert output.data[0].type_uid == 200301
+
+    def test_compliance_object_fields(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        compliance = output.data[0].compliance
+        assert compliance.standards == ["TestFW-1.0"]
+        assert compliance.requirements == ["REQ-1"]
+        assert compliance.control == "Description for REQ-1"
+        assert compliance.status_id == ComplianceStatusID.Pass
+
+    def test_check_object_fields(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a", "FAIL")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        checks = output.data[0].compliance.checks
+        assert len(checks) == 1
+        assert checks[0].uid == "check_a"
+        assert checks[0].name == "Title for check_a"
+        assert checks[0].status == "FAIL"
+        assert checks[0].status_id == ComplianceStatusID.Fail
+
+    def test_finding_info_fields(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        info = output.data[0].finding_info
+        assert info.uid == "test-finding-uid-REQ-1"
+        assert info.title == "REQ-1"
+        assert info.desc == "Description for REQ-1"
+
+    def test_metadata_fields(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        metadata = output.data[0].metadata
+        assert metadata.product.name == "Prowler"
+        assert metadata.product.uid == "prowler"
+        assert metadata.event_code == "check_a"
+
+    def test_status_mapping_pass(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert output.data[0].compliance.status_id == ComplianceStatusID.Pass
+
+    def test_status_mapping_fail(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a", "FAIL")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert output.data[0].compliance.status_id == ComplianceStatusID.Fail
+
+    def test_manual_requirement(self):
+        req = _simple_requirement("MANUAL-1", checks=[])
+        fw = _make_framework([req])
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert len(output.data) == 1
+        cf = output.data[0]
+        assert cf.compliance.status_id == ComplianceStatusID.Unknown
+        assert cf.status_code == "MANUAL"
+        assert cf.finding_info.uid == "manual-MANUAL-1"
+
+    def test_multi_provider_checks_dict(self):
+        req = UniversalComplianceRequirement(
+            id="REQ-1",
+            description="Multi-provider req",
+            attributes={},
+            checks={"aws": ["check_a"], "azure": ["check_b"]},
+        )
+        fw = _make_framework([req])
+        findings = [_make_finding("check_a", "PASS", provider="aws")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert len(output.data) == 1
+        assert output.data[0].compliance.checks[0].uid == "check_a"
+
+    def test_empty_findings(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+
+        output = OCSFComplianceOutput(findings=[], framework=fw, provider="aws")
+
+        assert output.data == []
+
+    def test_cloud_info_in_unmapped(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a", provider="aws")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        cf = output.data[0]
+        assert cf.unmapped is not None
+        assert cf.unmapped["cloud"]["provider"] == "aws"
+        assert cf.unmapped["cloud"]["account"]["uid"] == "123456789012"
+        assert cf.unmapped["cloud"]["account"]["name"] == "test-account"
+
+    def test_resources_populated(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        resources = output.data[0].resources
+        assert len(resources) == 1
+        assert resources[0].name == "check_a"
+        assert resources[0].uid == "arn:aws:iam::123456789012:check_a"
+        assert resources[0].type == "aws-iam-role"
+
+    def test_batch_write_to_file(self, tmp_path):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [_make_finding("check_a", "PASS")]
+        filepath = str(tmp_path / "compliance.ocsf.json")
+
+        output = OCSFComplianceOutput(
+            findings=findings, framework=fw, file_path=filepath, provider="aws"
+        )
+        output.batch_write_data_to_file()
+
+        with open(filepath) as f:
+            data = json.load(f)
+
+        assert isinstance(data, list)
+        assert len(data) == 1
+        assert data[0]["class_uid"] == 2003
+        assert data[0]["compliance"]["standards"] == ["TestFW-1.0"]
+        assert data[0]["compliance"]["requirements"] == ["REQ-1"]
+
+    def test_multiple_findings_same_requirement(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        findings = [
+            _make_finding("check_a", "PASS"),
+            _make_finding("check_a", "FAIL"),
+        ]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert len(output.data) == 2
+        statuses = [cf.compliance.status_id for cf in output.data]
+        assert ComplianceStatusID.Pass in statuses
+        assert ComplianceStatusID.Fail in statuses
+
+    def test_requirement_attributes_in_unmapped(self):
+        req = UniversalComplianceRequirement(
+            id="REQ-1",
+            description="Test requirement",
+            attributes={"Section": "IAM", "Profile": "Level 1"},
+            checks={"aws": ["check_a"]},
+        )
+        fw = _make_framework([req])
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        cf = output.data[0]
+        assert cf.unmapped is not None
+        assert "requirement_attributes" in cf.unmapped
+        assert cf.unmapped["requirement_attributes"]["section"] == "IAM"
+        assert cf.unmapped["requirement_attributes"]["profile"] == "Level 1"
+
+    def test_requirement_attributes_keys_are_snake_case(self):
+        req = UniversalComplianceRequirement(
+            id="REQ-1",
+            description="Test requirement",
+            attributes={"Section": "IAM", "CCMLite": "Yes", "SubSection": "1.1"},
+            checks={"aws": ["check_a"]},
+        )
+        fw = _make_framework([req])
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        attrs = output.data[0].unmapped["requirement_attributes"]
+        assert "section" in attrs
+        assert "ccm_lite" in attrs
+        assert "sub_section" in attrs
+
+    def test_requirement_attributes_empty_attrs_excluded(self):
+        req = _simple_requirement("REQ-1", checks=["check_a"])
+        fw = _make_framework([req])
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        cf = output.data[0]
+        # Cloud info is still present, but no requirement_attributes key
+        assert cf.unmapped is not None
+        assert "cloud" in cf.unmapped
+        assert "requirement_attributes" not in cf.unmapped
+
+    def test_manual_requirement_has_attributes_in_unmapped(self):
+        req = UniversalComplianceRequirement(
+            id="MANUAL-1",
+            description="Manual check",
+            attributes={"Section": "Logging", "Type": "manual"},
+            checks={},
+        )
+        fw = _make_framework([req])
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        assert len(output.data) == 1
+        cf = output.data[0]
+        assert cf.unmapped is not None
+        assert cf.unmapped["requirement_attributes"]["section"] == "Logging"
+        assert cf.unmapped["requirement_attributes"]["type"] == "manual"
+        # Manual findings have no cloud info (finding is None)
+        assert "cloud" not in cf.unmapped
+
+    def test_ocsf_metadata_filters_attributes(self):
+        """Attributes with output_formats.ocsf=False should be excluded from unmapped."""
+        metadata = [
+            AttributeMetadata(
+                key="Section",
+                type="str",
+                output_formats=OutputFormats(ocsf=True),
+            ),
+            AttributeMetadata(
+                key="InternalNote",
+                type="str",
+                output_formats=OutputFormats(ocsf=False),
+            ),
+            AttributeMetadata(
+                key="Profile",
+                type="str",
+                output_formats=OutputFormats(ocsf=True),
+            ),
+        ]
+        req = UniversalComplianceRequirement(
+            id="REQ-1",
+            description="Test",
+            attributes={
+                "Section": "IAM",
+                "InternalNote": "skip me",
+                "Profile": "Level 1",
+            },
+            checks={"aws": ["check_a"]},
+        )
+        fw = _make_framework([req], attrs_metadata=metadata)
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        attrs = output.data[0].unmapped["requirement_attributes"]
+        assert "section" in attrs
+        assert "profile" in attrs
+        assert "internal_note" not in attrs
+
+    def test_ocsf_metadata_all_false_excludes_all(self):
+        """When all attributes have output_formats.ocsf=False, requirement_attributes should be empty."""
+        metadata = [
+            AttributeMetadata(
+                key="Section", type="str", output_formats=OutputFormats(ocsf=False)
+            ),
+        ]
+        req = UniversalComplianceRequirement(
+            id="REQ-1",
+            description="Test",
+            attributes={"Section": "IAM"},
+            checks={"aws": ["check_a"]},
+        )
+        fw = _make_framework([req], attrs_metadata=metadata)
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        cf = output.data[0]
+        assert cf.unmapped is not None
+        # requirement_attributes should not appear since all attrs are filtered out
+        assert "requirement_attributes" not in cf.unmapped
+
+    def test_ocsf_no_metadata_includes_all(self):
+        """Without attributes_metadata, all attributes should be included (backward compat)."""
+        req = UniversalComplianceRequirement(
+            id="REQ-1",
+            description="Test",
+            attributes={"Section": "IAM", "Custom": "value"},
+            checks={"aws": ["check_a"]},
+        )
+        fw = _make_framework([req], attrs_metadata=None)
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        attrs = output.data[0].unmapped["requirement_attributes"]
+        assert "section" in attrs
+        assert "custom" in attrs
+
+    def test_ocsf_default_is_true(self):
+        """output_formats.ocsf defaults to True — attributes are included unless explicitly excluded."""
+        metadata = [
+            AttributeMetadata(key="Section", type="str"),
+            AttributeMetadata(key="Profile", type="str"),
+        ]
+        req = UniversalComplianceRequirement(
+            id="REQ-1",
+            description="Test",
+            attributes={"Section": "IAM", "Profile": "Level 1"},
+            checks={"aws": ["check_a"]},
+        )
+        fw = _make_framework([req], attrs_metadata=metadata)
+        findings = [_make_finding("check_a", "PASS")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        attrs = output.data[0].unmapped["requirement_attributes"]
+        assert "section" in attrs
+        assert "profile" in attrs
+
+    def test_ocsf_filter_on_manual_requirements(self):
+        """OCSF filtering should also apply to manual requirements."""
+        metadata = [
+            AttributeMetadata(
+                key="Section", type="str", output_formats=OutputFormats(ocsf=True)
+            ),
+            AttributeMetadata(
+                key="InternalNote",
+                type="str",
+                output_formats=OutputFormats(ocsf=False),
+            ),
+        ]
+        req = UniversalComplianceRequirement(
+            id="MANUAL-1",
+            description="Manual",
+            attributes={"Section": "Logging", "InternalNote": "hidden"},
+            checks={},
+        )
+        fw = _make_framework([req], attrs_metadata=metadata)
+        findings = [_make_finding("check_a")]
+
+        output = OCSFComplianceOutput(findings=findings, framework=fw, provider="aws")
+
+        cf = output.data[0]
+        assert cf.unmapped["requirement_attributes"]["section"] == "Logging"
+        assert "internal_note" not in cf.unmapped["requirement_attributes"]

tests/lib/outputs/compliance/universal/universal_table_test.py

@@ -0,0 +1,384 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from prowler.lib.check.compliance_models import (
+    ComplianceFramework,
+    OutputsConfig,
+    ScoringConfig,
+    SplitByConfig,
+    TableConfig,
+    TableLabels,
+    UniversalComplianceRequirement,
+)
+from prowler.lib.outputs.compliance.universal.universal_table import (
+    _build_requirement_check_map,
+    _get_group_key,
+    get_universal_table,
+)
+
+
+def _make_finding(check_id, status="PASS", muted=False):
+    """Create a mock finding for table tests."""
+    finding = SimpleNamespace()
+    finding.check_metadata = SimpleNamespace(CheckID=check_id)
+    finding.status = status
+    finding.muted = muted
+    return finding
+
+
+def _make_framework(requirements, table_config, provider="AWS"):
+    return ComplianceFramework(
+        framework="TestFW",
+        name="Test Framework",
+        provider=provider,
+        version="1.0",
+        description="Test",
+        requirements=requirements,
+        outputs=OutputsConfig(table_config=table_config) if table_config else None,
+    )
+
+
+class TestBuildRequirementCheckMap:
+    def test_basic(self):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "IAM"},
+                checks={"aws": ["check_a", "check_b"]},
+            ),
+            UniversalComplianceRequirement(
+                id="1.2",
+                description="test2",
+                attributes={"Section": "IAM"},
+                checks={"aws": ["check_b", "check_c"]},
+            ),
+        ]
+        fw = _make_framework(reqs, TableConfig(group_by="Section"))
+        check_map = _build_requirement_check_map(fw)
+        assert "check_a" in check_map
+        assert len(check_map["check_b"]) == 2
+        assert "check_c" in check_map
+
+    def test_dict_checks_no_provider_filter(self):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "IAM"},
+                checks={"aws": ["check_a"], "azure": ["check_b"]},
+            ),
+        ]
+        fw = _make_framework(reqs, TableConfig(group_by="Section"))
+        check_map = _build_requirement_check_map(fw)
+        assert "check_a" in check_map
+        assert "check_b" in check_map
+
+    def test_dict_checks_filtered_by_provider(self):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "IAM"},
+                checks={"aws": ["check_a"], "azure": ["check_b"]},
+            ),
+        ]
+        fw = _make_framework(reqs, TableConfig(group_by="Section"))
+        check_map = _build_requirement_check_map(fw, provider="aws")
+        assert "check_a" in check_map
+        assert "check_b" not in check_map
+
+    def test_dict_checks_provider_not_present(self):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "IAM"},
+                checks={"aws": ["check_a"], "azure": ["check_b"]},
+            ),
+        ]
+        fw = _make_framework(reqs, TableConfig(group_by="Section"))
+        check_map = _build_requirement_check_map(fw, provider="gcp")
+        assert len(check_map) == 0
+
+
+class TestGetGroupKey:
+    def test_normal_field(self):
+        req = UniversalComplianceRequirement(
+            id="1.1",
+            description="test",
+            attributes={"Section": "IAM"},
+            checks={},
+        )
+        assert _get_group_key(req, "Section") == ["IAM"]
+
+    def test_tactics(self):
+        req = UniversalComplianceRequirement(
+            id="T1190",
+            description="test",
+            attributes={},
+            checks={},
+            tactics=["Initial Access", "Execution"],
+        )
+        assert _get_group_key(req, "_Tactics") == ["Initial Access", "Execution"]
+
+
+class TestGroupedMode:
+    def test_grouped_rendering(self, capsys):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "IAM"},
+                checks={"aws": ["check_a"]},
+            ),
+            UniversalComplianceRequirement(
+                id="2.1",
+                description="test2",
+                attributes={"Section": "Logging"},
+                checks={"aws": ["check_b"]},
+            ),
+        ]
+        tc = TableConfig(group_by="Section")
+        fw = _make_framework(reqs, tc)
+
+        findings = [
+            _make_finding("check_a", "PASS"),
+            _make_finding("check_b", "FAIL"),
+        ]
+        bulk_metadata = {
+            "check_a": MagicMock(Compliance=[]),
+            "check_b": MagicMock(Compliance=[]),
+        }
+
+        get_universal_table(
+            findings,
+            bulk_metadata,
+            "test_fw",
+            "output",
+            "/tmp",
+            False,
+            framework=fw,
+        )
+
+        captured = capsys.readouterr()
+        assert "IAM" in captured.out
+        assert "Logging" in captured.out
+        assert "PASS" in captured.out
+        assert "FAIL" in captured.out
+
+
+class TestSplitMode:
+    def test_split_rendering(self, capsys):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "Storage", "Profile": "Level 1"},
+                checks={"aws": ["check_a"]},
+            ),
+            UniversalComplianceRequirement(
+                id="1.2",
+                description="test2",
+                attributes={"Section": "Storage", "Profile": "Level 2"},
+                checks={"aws": ["check_b"]},
+            ),
+        ]
+        tc = TableConfig(
+            group_by="Section",
+            split_by=SplitByConfig(field="Profile", values=["Level 1", "Level 2"]),
+        )
+        fw = _make_framework(reqs, tc)
+
+        findings = [
+            _make_finding("check_a", "PASS"),
+            _make_finding("check_b", "FAIL"),
+        ]
+        bulk_metadata = {
+            "check_a": MagicMock(Compliance=[]),
+            "check_b": MagicMock(Compliance=[]),
+        }
+
+        get_universal_table(
+            findings,
+            bulk_metadata,
+            "test_fw",
+            "output",
+            "/tmp",
+            False,
+            framework=fw,
+        )
+
+        captured = capsys.readouterr()
+        assert "Storage" in captured.out
+        assert "Level 1" in captured.out
+        assert "Level 2" in captured.out
+
+
+class TestScoredMode:
+    def test_scored_rendering(self, capsys):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "IAM", "LevelOfRisk": 5, "Weight": 100},
+                checks={"aws": ["check_a"]},
+            ),
+            UniversalComplianceRequirement(
+                id="1.2",
+                description="test2",
+                attributes={"Section": "IAM", "LevelOfRisk": 3, "Weight": 50},
+                checks={"aws": ["check_b"]},
+            ),
+        ]
+        tc = TableConfig(
+            group_by="Section",
+            scoring=ScoringConfig(risk_field="LevelOfRisk", weight_field="Weight"),
+        )
+        fw = _make_framework(reqs, tc)
+
+        findings = [
+            _make_finding("check_a", "PASS"),
+            _make_finding("check_b", "FAIL"),
+        ]
+        bulk_metadata = {
+            "check_a": MagicMock(Compliance=[]),
+            "check_b": MagicMock(Compliance=[]),
+        }
+
+        get_universal_table(
+            findings,
+            bulk_metadata,
+            "test_fw",
+            "output",
+            "/tmp",
+            False,
+            framework=fw,
+        )
+
+        captured = capsys.readouterr()
+        assert "IAM" in captured.out
+        assert "Score" in captured.out
+        assert "Threat Score" in captured.out
+
+
+class TestCustomLabels:
+    def test_ens_spanish_labels(self, capsys):
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Marco": "operacional"},
+                checks={"aws": ["check_a"]},
+            ),
+            UniversalComplianceRequirement(
+                id="1.2",
+                description="test2",
+                attributes={"Marco": "organizativo"},
+                checks={"aws": ["check_b"]},
+            ),
+        ]
+        tc = TableConfig(
+            group_by="Marco",
+            labels=TableLabels(
+                pass_label="CUMPLE",
+                fail_label="NO CUMPLE",
+                provider_header="Proveedor",
+                title="Estado de Cumplimiento",
+            ),
+        )
+        fw = _make_framework(reqs, tc)
+
+        findings = [_make_finding("check_a", "PASS"), _make_finding("check_b", "FAIL")]
+        bulk_metadata = {
+            "check_a": MagicMock(Compliance=[]),
+            "check_b": MagicMock(Compliance=[]),
+        }
+
+        get_universal_table(
+            findings,
+            bulk_metadata,
+            "test_fw",
+            "output",
+            "/tmp",
+            False,
+            framework=fw,
+        )
+
+        captured = capsys.readouterr()
+        assert "CUMPLE" in captured.out
+        assert "Estado de Cumplimiento" in captured.out
+
+
+class TestMultiProviderDictChecks:
+    def test_only_aws_checks_matched(self, capsys):
+        """With dict checks and provider='aws', only AWS checks match findings."""
+        reqs = [
+            UniversalComplianceRequirement(
+                id="1.1",
+                description="test",
+                attributes={"Section": "IAM"},
+                checks={"aws": ["check_a"], "azure": ["check_b"]},
+            ),
+            UniversalComplianceRequirement(
+                id="2.1",
+                description="test2",
+                attributes={"Section": "Logging"},
+                checks={"aws": ["check_c"], "gcp": ["check_d"]},
+            ),
+        ]
+        tc = TableConfig(group_by="Section")
+        fw = ComplianceFramework(
+            framework="MultiCloud",
+            name="Multi",
+            description="Test",
+            requirements=reqs,
+            outputs=OutputsConfig(table_config=tc),
+        )
+
+        findings = [
+            _make_finding("check_a", "PASS"),
+            _make_finding("check_b", "FAIL"),  # Azure check, should be ignored
+            _make_finding("check_c", "PASS"),
+        ]
+        bulk_metadata = {
+            "check_a": MagicMock(Compliance=[]),
+            "check_b": MagicMock(Compliance=[]),
+            "check_c": MagicMock(Compliance=[]),
+        }
+
+        get_universal_table(
+            findings,
+            bulk_metadata,
+            "multi_cloud",
+            "output",
+            "/tmp",
+            False,
+            framework=fw,
+            provider="aws",
+        )
+
+        captured = capsys.readouterr()
+        assert "IAM" in captured.out
+        assert "Logging" in captured.out
+        # check_b (azure) should not have been counted as FAIL for AWS
+        assert "PASS" in captured.out
+
+
+class TestNoTableConfig:
+    def test_returns_early_without_table_config(self, capsys):
+        fw = ComplianceFramework(
+            framework="TestFW",
+            name="Test",
+            provider="AWS",
+            description="Test",
+            requirements=[],
+        )
+        get_universal_table([], {}, "test", "out", "/tmp", False, framework=fw)
+        captured = capsys.readouterr()
+        assert captured.out == ""
+
+    def test_returns_early_without_framework(self, capsys):
+        get_universal_table([], {}, "test", "out", "/tmp", False, framework=None)
+        captured = capsys.readouterr()
+        assert captured.out == ""

tests/lib/outputs/sarif/sarif_test.py

@@ -0,0 +1,312 @@
+import json
+import os
+import tempfile
+
+import pytest
+
+from prowler.lib.outputs.sarif.sarif import SARIF, SARIF_SCHEMA_URL, SARIF_VERSION
+from tests.lib.outputs.fixtures.fixtures import generate_finding_output
+
+
+class TestSARIF:
+    def test_transform_fail_finding(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            status_extended="S3 bucket is not encrypted",
+            severity="high",
+            resource_name="main.tf",
+            service_name="s3",
+            check_id="s3_encryption_check",
+            check_title="S3 Bucket Encryption",
+        )
+        sarif = SARIF(findings=[finding], file_path=None)
+
+        assert sarif.data[0]["$schema"] == SARIF_SCHEMA_URL
+        assert sarif.data[0]["version"] == SARIF_VERSION
+        assert len(sarif.data[0]["runs"]) == 1
+
+        run = sarif.data[0]["runs"][0]
+        assert run["tool"]["driver"]["name"] == "Prowler"
+        assert len(run["tool"]["driver"]["rules"]) == 1
+        assert len(run["results"]) == 1
+
+        rule = run["tool"]["driver"]["rules"][0]
+        assert rule["id"] == "s3_encryption_check"
+        assert rule["shortDescription"]["text"] == "S3 Bucket Encryption"
+        assert rule["defaultConfiguration"]["level"] == "error"
+        assert rule["properties"]["security-severity"] == "7.0"
+
+        result = run["results"][0]
+        assert result["ruleId"] == "s3_encryption_check"
+        assert result["ruleIndex"] == 0
+        assert result["level"] == "error"
+        assert result["message"]["text"] == "S3 bucket is not encrypted"
+
+    def test_transform_pass_finding_excluded(self):
+        finding = generate_finding_output(status="PASS", severity="high")
+        sarif = SARIF(findings=[finding], file_path=None)
+
+        run = sarif.data[0]["runs"][0]
+        assert len(run["results"]) == 0
+        assert len(run["tool"]["driver"]["rules"]) == 0
+
+    def test_transform_muted_finding_excluded(self):
+        finding = generate_finding_output(status="FAIL", severity="high", muted=True)
+        sarif = SARIF(findings=[finding], file_path=None)
+        run = sarif.data[0]["runs"][0]
+        assert len(run["results"]) == 0
+        assert len(run["tool"]["driver"]["rules"]) == 0
+
+    @pytest.mark.parametrize(
+        "severity,expected_level,expected_security_severity",
+        [
+            ("critical", "error", "9.0"),
+            ("high", "error", "7.0"),
+            ("medium", "warning", "4.0"),
+            ("low", "note", "2.0"),
+            ("informational", "note", "0.0"),
+        ],
+    )
+    def test_transform_severity_mapping(
+        self, severity, expected_level, expected_security_severity
+    ):
+        finding = generate_finding_output(
+            status="FAIL",
+            severity=severity,
+        )
+        sarif = SARIF(findings=[finding], file_path=None)
+
+        run = sarif.data[0]["runs"][0]
+        result = run["results"][0]
+        rule = run["tool"]["driver"]["rules"][0]
+
+        assert result["level"] == expected_level
+        assert rule["defaultConfiguration"]["level"] == expected_level
+        assert rule["properties"]["security-severity"] == expected_security_severity
+
+    def test_transform_multiple_findings_dedup_rules(self):
+        findings = [
+            generate_finding_output(
+                status="FAIL",
+                resource_name="file1.tf",
+                status_extended="Finding in file1",
+            ),
+            generate_finding_output(
+                status="FAIL",
+                resource_name="file2.tf",
+                status_extended="Finding in file2",
+            ),
+        ]
+        sarif = SARIF(findings=findings, file_path=None)
+
+        run = sarif.data[0]["runs"][0]
+        assert len(run["tool"]["driver"]["rules"]) == 1
+        assert len(run["results"]) == 2
+        assert run["results"][0]["ruleIndex"] == 0
+        assert run["results"][1]["ruleIndex"] == 0
+
+    def test_transform_multiple_different_rules(self):
+        findings = [
+            generate_finding_output(
+                status="FAIL",
+                service_name="alpha",
+                check_id="alpha_check_one",
+                status_extended="Finding A",
+            ),
+            generate_finding_output(
+                status="FAIL",
+                service_name="beta",
+                check_id="beta_check_two",
+                status_extended="Finding B",
+            ),
+        ]
+        sarif = SARIF(findings=findings, file_path=None)
+
+        run = sarif.data[0]["runs"][0]
+        assert len(run["tool"]["driver"]["rules"]) == 2
+        assert run["results"][0]["ruleIndex"] == 0
+        assert run["results"][1]["ruleIndex"] == 1
+
+    def test_transform_location_with_line_range(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            resource_name="modules/s3/main.tf",
+        )
+        finding.raw = {"resource_line_range": "10:25"}
+
+        sarif = SARIF(findings=[finding], file_path=None)
+
+        result = sarif.data[0]["runs"][0]["results"][0]
+        location = result["locations"][0]["physicalLocation"]
+        assert location["artifactLocation"]["uri"] == "modules/s3/main.tf"
+        assert location["region"]["startLine"] == 10
+        assert location["region"]["endLine"] == 25
+
+    def test_transform_location_without_line_range(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            resource_name="main.tf",
+        )
+        sarif = SARIF(findings=[finding], file_path=None)
+
+        result = sarif.data[0]["runs"][0]["results"][0]
+        location = result["locations"][0]["physicalLocation"]
+        assert location["artifactLocation"]["uri"] == "main.tf"
+        assert "region" not in location
+
+    def test_transform_no_resource_name(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            resource_name="",
+        )
+        sarif = SARIF(findings=[finding], file_path=None)
+
+        result = sarif.data[0]["runs"][0]["results"][0]
+        assert "locations" not in result
+
+    def test_batch_write_data_to_file(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            status_extended="test finding",
+            resource_name="main.tf",
+        )
+
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".sarif", delete=False
+        ) as tmp:
+            tmp_path = tmp.name
+
+        sarif = SARIF(
+            findings=[finding],
+            file_path=tmp_path,
+        )
+        sarif.batch_write_data_to_file()
+
+        with open(tmp_path) as f:
+            content = json.load(f)
+
+        assert content["$schema"] == SARIF_SCHEMA_URL
+        assert content["version"] == SARIF_VERSION
+        assert len(content["runs"][0]["results"]) == 1
+
+        os.unlink(tmp_path)
+
+    def test_sarif_schema_structure(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            severity="critical",
+            resource_name="infra/main.tf",
+            service_name="iac",
+            check_id="iac_misconfig_check",
+            check_title="IaC Misconfiguration",
+            description="Checks for misconfigurations",
+            remediation_recommendation_text="Fix the configuration",
+        )
+        finding.raw = {"resource_line_range": "5:15"}
+
+        sarif = SARIF(findings=[finding], file_path=None)
+        doc = sarif.data[0]
+
+        assert "$schema" in doc
+        assert "version" in doc
+        assert "runs" in doc
+
+        run = doc["runs"][0]
+
+        assert "tool" in run
+        assert "driver" in run["tool"]
+        driver = run["tool"]["driver"]
+        assert "name" in driver
+        assert "version" in driver
+        assert "informationUri" in driver
+        assert "rules" in driver
+
+        rule = driver["rules"][0]
+        assert "id" in rule
+        assert "shortDescription" in rule
+        assert "fullDescription" in rule
+        assert "help" in rule
+        assert "defaultConfiguration" in rule
+        assert "properties" in rule
+        assert "tags" in rule["properties"]
+        assert "security-severity" in rule["properties"]
+
+        result = run["results"][0]
+        assert "ruleId" in result
+        assert "ruleIndex" in result
+        assert "level" in result
+        assert "message" in result
+        assert "locations" in result
+
+        loc = result["locations"][0]["physicalLocation"]
+        assert "artifactLocation" in loc
+        assert "uri" in loc["artifactLocation"]
+        assert "region" in loc
+        assert "startLine" in loc["region"]
+        assert "endLine" in loc["region"]
+
+    def test_transform_helpuri_present_when_related_url_set(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            provider="iac",
+            related_url="https://docs.example.com/check",
+        )
+        sarif = SARIF(findings=[finding], file_path=None)
+        rule = sarif.data[0]["runs"][0]["tool"]["driver"]["rules"][0]
+        assert rule["helpUri"] == "https://docs.example.com/check"
+
+    def test_transform_helpuri_absent_when_related_url_empty(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            related_url="",
+        )
+        sarif = SARIF(findings=[finding], file_path=None)
+        rule = sarif.data[0]["runs"][0]["tool"]["driver"]["rules"][0]
+        assert "helpUri" not in rule
+
+    def test_location_with_non_numeric_line_range(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            resource_name="main.tf",
+        )
+        finding.raw = {"resource_line_range": "abc:def"}
+        sarif = SARIF(findings=[finding], file_path=None)
+        location = sarif.data[0]["runs"][0]["results"][0]["locations"][0][
+            "physicalLocation"
+        ]
+        assert "region" not in location
+
+    def test_location_with_single_value_line_range(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            resource_name="main.tf",
+        )
+        finding.raw = {"resource_line_range": "10"}
+        sarif = SARIF(findings=[finding], file_path=None)
+        location = sarif.data[0]["runs"][0]["results"][0]["locations"][0][
+            "physicalLocation"
+        ]
+        assert "region" not in location
+
+    def test_location_with_zero_line_numbers(self):
+        finding = generate_finding_output(
+            status="FAIL",
+            resource_name="main.tf",
+        )
+        finding.raw = {"resource_line_range": "0:0"}
+        sarif = SARIF(findings=[finding], file_path=None)
+        location = sarif.data[0]["runs"][0]["results"][0]["locations"][0][
+            "physicalLocation"
+        ]
+        assert "region" not in location
+
+    def test_only_pass_findings(self):
+        findings = [
+            generate_finding_output(status="PASS"),
+            generate_finding_output(status="PASS"),
+        ]
+        sarif = SARIF(findings=findings, file_path=None)
+
+        run = sarif.data[0]["runs"][0]
+        assert len(run["results"]) == 0
+        assert len(run["tool"]["driver"]["rules"]) == 0

tests/providers/common/common_arguments_test.py

@@ -0,0 +1,53 @@
+from prowler.providers.common.arguments import (
+    validate_asff_usage,
+    validate_sarif_usage,
+)
+
+
+class TestValidateAsffUsage:
+    def test_asff_with_aws_provider(self):
+        valid, msg = validate_asff_usage("aws", ["json-asff"])
+        assert valid is True
+        assert msg == ""
+
+    def test_asff_with_non_aws_provider(self):
+        valid, msg = validate_asff_usage("gcp", ["json-asff"])
+        assert valid is False
+        assert "aws" in msg
+
+    def test_no_asff_in_formats(self):
+        valid, msg = validate_asff_usage("gcp", ["csv", "html"])
+        assert valid is True
+
+    def test_no_output_formats(self):
+        valid, msg = validate_asff_usage("aws", None)
+        assert valid is True
+
+
+class TestValidateSarifUsage:
+    def test_sarif_with_iac_provider(self):
+        valid, msg = validate_sarif_usage("iac", ["sarif"])
+        assert valid is True
+        assert msg == ""
+
+    def test_sarif_with_non_iac_provider(self):
+        valid, msg = validate_sarif_usage("aws", ["sarif"])
+        assert valid is False
+        assert "iac" in msg
+
+    def test_sarif_with_other_provider(self):
+        valid, msg = validate_sarif_usage("gcp", ["csv", "sarif"])
+        assert valid is False
+        assert "gcp" in msg
+
+    def test_no_sarif_in_formats(self):
+        valid, msg = validate_sarif_usage("aws", ["csv", "html"])
+        assert valid is True
+
+    def test_no_output_formats(self):
+        valid, msg = validate_sarif_usage("iac", None)
+        assert valid is True
+
+    def test_empty_output_formats(self):
+        valid, msg = validate_sarif_usage("aws", [])
+        assert valid is True

ui/CHANGELOG.md

@@ -6,6 +6,10 @@ All notable changes to the **Prowler UI** are documented in this file.
 
 ### 🔄 Changed
 
+- Allows tenant owners to expel users from their organizations  [(#10787)](https://github.com/prowler-cloud/prowler/pull/10787)
+
+### ❌ Removed
+
 - Redesign compliance page with a horizontal ThreatScore card (always-visible pillar breakdown + ActionDropdown), client-side search for compliance frameworks, compact scan selector trigger, responsive mobile filters, download-started toasts for CSV/PDF exports, enhanced compliance cards with truncated titles, and Alert-based empty/error states; migrate Progress component from HeroUI to shadcn [(#10767)](https://github.com/prowler-cloud/prowler/pull/10767)
 - Backward-compatibility middleware redirect from `/sign-up?invitation_token=…` to `/invitation/accept?invitation_token=…`; new invitation emails use `/invitation/accept` directly [(#10797)](https://github.com/prowler-cloud/prowler/pull/10797)
 

ui/actions/users/users.ts

@@ -2,17 +2,64 @@
 
 import { revalidatePath } from "next/cache";
 import { redirect } from "next/navigation";
+import { z } from "zod";
 
+import { auth } from "@/auth.config";
 import { apiBaseUrl, getAuthHeaders } from "@/lib";
 import { handleApiError, handleApiResponse } from "@/lib/server-actions-helper";
+import {
+  TENANT_MEMBERSHIP_ROLE,
+  type TenantMembershipRole,
+} from "@/types/users";
+
+const getUsersSchema = z.object({
+  page: z.coerce.number().int().min(1).default(1),
+  query: z.string().default(""),
+  sort: z.string().optional().default(""),
+  filters: z
+    .record(
+      z.string(),
+      z.union([z.string(), z.array(z.string()), z.number()]).optional(),
+    )
+    .default({}),
+  pageSize: z.coerce.number().int().min(1).default(10),
+});
+
+const updateUserSchema = z.object({
+  userId: z.uuid(),
+  name: z.string().min(1).optional(),
+  email: z.email().optional(),
+  company_name: z.string().optional(),
+  password: z.string().min(1).optional(),
+});
+
+const deleteUserSchema = z.object({
+  userId: z.uuid(),
+});
+
+const removeUserFromTenantSchema = z.object({
+  userId: z.uuid(),
+  tenantId: z.uuid(),
+});
+
+const updateUserRoleSchema = z.object({
+  userId: z.uuid(),
+  roleId: z.uuid(),
+});
+
+type GetUsersInput = z.input<typeof getUsersSchema>;
+type UpdateUserData = z.infer<typeof updateUserSchema>;
+type UserAttributes = Omit<UpdateUserData, "userId">;
+type MembershipResource = { id: string };
+
+export const getUsers = async (rawParams: Partial<GetUsersInput> = {}) => {
+  const parsed = getUsersSchema.safeParse(rawParams);
+  if (!parsed.success) {
+    console.error("Invalid getUsers params:", parsed.error.flatten());
+    return undefined;
+  }
+  const { page, query, sort, filters, pageSize } = parsed.data;
 
-export const getUsers = async ({
-  page = 1,
-  query = "",
-  sort = "",
-  filters = {},
-  pageSize = 10,
-}) => {
   const headers = await getAuthHeaders({ contentType: false });
 
   if (isNaN(Number(page)) || page < 1) redirect("/users?include=roles");
@@ -46,22 +93,29 @@ export const getUsers = async ({
 export const updateUser = async (formData: FormData) => {
   const headers = await getAuthHeaders({ contentType: true });
 
-  const userId = formData.get("userId") as string; // Ensure userId is a string
-  const userName = formData.get("name") as string | null;
-  const userPassword = formData.get("password") as string | null;
-  const userEmail = formData.get("email") as string | null;
-  const userCompanyName = formData.get("company_name") as string | null;
+  const rawData = {
+    userId: formData.get("userId"),
+    name: formData.get("name") ?? undefined,
+    email: formData.get("email") ?? undefined,
+    company_name: formData.get("company_name") ?? undefined,
+    password: formData.get("password") ?? undefined,
+  };
+  const parsed = updateUserSchema.safeParse(rawData);
+  if (!parsed.success) {
+    return { error: "Invalid user data" };
+  }
+  const { userId, name, email, company_name, password } = parsed.data;
 
   const url = new URL(`${apiBaseUrl}/users/${userId}`);
 
   // Prepare attributes to send based on changes
-  const attributes: Record<string, any> = {};
+  const attributes: UserAttributes = {};
 
   // Add only changed fields
-  if (userName !== null) attributes.name = userName;
-  if (userEmail !== null) attributes.email = userEmail;
-  if (userCompanyName !== null) attributes.company_name = userCompanyName;
-  if (userPassword !== null) attributes.password = userPassword;
+  if (name !== undefined) attributes.name = name;
+  if (email !== undefined) attributes.email = email;
+  if (company_name !== undefined) attributes.company_name = company_name;
+  if (password !== undefined) attributes.password = password;
 
   // If no fields have changed, don't send the request
   if (Object.keys(attributes).length === 0) {
@@ -90,13 +144,14 @@ export const updateUser = async (formData: FormData) => {
 export const updateUserRole = async (formData: FormData) => {
   const headers = await getAuthHeaders({ contentType: true });
 
-  const userId = formData.get("userId") as string;
-  const roleId = formData.get("roleId") as string;
-
-  // Validate required fields
-  if (!userId || !roleId) {
+  const parsed = updateUserRoleSchema.safeParse({
+    userId: formData.get("userId"),
+    roleId: formData.get("roleId"),
+  });
+  if (!parsed.success) {
     return { error: "userId and roleId are required" };
   }
+  const { userId, roleId } = parsed.data;
 
   const url = new URL(`${apiBaseUrl}/users/${userId}/relationships/roles`);
 
@@ -124,11 +179,11 @@ export const updateUserRole = async (formData: FormData) => {
 
 export const deleteUser = async (formData: FormData) => {
   const headers = await getAuthHeaders({ contentType: false });
-  const userId = formData.get("userId");
-
-  if (!userId) {
+  const parsed = deleteUserSchema.safeParse({ userId: formData.get("userId") });
+  if (!parsed.success) {
     return { error: "User ID is required" };
   }
+  const { userId } = parsed.data;
 
   const url = new URL(`${apiBaseUrl}/users/${userId}`);
 
@@ -158,6 +213,156 @@ export const deleteUser = async (formData: FormData) => {
   }
 };
 
+interface ServerActionErrorDetail {
+  detail: string;
+  code?: string;
+}
+
+interface ServerActionErrorResponse {
+  errors: ServerActionErrorDetail[];
+}
+
+interface RemoveUserFromTenantSuccess {
+  success: true;
+}
+
+type RemoveUserFromTenantResult =
+  | RemoveUserFromTenantSuccess
+  | ServerActionErrorResponse;
+
+const toErrorResponse = (detail: string): ServerActionErrorResponse => ({
+  errors: [{ detail }],
+});
+
+export const removeUserFromTenant = async (
+  formData: FormData,
+): Promise<RemoveUserFromTenantResult> => {
+  const headers = await getAuthHeaders({ contentType: false });
+  const parsed = removeUserFromTenantSchema.safeParse({
+    userId: formData.get("userId"),
+    tenantId: formData.get("tenantId"),
+  });
+  if (!parsed.success) {
+    return toErrorResponse("userId and tenantId are required");
+  }
+  const { userId, tenantId } = parsed.data;
+
+  // Resolve the target user's membership id for the current tenant on the
+  // server so the client form can open instantly without a prefetch.
+  //
+  // We cannot use `/users/{userId}/memberships` here: that endpoint ignores
+  // the path user id and always returns the authenticated user's memberships,
+  // which would make us try to delete the caller's own membership.
+  const listUrl = new URL(`${apiBaseUrl}/tenants/${tenantId}/memberships`);
+  listUrl.searchParams.append("filter[user]", userId);
+  listUrl.searchParams.append("page[size]", "1");
+
+  let targetMembershipId: string | null = null;
+  try {
+    const listResponse = await fetch(listUrl.toString(), { headers });
+    if (!listResponse.ok) {
+      const errorData = await listResponse.json().catch(() => ({}));
+      return {
+        errors: errorData.errors ?? [
+          { detail: "Failed to resolve the user's membership" },
+        ],
+      };
+    }
+    const listData = (await listResponse.json()) as {
+      data?: MembershipResource[];
+    };
+    targetMembershipId = listData?.data?.[0]?.id ?? null;
+  } catch (error) {
+    const handled = handleApiError(error);
+    return toErrorResponse(
+      handled?.error ?? "Failed to resolve the user's membership",
+    );
+  }
+
+  if (!targetMembershipId) {
+    return toErrorResponse(
+      "This user is not a member of the current organization.",
+    );
+  }
+
+  const url = new URL(
+    `${apiBaseUrl}/tenants/${tenantId}/memberships/${targetMembershipId}`,
+  );
+
+  try {
+    const response = await fetch(url.toString(), {
+      method: "DELETE",
+      headers,
+    });
+
+    if (!response.ok) {
+      const errorData = await response.json().catch(() => ({}));
+      return {
+        errors: errorData.errors ?? [
+          { detail: "Failed to expel the user from the organization" },
+        ],
+      };
+    }
+
+    revalidatePath("/users");
+    return { success: true };
+  } catch (error) {
+    const handled = handleApiError(error);
+    return toErrorResponse(
+      handled?.error ?? "Failed to expel the user from the organization",
+    );
+  }
+};
+
+interface MembershipAttributesResource {
+  id: string;
+  attributes?: {
+    role?: string;
+  };
+}
+
+/**
+ * Resolve the active user's role inside the current tenant by querying the
+ * tenant memberships list with `filter[user]`. Returns `null` if the role
+ * cannot be determined (missing session, API error, or no match), so the
+ * caller can default-deny the destructive UI action.
+ */
+export const getCurrentUserTenantRole =
+  async (): Promise<TenantMembershipRole | null> => {
+    const session = await auth();
+    const userId = session?.userId;
+    const tenantId = session?.tenantId;
+    if (!userId || !tenantId) {
+      return null;
+    }
+
+    const headers = await getAuthHeaders({ contentType: false });
+    const url = new URL(`${apiBaseUrl}/tenants/${tenantId}/memberships`);
+    url.searchParams.append("filter[user]", userId);
+    url.searchParams.append("page[size]", "1");
+
+    try {
+      const response = await fetch(url.toString(), { headers });
+      if (!response.ok) {
+        return null;
+      }
+      const body = (await response.json()) as {
+        data?: MembershipAttributesResource[];
+      };
+      const role = body?.data?.[0]?.attributes?.role;
+      if (
+        role === TENANT_MEMBERSHIP_ROLE.Owner ||
+        role === TENANT_MEMBERSHIP_ROLE.Member
+      ) {
+        return role;
+      }
+      return null;
+    } catch (error) {
+      console.error("Error resolving current user's tenant role:", error);
+      return null;
+    }
+  };
+
 export const getUserInfo = async () => {
   const headers = await getAuthHeaders({ contentType: false });
   const url = new URL(

ui/app/(prowler)/users/page.tsx

@@ -2,7 +2,8 @@ import Link from "next/link";
 import { Suspense } from "react";
 
 import { getRoles } from "@/actions/roles/roles";
-import { getUsers } from "@/actions/users/users";
+import { getCurrentUserTenantRole, getUsers } from "@/actions/users/users";
+import { auth } from "@/auth.config";
 import { FilterControls } from "@/components/filters";
 import { AddIcon } from "@/components/icons";
 import { Button } from "@/components/shadcn";
@@ -10,6 +11,7 @@ import { ContentLayout } from "@/components/ui";
 import { DataTable } from "@/components/ui/table";
 import { ColumnsUser, SkeletonTableUser } from "@/components/users/table";
 import { Role, SearchParamsProps, UserProps } from "@/types";
+import { TENANT_MEMBERSHIP_ROLE } from "@/types/users";
 
 export default async function Users({
   searchParams,
@@ -58,19 +60,24 @@ const SSRDataTable = async ({
   // Extract query from filters
   const query = (filters["filter[search]"] as string) || "";
 
-  const usersData = await getUsers({ query, page, sort, filters, pageSize });
-  const rolesData = await getRoles({});
+  const [usersData, rolesData, currentTenantRole, session] = await Promise.all([
+    getUsers({ query, page, sort, filters, pageSize }),
+    getRoles({}),
+    getCurrentUserTenantRole(),
+    auth(),
+  ]);
+
+  const currentUserId = session?.userId;
+  const currentTenantId = session?.tenantId;
+  const isCurrentUserOwner = currentTenantRole === TENANT_MEMBERSHIP_ROLE.Owner;
 
   // Create a dictionary for roles by user ID
-  const roleDict = (usersData?.included || []).reduce(
-    (acc: Record<string, any>, item: Role) => {
-      if (item.type === "roles") {
-        acc[item.id] = item.attributes;
-      }
-      return acc;
-    },
-    {} as Record<string, Role>,
-  );
+  const roleDict: Record<string, Role["attributes"]> = {};
+  for (const item of (usersData?.included || []) as Role[]) {
+    if (item.type === "roles") {
+      roleDict[item.id] = item.attributes;
+    }
+  }
 
   // Generate the array of roles with all the roles available
   const roles = Array.from(
@@ -88,6 +95,11 @@ const SSRDataTable = async ({
     const roleId = user?.relationships?.roles?.data?.[0]?.id;
     const role = roleDict?.[roleId] || null;
 
+    // Gate the "Expel" action server-side: only tenant owners may expel,
+    // and never against themselves (self-leave lives elsewhere).
+    const canBeExpelled =
+      isCurrentUserOwner && !!currentTenantId && user.id !== currentUserId;
+
     return {
       ...user,
       attributes: {
@@ -95,6 +107,8 @@ const SSRDataTable = async ({
         role,
       },
       roles,
+      canBeExpelled,
+      currentTenantId: canBeExpelled ? currentTenantId : undefined,
     };
   });
 

ui/auth.config.ts

@@ -15,7 +15,8 @@ import { apiBaseUrl } from "./lib";
 import type { RolePermissionAttributes } from "./types/users";
 
 interface CustomJwtPayload extends JwtPayload {
-  user_id: string;
+  user_id?: string; // Optional - doesn't actually exist in JWT tokens
+  sub: string; // Standard JWT subject field - contains the actual user ID
   tenant_id: string;
 }
 
@@ -90,7 +91,8 @@ const applyDecodedClaims = (
     target.accessTokenExpires = decodedToken.exp
       ? decodedToken.exp * 1000
       : target.accessTokenExpires;
-    target.user_id = decodedToken.user_id ?? target.user_id;
+    // Map standard JWT "sub" field to user_id
+    target.user_id = decodedToken.sub ?? target.user_id;
     target.tenant_id = decodedToken.tenant_id ?? target.tenant_id;
   } catch (decodeError) {
     // eslint-disable-next-line no-console

ui/components/users/forms/expel-user-form.tsx

@@ -0,0 +1,88 @@
+"use client";
+
+import { Dispatch, SetStateAction, useTransition } from "react";
+
+import { removeUserFromTenant } from "@/actions/users/users";
+import { DeleteIcon } from "@/components/icons";
+import { Button } from "@/components/shadcn";
+import { useToast } from "@/components/ui";
+
+interface ExpelUserFormProps {
+  userId: string;
+  userName?: string;
+  tenantId: string;
+  setIsOpen: Dispatch<SetStateAction<boolean>>;
+}
+
+export const ExpelUserForm = ({
+  userId,
+  userName,
+  tenantId,
+  setIsOpen,
+}: ExpelUserFormProps) => {
+  const { toast } = useToast();
+  const [isPending, startTransition] = useTransition();
+
+  const handleExpel = () => {
+    startTransition(async () => {
+      const formData = new FormData();
+      formData.append("userId", userId);
+      formData.append("tenantId", tenantId);
+
+      const data = await removeUserFromTenant(formData);
+
+      if (!data || !("success" in data) || data.success !== true) {
+        const detail =
+          data && "errors" in data && data.errors?.[0]?.detail
+            ? data.errors[0].detail
+            : "Failed to expel the user";
+        toast({
+          variant: "destructive",
+          title: "Oops! Something went wrong",
+          description: detail,
+        });
+        return;
+      }
+
+      toast({
+        title: "User expelled",
+        description: `${userName ?? "The user"} was removed from this organization.`,
+      });
+      setIsOpen(false);
+    });
+  };
+
+  const displayName = userName ?? "this user";
+
+  return (
+    <div className="flex flex-col gap-4">
+      <p className="text-sm">
+        <span className="font-semibold">{displayName}</span> will lose access to
+        this organization. If they don&apos;t belong to any other organization,
+        their account will be permanently deleted.
+      </p>
+
+      <div className="flex w-full justify-end gap-4">
+        <Button
+          type="button"
+          variant="ghost"
+          size="lg"
+          onClick={() => setIsOpen(false)}
+          disabled={isPending}
+        >
+          Cancel
+        </Button>
+        <Button
+          type="button"
+          variant="destructive"
+          size="lg"
+          onClick={handleExpel}
+          disabled={isPending}
+        >
+          {!isPending && <DeleteIcon size={24} aria-hidden="true" />}
+          {isPending ? "Expelling…" : "Expel user"}
+        </Button>
+      </div>
+    </div>
+  );
+};

ui/components/users/forms/index.ts

@@ -1,3 +1,4 @@
 export * from "./delete-form";
 export * from "./edit-form";
 export * from "./edit-tenant-form";
+export * from "./expel-user-form";

ui/components/users/table/data-table-row-actions.tsx

@@ -1,7 +1,7 @@
 "use client";
 
 import { Row } from "@tanstack/react-table";
-import { Pencil, Trash2 } from "lucide-react";
+import { Pencil, Trash2, UserMinus } from "lucide-react";
 import { useState } from "react";
 
 import {
@@ -11,24 +11,51 @@ import {
 } from "@/components/shadcn/dropdown";
 import { Modal } from "@/components/shadcn/modal";
 
-import { DeleteForm, EditForm } from "../forms";
+import { DeleteForm, EditForm, ExpelUserForm } from "../forms";
 
-interface DataTableRowActionsProps<UserProps> {
+interface UserRowRole {
+  name?: string;
+}
+
+interface UserRowAttributes {
+  name?: string;
+  email?: string;
+  company_name?: string;
+  role?: UserRowRole;
+}
+
+interface UserRowData {
+  id: string;
+  attributes?: UserRowAttributes;
+  canBeExpelled?: boolean;
+  currentTenantId?: string;
+}
+
+interface DataTableRowActionsProps<UserProps extends UserRowData> {
   row: Row<UserProps>;
   roles?: { id: string; name: string }[];
 }
 
-export function DataTableRowActions<UserProps>({
+export function DataTableRowActions<UserProps extends UserRowData>({
   row,
   roles,
 }: DataTableRowActionsProps<UserProps>) {
   const [isEditOpen, setIsEditOpen] = useState(false);
   const [isDeleteOpen, setIsDeleteOpen] = useState(false);
-  const userId = (row.original as { id: string }).id;
-  const userName = (row.original as any).attributes?.name;
-  const userEmail = (row.original as any).attributes?.email;
-  const userCompanyName = (row.original as any).attributes?.company_name;
-  const userRole = (row.original as any).attributes?.role?.name;
+  const [isExpelOpen, setIsExpelOpen] = useState(false);
+  const userId = row.original.id;
+  const userName = row.original.attributes?.name;
+  const userEmail = row.original.attributes?.email;
+  const userCompanyName = row.original.attributes?.company_name;
+  const userRole = row.original.attributes?.role?.name;
+
+  // Expel gate is resolved server-side against the active tenant's membership
+  // role (owner vs member), mirroring the backend rule in
+  // TenantMembersViewSet.destroy. The row is only expel-eligible when the
+  // current user is an owner of the active tenant and the row is not theirs.
+  const canExpelUser =
+    row.original.canBeExpelled === true && !!row.original.currentTenantId;
+  const currentTenantId = row.original.currentTenantId;
 
   return (
     <>
@@ -55,17 +82,39 @@ export function DataTableRowActions<UserProps>({
       >
         <DeleteForm userId={userId} setIsOpen={setIsDeleteOpen} />
       </Modal>
+      {canExpelUser && currentTenantId && (
+        <Modal
+          open={isExpelOpen}
+          onOpenChange={setIsExpelOpen}
+          title="Expel user from this organization"
+        >
+          <ExpelUserForm
+            userId={userId}
+            userName={userName}
+            tenantId={currentTenantId}
+            setIsOpen={setIsExpelOpen}
+          />
+        </Modal>
+      )}
 
       <div className="relative flex items-center justify-end gap-2">
         <ActionDropdown>
           <ActionDropdownItem
-            icon={<Pencil />}
+            icon={<Pencil aria-hidden="true" />}
             label="Edit User"
             onSelect={() => setIsEditOpen(true)}
           />
           <ActionDropdownDangerZone>
+            {canExpelUser && (
+              <ActionDropdownItem
+                icon={<UserMinus aria-hidden="true" />}
+                label="Expel from organization"
+                destructive
+                onSelect={() => setIsExpelOpen(true)}
+              />
+            )}
             <ActionDropdownItem
-              icon={<Trash2 />}
+              icon={<Trash2 aria-hidden="true" />}
               label="Delete User"
               destructive
               onSelect={() => setIsDeleteOpen(true)}

ui/types/users.ts

@@ -70,6 +70,14 @@ export type RolePermissionAttributes = Pick<
   PermissionKey
 >;
 
+export const TENANT_MEMBERSHIP_ROLE = {
+  Owner: "owner",
+  Member: "member",
+} as const;
+
+export type TenantMembershipRole =
+  (typeof TENANT_MEMBERSHIP_ROLE)[keyof typeof TENANT_MEMBERSHIP_ROLE];
+
 export interface RoleDetail {
   id: string;
   type: "roles";