fix(api): safely decode stored Celery task arguments (#12165)

(cherry picked from commit f8be9afa7c)

.env

@@ -158,7 +158,7 @@ SENTRY_RELEASE=local
 # REO_DEV_CLIENT_ID=
 
 #### Prowler release version ####
-NEXT_PUBLIC_PROWLER_RELEASE_VERSION=v5.36.0
+NEXT_PUBLIC_PROWLER_RELEASE_VERSION=v5.36.1
 
 # Social login credentials
 SOCIAL_GOOGLE_OAUTH_CALLBACK_URL="${AUTH_URL}/api/auth/callback/google"

api/changelog.d/12210.fixed.md

@@ -0,0 +1 @@
+Authentication with an API key whose owning user was deleted now returns `401` instead of an unhandled `AttributeError`, and user deletion now revokes the user's API keys across all their tenants

api/changelog.d/celery-task-kwargs-parsing.fixed.md

@@ -0,0 +1 @@
+`task_args` serialization no longer returns HTTP 500 errors when Celery truncates stored task keyword arguments

api/changelog.d/security-hub-connection-status.fixed.md

@@ -0,0 +1 @@
+AWS Security Hub integrations now persist successful connection checks during finding delivery so their connection status and last checked timestamp stay current

api/pyproject.toml

@@ -45,7 +45,7 @@ dependencies = [
   "gunicorn==26.0.0",
   "uvloop==0.22.1",
   "lxml==6.1.0",
-  "prowler @ git+https://github.com/prowler-cloud/prowler.git@master",
+  "prowler @ git+https://github.com/prowler-cloud/prowler.git@v5.36",
   "psycopg2-binary==2.9.9",
   "pytest-celery[redis] (==1.3.0)",
   "sentry-sdk[django] (==2.56.0)",
@@ -71,7 +71,7 @@ name = "prowler-api"
 package-mode = false
 # Needed for the SDK compatibility
 requires-python = ">=3.11,<3.13"
-version = "1.37.0"
+version = "1.37.1"
 
 # Shared ruff baseline (kept in sync with mcp_server/pyproject.toml).
 # target-version tracks this project's lowest supported Python.

api/src/backend/api/authentication.py

@@ -1,3 +1,4 @@
+import logging
 from math import isfinite
 from uuid import UUID
 
@@ -5,6 +6,7 @@ from api.db_router import MainRouter
 from api.models import TenantAPIKey, TenantAPIKeyManager
 from cryptography.fernet import InvalidToken
 from django.core.exceptions import ObjectDoesNotExist
+from django.db import transaction
 from django.utils import timezone
 from drf_simple_apikey.backends import APIKeyAuthentication as BaseAPIKeyAuth
 from drf_simple_apikey.crypto import get_crypto
@@ -14,6 +16,16 @@ from rest_framework.exceptions import AuthenticationFailed
 from rest_framework.request import Request
 from rest_framework_simplejwt.authentication import JWTAuthentication
 
+logger = logging.getLogger(__name__)
+
+
+class OrphanedAPIKeyError(Exception):
+    """Raised when an API key outlived the user that owns it.
+
+    Handled by `authenticate`, which commits the revocation written while detecting it
+    and then rejects the request with `AuthenticationFailed`.
+    """
+
 
 class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
     model = TenantAPIKey
@@ -24,10 +36,13 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
     def _authenticate_credentials(self, request, key):
         """
         Override to use admin connection, bypassing RLS during authentication.
+
+        Returns the validated API key row, locked with `select_for_update`, so callers
+        must run inside `transaction.atomic(using=MainRouter.admin_db)`.
         """
         try:
             payload = self.key_crypto.decrypt(key)
-        except ValueError:
+        except (ValueError, InvalidToken):
             raise AuthenticationFailed("Invalid API Key.")
 
         if not isinstance(payload, dict):
@@ -52,13 +67,33 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
             raise AuthenticationFailed("API Key has already expired.")
 
         try:
-            api_key = self.model.objects.using(MainRouter.admin_db).get(id=api_key_pk)
+            api_key = (
+                self.model.objects.using(MainRouter.admin_db)
+                .select_for_update()
+                .get(id=api_key_pk)
+            )
         except ObjectDoesNotExist:
             raise AuthenticationFailed("No entity matching this api key.")
 
         if api_key.revoked:
             raise AuthenticationFailed("This API Key has been revoked.")
 
+        # `entity` is nullable and `on_delete=SET_NULL` leaves the key behind when its
+        # owner is deleted, so a key can outlive its user. Reject it here: further down
+        # the authentication would return `None` as the authenticated user, which blows
+        # up while building the auth dict and surfaces as a 500 instead of a 401.
+        # Revoke it as well, so it stops showing up as active and later attempts fail
+        # the `revoked` check above like any other revoked key.
+        if api_key.entity_id is None:
+            api_key.revoked = True
+            api_key.save(update_fields=["revoked"], using=MainRouter.admin_db)
+            logger.warning(
+                "Revoked orphaned API key: prefix=%s tenant=%s",
+                api_key.prefix,
+                api_key.tenant_id,
+            )
+            raise OrphanedAPIKeyError
+
         client_ip = request.META.get(package_settings.IP_ADDRESS_HEADER)
         if api_key.blacklisted_ips and client_ip in api_key.blacklisted_ips:
             raise AuthenticationFailed("Access denied from blacklisted IP.")
@@ -66,7 +101,7 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
         if api_key.whitelisted_ips and client_ip not in api_key.whitelisted_ips:
             raise AuthenticationFailed("Access restricted to specific IP addresses.")
 
-        return api_key.entity, key
+        return api_key
 
     def authenticate(self, request: Request):
         prefixed_key = self.get_key(request)
@@ -77,36 +112,34 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
         except ValueError:
             raise AuthenticationFailed("Invalid API Key.")
 
-        try:
-            entity, _ = self._authenticate_credentials(request, key)
-        except InvalidToken:
-            raise AuthenticationFailed("Invalid API Key.")
+        # Validation, the `last_used_at` update and the auth claims all read the same
+        # row, locked until the transaction ends. Looking the key up a second time to
+        # build the claims used to leave a window where a key revoked or orphaned right
+        # after passing validation still authenticated.
+        with transaction.atomic(using=MainRouter.admin_db):
+            try:
+                api_key = self._authenticate_credentials(request, key)
+            except OrphanedAPIKeyError:
+                # Rejected below instead of here: leaving the block normally commits
+                # the revocation `_authenticate_credentials` wrote, while raising from
+                # inside would roll it back.
+                pass
+            else:
+                # The prefix used to be checked by the second lookup
+                if api_key.prefix != prefix:
+                    raise AuthenticationFailed("Invalid API Key.")
 
-        # Get the API key instance to update last_used_at and retrieve tenant info
-        # We need to decrypt again to get the pk (already validated by _authenticate_credentials)
-        payload = self.key_crypto.decrypt(key)
-        api_key_pk = payload["_pk"]
+                api_key.last_used_at = timezone.now()
+                api_key.save(update_fields=["last_used_at"], using=MainRouter.admin_db)
 
-        # Convert string UUID back to UUID object for lookup
-        if isinstance(api_key_pk, str):
-            api_key_pk = UUID(api_key_pk)
+                entity = api_key.entity
+                return entity, {
+                    "tenant_id": str(api_key.tenant_id),
+                    "sub": str(entity.id),
+                    "api_key_prefix": api_key.prefix,
+                }
 
-        try:
-            api_key_instance = TenantAPIKey.objects.using(MainRouter.admin_db).get(
-                id=api_key_pk, prefix=prefix
-            )
-        except TenantAPIKey.DoesNotExist:
-            raise AuthenticationFailed("Invalid API Key.")
-
-        # Update last_used_at
-        api_key_instance.last_used_at = timezone.now()
-        api_key_instance.save(update_fields=["last_used_at"], using=MainRouter.admin_db)
-
-        return entity, {
-            "tenant_id": str(api_key_instance.tenant_id),
-            "sub": str(api_key_instance.entity.id),
-            "api_key_prefix": prefix,
-        }
+        raise AuthenticationFailed("No entity matching this api key.")
 
 
 class CombinedJWTOrAPIKeyAuthentication(BaseAuthentication):

api/src/backend/api/celery_utils.py

@@ -0,0 +1,40 @@
+import ast
+import json
+from typing import Any
+
+_UNPARSED = object()
+
+
+def decode_celery_field(value: Any, default: Any) -> Any:
+    """Decode a Celery result field and require JSON-serializable output."""
+    decoded = value
+    for _ in range(2):
+        if not isinstance(decoded, str):
+            break
+
+        text = decoded.strip()
+        if not text:
+            decoded = default
+            break
+
+        parsed = _UNPARSED
+        for parser in (json.loads, ast.literal_eval):
+            try:
+                parsed = parser(text)
+                break
+            except (TypeError, ValueError, SyntaxError):
+                continue
+
+        if parsed is _UNPARSED:
+            raise ValueError("Unable to decode Celery result field")
+        decoded = parsed
+
+    decoded = default if decoded is None else decoded
+    try:
+        json.dumps(decoded, allow_nan=False)
+    except (TypeError, ValueError) as error:
+        raise ValueError(
+            "Decoded Celery result field is not JSON serializable"
+        ) from error
+
+    return decoded

api/src/backend/api/signals.py

@@ -1,3 +1,4 @@
+from api.db_router import MainRouter
 from api.db_utils import delete_related_daily_task
 from api.models import (
     LighthouseProviderConfiguration,
@@ -47,8 +48,15 @@ def revoke_user_api_keys(sender, instance, **kwargs):  # noqa: F841
 
     The entity field will be set to NULL by on_delete=SET_NULL,
     but we explicitly revoke the keys to prevent further use.
+
+    The update runs on the admin connection because `api_keys` is RLS protected and its
+    policy denies every row when `api.tenant_id` is unset. Users are deleted through the
+    admin connection and may belong to several tenants, so going through the default
+    connection would silently revoke nothing, or only the keys of the active tenant.
     """
-    TenantAPIKey.objects.filter(entity=instance).update(revoked=True)
+    TenantAPIKey.objects.using(MainRouter.admin_db).filter(entity=instance).update(
+        revoked=True
+    )
 
 
 @receiver(post_delete, sender=Membership)
@@ -58,8 +66,12 @@ def revoke_membership_api_keys(sender, instance, **kwargs):  # noqa: F841
 
     When a membership is deleted, all API keys created by that user
     in that tenant should be revoked to prevent further access.
+
+    Uses the admin connection for the same reason as `revoke_user_api_keys`: the RLS
+    policy on `api_keys` denies every row when `api.tenant_id` is unset, which is the
+    case when the membership is removed as a cascade of a user deletion.
     """
-    TenantAPIKey.objects.filter(
+    TenantAPIKey.objects.using(MainRouter.admin_db).filter(
         entity_id=instance.user_id, tenant_id=instance.tenant_id
     ).update(revoked=True)
 

api/src/backend/api/specs/v1.yaml

@@ -1,7 +1,7 @@
 openapi: 3.0.3
 info:
   title: Prowler API
-  version: 1.37.0
+  version: 1.37.1
   description: |-
     Prowler API specification.
 

api/src/backend/api/tests/integration/test_authentication.py

@@ -4,8 +4,11 @@ from datetime import UTC, datetime, timedelta
 from uuid import uuid4
 
 import pytest
+from api.db_router import MainRouter
 from api.models import Membership, Role, TenantAPIKey, User, UserRoleRelationship
+from api.signals import revoke_membership_api_keys, revoke_user_api_keys
 from conftest import TEST_PASSWORD, get_api_tokens, get_authorization_header
+from django.db.utils import ConnectionDoesNotExist
 from django.urls import reverse
 from drf_simple_apikey.crypto import get_crypto
 from rest_framework.test import APIClient
@@ -625,6 +628,34 @@ class TestAPIKeyErrors:
         assert response.status_code == 401
         assert "API Key has been revoked." in response.json()["errors"][0]["detail"]
 
+    def test_orphaned_api_key_rejected(
+        self, create_test_user, tenants_fixture, api_keys_fixture
+    ):
+        """Key whose owning user was deleted returns 401 instead of 500."""
+        client = APIClient()
+
+        api_key = api_keys_fixture[0]
+        # `on_delete=SET_NULL` leaves the key behind with no entity when the owner goes
+        TenantAPIKey.objects.filter(id=api_key.id).update(entity=None)
+
+        api_key_headers = get_api_key_header(api_key._raw_key)
+        response = client.get(reverse("provider-list"), headers=api_key_headers)
+
+        assert response.status_code == 401
+        assert (
+            "No entity matching this api key." in response.json()["errors"][0]["detail"]
+        )
+
+        # The orphaned key is revoked on use; retries fail the regular revoked check
+        api_key.refresh_from_db()
+        assert api_key.revoked is True
+
+        retry_response = client.get(reverse("provider-list"), headers=api_key_headers)
+        assert retry_response.status_code == 401
+        assert (
+            "API Key has been revoked." in retry_response.json()["errors"][0]["detail"]
+        )
+
     def test_non_existent_api_key(self, create_test_user, tenants_fixture):
         """Key UUID doesn't exist in database."""
         client = APIClient()
@@ -817,6 +848,93 @@ class TestAPIKeyTenantIsolation:
         error_detail = response_json["errors"][0]["detail"]
         assert "revoked" in error_detail.lower()
 
+    def test_deleting_user_revokes_api_keys_in_every_tenant(self, tenants_fixture):
+        """Deleting a user revokes their keys in all their tenants, not just one."""
+        first_tenant, second_tenant = tenants_fixture[0], tenants_fixture[1]
+
+        test_user = User.objects.create_user(
+            name="multi_tenant_user",
+            email="multi_tenant_user@prowler.com",
+            password=TEST_PASSWORD,
+        )
+        for tenant in (first_tenant, second_tenant):
+            Membership.objects.create(
+                user=test_user, tenant=tenant, role=Membership.RoleChoices.OWNER
+            )
+
+        first_key, _ = TenantAPIKey.objects.create_api_key(
+            name="Key in first tenant", tenant_id=first_tenant.id, entity=test_user
+        )
+        second_key, _ = TenantAPIKey.objects.create_api_key(
+            name="Key in second tenant", tenant_id=second_tenant.id, entity=test_user
+        )
+
+        test_user.delete()
+
+        first_key.refresh_from_db()
+        second_key.refresh_from_db()
+        assert first_key.revoked is True
+        assert second_key.revoked is True
+        # `on_delete=SET_NULL` orphans the keys, so revoking them is what keeps them
+        # from authenticating
+        assert first_key.entity_id is None
+        assert second_key.entity_id is None
+
+    def test_revoke_user_api_keys_uses_the_admin_connection(
+        self, monkeypatch, tenants_fixture
+    ):
+        """The revocation must not go through the default connection.
+
+        `api_keys` is RLS protected and its policy denies every row when `api.tenant_id`
+        is unset, which is the case while a user is deleted through the admin
+        connection: the update would silently revoke nothing and leave usable orphaned
+        keys behind.
+
+        Pointing `admin_db` at a missing alias is the only way to assert the connection
+        here, because the test suite runs on a single superuser database with
+        `MainRouter.admin_db` patched to "default" (see `conftest.py`), so RLS never
+        applies and both connections are otherwise indistinguishable.
+        """
+        test_user = User.objects.create_user(
+            name="admin_connection_user",
+            email="admin_connection_user@prowler.com",
+            password=TEST_PASSWORD,
+        )
+        Membership.objects.create(user=test_user, tenant=tenants_fixture[0])
+        TenantAPIKey.objects.create_api_key(
+            name="Key for admin connection check",
+            tenant_id=tenants_fixture[0].id,
+            entity=test_user,
+        )
+
+        monkeypatch.setattr(MainRouter, "admin_db", "missing_admin_alias")
+
+        with pytest.raises(ConnectionDoesNotExist):
+            revoke_user_api_keys(sender=User, instance=test_user)
+
+    def test_revoke_membership_api_keys_uses_the_admin_connection(
+        self, monkeypatch, tenants_fixture
+    ):
+        """Same as the user deletion case: this receiver also runs as its cascade."""
+        test_user = User.objects.create_user(
+            name="admin_connection_membership_user",
+            email="admin_connection_membership_user@prowler.com",
+            password=TEST_PASSWORD,
+        )
+        membership = Membership.objects.create(
+            user=test_user, tenant=tenants_fixture[0]
+        )
+        TenantAPIKey.objects.create_api_key(
+            name="Key for membership admin connection check",
+            tenant_id=tenants_fixture[0].id,
+            entity=test_user,
+        )
+
+        monkeypatch.setattr(MainRouter, "admin_db", "missing_admin_alias")
+
+        with pytest.raises(ConnectionDoesNotExist):
+            revoke_membership_api_keys(sender=Membership, instance=membership)
+
 
 @pytest.mark.django_db
 class TestAPIKeyLifecycle:

api/src/backend/api/tests/test_authentication.py

@@ -4,11 +4,17 @@ from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
 import pytest
-from api.authentication import SSEAuthentication, TenantAPIKeyAuthentication
+from api.authentication import (
+    OrphanedAPIKeyError,
+    SSEAuthentication,
+    TenantAPIKeyAuthentication,
+)
 from api.db_router import MainRouter
 from api.models import TenantAPIKey
+from django.db import connections
 from django.db.models.query import QuerySet
 from django.test import RequestFactory
+from django.test.utils import CaptureQueriesContext
 from rest_framework.exceptions import AuthenticationFailed
 
 
@@ -38,13 +44,12 @@ class TestTenantAPIKeyAuthentication:
         request = request_factory.get("/")
 
         # Call the method
-        entity, auth_dict = auth_backend._authenticate_credentials(
-            request, encrypted_key
-        )
+        validated_key = auth_backend._authenticate_credentials(request, encrypted_key)
 
         # Verify that the entity is the user associated with the API key
-        assert entity == api_key.entity
-        assert entity.id == api_key.entity.id
+        assert validated_key.id == api_key.id
+        assert validated_key.entity == api_key.entity
+        assert validated_key.entity.id == api_key.entity.id
 
     def test_authenticate_credentials_restores_manager_on_success(
         self, auth_backend, api_keys_fixture, request_factory
@@ -231,6 +236,120 @@ class TestTenantAPIKeyAuthentication:
 
         assert str(exc_info.value.detail) == "This API Key has been revoked."
 
+    def test_authenticate_credentials_orphaned_api_key(
+        self, auth_backend, api_keys_fixture, request_factory
+    ):
+        """Test credential validation fails when the owning user no longer exists."""
+        api_key = api_keys_fixture[0]
+        _, encrypted_key = api_key._raw_key.split(TenantAPIKey.objects.separator, 1)
+
+        # `entity` is what `on_delete=SET_NULL` leaves behind when the owner is deleted
+        TenantAPIKey.objects.filter(id=api_key.id).update(entity=None)
+
+        request = request_factory.get("/")
+
+        with pytest.raises(OrphanedAPIKeyError):
+            auth_backend._authenticate_credentials(request, encrypted_key)
+
+        # The orphaned key is revoked on use, so it stops showing up as active
+        api_key.refresh_from_db()
+        assert api_key.revoked is True
+
+    def test_authenticate_orphaned_api_key(
+        self, auth_backend, api_keys_fixture, request_factory
+    ):
+        """Test authentication fails with a key whose owning user was deleted.
+
+        Regression test: this used to raise `AttributeError: 'NoneType' object has no
+        attribute 'id'` while building the auth dict, which DRF re-raises as
+        `WrappedAttributeError` and turns into a 500 instead of a 401.
+        """
+        api_key = api_keys_fixture[0]
+        raw_key = api_key._raw_key
+
+        TenantAPIKey.objects.filter(id=api_key.id).update(entity=None)
+
+        request = request_factory.get("/")
+        request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
+
+        with pytest.raises(AuthenticationFailed) as exc_info:
+            auth_backend.authenticate(request)
+
+        assert str(exc_info.value.detail) == "No entity matching this api key."
+
+        # The orphaned key is revoked on use; retries fail the regular revoked check
+        api_key.refresh_from_db()
+        assert api_key.revoked is True
+
+        with pytest.raises(AuthenticationFailed) as exc_info:
+            auth_backend.authenticate(request)
+
+        assert str(exc_info.value.detail) == "This API Key has been revoked."
+
+    def test_authenticate_reads_the_api_key_once_under_a_row_lock(
+        self, auth_backend, api_keys_fixture, request_factory
+    ):
+        """Test the API key is read a single time and the row is locked.
+
+        Validation, the `last_used_at` update and the claims must all come from the
+        same authoritative row: a second, unlocked lookup would reopen the window
+        where a key revoked in between still authenticates.
+        """
+        api_key = api_keys_fixture[0]
+
+        request = request_factory.get("/")
+        request.META["HTTP_AUTHORIZATION"] = f"Api-Key {api_key._raw_key}"
+
+        with CaptureQueriesContext(connections[MainRouter.admin_db]) as captured:
+            auth_backend.authenticate(request)
+
+        api_key_selects = [
+            query["sql"]
+            for query in captured.captured_queries
+            if query["sql"].startswith("SELECT") and '"api_keys"' in query["sql"]
+        ]
+
+        assert len(api_key_selects) == 1
+        assert "FOR UPDATE" in api_key_selects[0]
+
+    def test_authenticate_ignores_revocation_after_the_locked_read(
+        self, auth_backend, api_keys_fixture, request_factory
+    ):
+        """Test the claims describe the row that was validated, not a later state.
+
+        Regression test: the key used to be looked up again to build the auth dict,
+        without rechecking `revoked` or `entity`. A key revoked or orphaned between
+        both reads still authenticated, and the claims came from that stale row. With
+        a single locked read the write below cannot land mid-authentication, and the
+        revocation only takes effect on the next request.
+        """
+        api_key = api_keys_fixture[0]
+        entity_at_validation = api_key.entity
+        original_save = TenantAPIKey.save
+
+        def revoke_and_orphan_before_saving(instance, *args, **kwargs):
+            # Runs after validation, right before the claims are built: the exact
+            # window a concurrent revocation or user deletion used to slip into
+            TenantAPIKey.objects.filter(id=api_key.id).update(revoked=True, entity=None)
+            return original_save(instance, *args, **kwargs)
+
+        request = request_factory.get("/")
+        request.META["HTTP_AUTHORIZATION"] = f"Api-Key {api_key._raw_key}"
+
+        with patch.object(TenantAPIKey, "save", revoke_and_orphan_before_saving):
+            entity, auth_dict = auth_backend.authenticate(request)
+
+        assert entity == entity_at_validation
+        assert auth_dict["sub"] == str(entity_at_validation.id)
+        assert auth_dict["tenant_id"] == str(api_key.tenant_id)
+        assert auth_dict["api_key_prefix"] == api_key.prefix
+
+        # The revoked key is rejected from the next request on
+        with pytest.raises(AuthenticationFailed) as exc_info:
+            auth_backend.authenticate(request)
+
+        assert str(exc_info.value.detail) == "This API Key has been revoked."
+
     def test_authenticate_expired_api_key(
         self, auth_backend, create_test_user, tenants_fixture, request_factory
     ):

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

@@ -65,6 +65,7 @@ from api.v1.views import (
 )
 from botocore.exceptions import ClientError, NoCredentialsError
 from celery import states
+from celery.utils.saferepr import saferepr
 from conftest import (
     API_JSON_CONTENT_TYPE,
     TEST_PASSWORD,
@@ -5043,11 +5044,60 @@ class TestTaskViewSet:
             reverse("task-detail", kwargs={"pk": task1.id}),
         )
         assert response.status_code == status.HTTP_200_OK
+        assert response.json()["data"]["attributes"]["task_args"] == {
+            "kwarg1": "value1"
+        }
         assert (
             response.json()["data"]["attributes"]["name"]
             == task1.task_runner_task.task_name
         )
 
+    def test_tasks_retrieve_hides_tenant_id(
+        self, authenticated_client, tasks_fixture, tenants_fixture
+    ):
+        task, *_ = tasks_fixture
+        task.task_runner_task.task_kwargs = json.dumps(
+            repr(
+                {
+                    "tenant_id": str(tenants_fixture[0].id),
+                    "enabled": True,
+                    "scan_id": None,
+                    "label": "True North",
+                }
+            )
+        )
+        task.task_runner_task.save(update_fields=["task_kwargs"])
+
+        response = authenticated_client.get(
+            reverse("task-detail", kwargs={"pk": task.id}),
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["data"]["attributes"]["task_args"] == {
+            "enabled": True,
+            "scan_id": None,
+            "label": "True North",
+        }
+
+    def test_tasks_retrieve_with_truncated_kwargs_returns_empty_task_args(
+        self, authenticated_client, tasks_fixture
+    ):
+        task, *_ = tasks_fixture
+        kwargs_repr = saferepr(
+            {"finding_ids": [str(uuid4()) for _ in range(30)]}, maxlen=1024
+        )
+        assert "..." in kwargs_repr
+        task.task_runner_task.task_kwargs = json.dumps(kwargs_repr)
+        task.task_runner_task.save(update_fields=["task_kwargs"])
+
+        response = authenticated_client.get(
+            reverse("task-detail", kwargs={"pk": task.id}),
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.headers["Content-Type"] == API_JSON_CONTENT_TYPE
+        assert response.json()["data"]["attributes"]["task_args"] == {}
+
     def test_tasks_invalid_retrieve(self, authenticated_client):
         response = authenticated_client.get(
             reverse("task-detail", kwargs={"pk": "invalid_id"})
@@ -15748,6 +15798,23 @@ class TestTenantApiKeyViewSet:
         data = response.json()["data"]
         assert len(data) == len(api_keys_fixture)
 
+    def test_api_keys_list_with_orphaned_key(
+        self, authenticated_client, api_keys_fixture
+    ):
+        """Test listing keys whose owner was deleted: `entity` is serialized as null."""
+        orphaned_key = api_keys_fixture[0]
+        TenantAPIKey.objects.filter(id=orphaned_key.id).update(entity=None)
+
+        response = authenticated_client.get(reverse("api-key-list"))
+
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == len(api_keys_fixture)
+        serialized_key = next(
+            item for item in data if item["id"] == str(orphaned_key.id)
+        )
+        assert serialized_key["relationships"]["entity"]["data"] is None
+
     def test_api_keys_list_empty(self, authenticated_client, tenants_fixture):
         """Test listing API keys when none exist returns empty list."""
         response = authenticated_client.get(reverse("api-key-list"))

api/src/backend/api/v1/serializers.py

@@ -1,8 +1,10 @@
 import base64
 import json
+import logging
 from datetime import UTC, datetime, timedelta
 
 import yaml
+from api.celery_utils import decode_celery_field
 from api.db_router import MainRouter
 from api.exceptions import ConflictException
 from api.models import (
@@ -59,6 +61,7 @@ from api.v1.serializer_utils.lighthouse import (
 from api.v1.serializer_utils.processors import ProcessorConfigField
 from api.v1.serializer_utils.providers import ProviderSecretField
 from api.validators import validate_lighthouse_openai_compatible_base_url
+from config.custom_logging import BackendLogger
 from django.conf import settings
 from django.contrib.auth import authenticate
 from django.contrib.auth.models import update_last_login
@@ -79,6 +82,8 @@ from rest_framework_simplejwt.settings import api_settings
 from rest_framework_simplejwt.tokens import RefreshToken
 from rest_framework_simplejwt.utils import get_md5_hash_password
 
+logger = logging.getLogger(BackendLogger.API)
+
 # Base
 
 
@@ -607,13 +612,24 @@ class TaskSerializer(RLSSerializer, TaskBase):
 
     @extend_schema_field(serializers.JSONField())
     def get_task_args(self, obj):
-        task_args = self.get_json_field(obj, "task_kwargs")
-        # Celery task_kwargs are stored as a double string JSON in the database when not empty
-        if isinstance(task_args, str):
-            task_args = json.loads(task_args.replace("'", '"').replace("None", "null"))
-        # Remove tenant_id from task_kwargs if present
-        task_args.pop("tenant_id", None)
+        task_kwargs = (
+            getattr(obj.task_runner_task, "task_kwargs", None)
+            if obj.task_runner_task
+            else None
+        )
+        try:
+            task_args = decode_celery_field(task_kwargs, {})
+            if not isinstance(task_args, dict):
+                raise ValueError("Decoded task kwargs must be a dictionary")
+        except ValueError:
+            logger.warning(
+                "Unable to decode task kwargs for task %s; returning empty task_args.",
+                obj.id,
+            )
+            return {}
 
+        task_args = task_args.copy()
+        task_args.pop("tenant_id", None)
         return task_args
 
     @staticmethod

api/src/backend/tasks/jobs/integrations.py

@@ -1,5 +1,6 @@
 import os
 import time
+from datetime import UTC, datetime
 from glob import glob
 
 from api.db_router import READ_REPLICA_ALIAS, MainRouter
@@ -214,8 +215,10 @@ def get_security_hub_client_from_integration(
         for region in set(all_security_hub_regions):
             regions_status[region] = region in connection.enabled_regions
 
-        # Save regions information in the integration configuration
+        # Persist the successful connection check and regions information
         with rls_transaction(tenant_id, using=MainRouter.default_db):
+            integration.connected = True
+            integration.connection_last_checked_at = datetime.now(tz=UTC)
             integration.configuration["regions"] = regions_status
             integration.save()
 

api/src/backend/tasks/jobs/orphan_recovery.py

@@ -18,12 +18,11 @@ This is the shared engine behind both the periodic Beat watchdog and the
 `reconcile_orphan_tasks` management command.
 """
 
-import ast
-import json
 from contextlib import contextmanager
 from datetime import UTC, datetime, timedelta
 from uuid import uuid4
 
+from api.celery_utils import decode_celery_field
 from celery import current_app, states
 from celery.utils.log import get_task_logger
 from django.db import connections
@@ -138,34 +137,6 @@ def revoke_task(task_result, terminate: bool = True) -> None:
         logger.exception(f"Failed to revoke task {task_result.task_id}")
 
 
-def _decode_celery_field(value, default):
-    """Decode django-celery-results' stored task_args/task_kwargs to a Python object.
-
-    The backend stores them as a (sometimes double-encoded) repr/JSON string. An
-    empty or missing field returns ``default``; a non-empty value that cannot be
-    decoded raises ``ValueError`` so the caller can avoid re-enqueuing a task with
-    the wrong arguments.
-    """
-    obj = value
-    for _ in range(2):  # values can be double-encoded (a string holding a repr)
-        if not isinstance(obj, str):
-            break
-        text = obj.strip()
-        if not text:
-            return default
-        parsed = None
-        for parser in (ast.literal_eval, json.loads):
-            try:
-                parsed = parser(text)
-                break
-            except (ValueError, SyntaxError, TypeError):
-                continue
-        if parsed is None:
-            raise ValueError(f"undecodable celery field: {text[:120]!r}")
-        obj = parsed
-    return default if obj is None else obj
-
-
 def reconcile_orphans(
     grace_minutes: int = 2,
     max_attempts: int = 3,
@@ -313,8 +284,10 @@ def _recover_task(task_result, max_attempts: int, window_hours: int) -> str:
         return "failed"
 
     try:
-        args = _decode_celery_field(args_repr, [])
-        kwargs = _decode_celery_field(kwargs_repr, {})
+        args = decode_celery_field(args_repr, [])
+        kwargs = decode_celery_field(kwargs_repr, {})
+        if not isinstance(args, (list, tuple)) or not isinstance(kwargs, dict):
+            raise ValueError("Stored task arguments have invalid types")
     except ValueError:
         logger.error(
             "Orphan %s (%s): could not decode stored args/kwargs, not re-enqueuing",
@@ -324,8 +297,8 @@ def _recover_task(task_result, max_attempts: int, window_hours: int) -> str:
         return "failed"
     new_task_id = str(uuid4())
     task_obj.apply_async(
-        args=list(args) if isinstance(args, (list, tuple)) else [],
-        kwargs=kwargs if isinstance(kwargs, dict) else {},
+        args=list(args),
+        kwargs=kwargs,
         task_id=new_task_id,
     )
     logger.info(

api/src/backend/tasks/tests/test_integrations.py

@@ -1,3 +1,4 @@
+from datetime import UTC, datetime
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -671,6 +672,8 @@ class TestSecurityHubIntegrationUploads:
         mock_integration = MagicMock()
         mock_integration.configuration = {"send_only_fails": True}
         mock_integration.credentials = {}  # Empty credentials, use provider
+        mock_integration.connected = False
+        mock_integration.connection_last_checked_at = None
 
         # Mock tenant_id
         tenant_id = "550e8400-e29b-41d4-a716-446655440000"  # Valid UUID
@@ -723,12 +726,22 @@ class TestSecurityHubIntegrationUploads:
             # Configure the test_connection to return our mock_connection
             mock_security_hub_class.test_connection = mock_test_connection
 
+            checked_at_before = datetime.now(tz=UTC)
             connected, security_hub = get_security_hub_client_from_integration(
                 mock_integration, tenant_id, mock_findings
             )
+            checked_at_after = datetime.now(tz=UTC)
 
         assert connected is True
         assert security_hub == mock_security_hub
+        assert mock_integration.connected is True
+        assert mock_integration.connection_last_checked_at.tzinfo is UTC
+        assert (
+            checked_at_before
+            <= mock_integration.connection_last_checked_at
+            <= checked_at_after
+        )
+        mock_integration.save.assert_called_once()
 
         # Verify SecurityHub was called once to create the client
         assert mock_security_hub_class.call_count == 1

api/src/backend/tasks/tests/test_orphan_recovery.py

@@ -3,12 +3,13 @@ from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
 import pytest
+from api.celery_utils import decode_celery_field
 from celery import states
+from celery.utils.saferepr import saferepr
 from django.test import override_settings
 from django_celery_results.models import TaskResult
 from tasks.jobs.orphan_recovery import (
     _SKIP_RECOVERY,
-    _decode_celery_field,
     _reconcile_task_results,
     _recovery_attempt_count,
     advisory_lock,
@@ -36,24 +37,77 @@ def _orphan_result(*, name, kwargs, worker, created_minutes_ago, status=states.S
     return tr
 
 
-@pytest.mark.django_db
 class TestDecodeCeleryField:
+    def test_decodes_strict_json(self):
+        assert decode_celery_field('{"enabled": true, "scan_id": null}', {}) == {
+            "enabled": True,
+            "scan_id": None,
+        }
+
     def test_decodes_single_encoded_repr(self):
-        assert _decode_celery_field("{'tenant_id': 'abc'}", {}) == {"tenant_id": "abc"}
+        assert decode_celery_field("{'tenant_id': 'abc'}", {}) == {"tenant_id": "abc"}
 
     def test_decodes_double_encoded(self):
         import json
 
         stored = json.dumps(repr({"tenant_id": "abc", "scan_id": "s1"}))
-        assert _decode_celery_field(stored, {}) == {"tenant_id": "abc", "scan_id": "s1"}
+        assert decode_celery_field(stored, {}) == {
+            "tenant_id": "abc",
+            "scan_id": "s1",
+        }
+
+    def test_python_words_inside_strings_are_preserved(self):
+        stored = repr(
+            {
+                "enabled": True,
+                "scan_id": None,
+                "label": "True North",
+                "note": "None",
+            }
+        )
+
+        assert decode_celery_field(stored, {}) == {
+            "enabled": True,
+            "scan_id": None,
+            "label": "True North",
+            "note": "None",
+        }
 
     def test_empty_returns_default(self):
-        assert _decode_celery_field(None, {}) == {}
-        assert _decode_celery_field("", []) == []
+        assert decode_celery_field(None, {}) == {}
+        assert decode_celery_field("", []) == []
+        assert decode_celery_field("null", {}) == {}
+        assert decode_celery_field("None", []) == []
+
+    def test_empty_validates_default(self):
+        with pytest.raises(ValueError):
+            decode_celery_field("", {"value": ...})
 
     def test_unparseable_raises(self):
         with pytest.raises(ValueError):
-            _decode_celery_field("<<not a literal>>", {})
+            decode_celery_field("<<not a literal>>", {})
+
+    @pytest.mark.parametrize(
+        "value",
+        (
+            "{'value': ...}",
+            "{'value': {1, 2}}",
+            "{'value': b'bytes'}",
+            '{"value": NaN}',
+        ),
+    )
+    def test_non_json_values_raise(self, value):
+        with pytest.raises(ValueError):
+            decode_celery_field(value, {})
+
+    def test_truncated_repr_raises(self):
+        kwargs_repr = saferepr(
+            {"finding_ids": [str(uuid4()) for _ in range(30)]}, maxlen=1024
+        )
+        assert "..." in kwargs_repr
+
+        with pytest.raises(ValueError):
+            decode_celery_field(kwargs_repr, {})
 
 
 @pytest.mark.django_db
@@ -98,6 +152,58 @@ class TestReconcileTaskResults:
         assert call["kwargs"] == {"tenant_id": str(tenant.id)}
         assert call["task_id"] != tr.task_id  # fresh task id
 
+    def test_truncated_kwargs_are_not_reenqueued(self, tenants_fixture):
+        tenant = tenants_fixture[0]
+        tr = _orphan_result(
+            name="tenant-deletion",
+            kwargs={"tenant_id": str(tenant.id)},
+            worker="dead@gone",
+            created_minutes_ago=60,
+        )
+        tr.task_kwargs = saferepr(
+            {"finding_ids": [str(uuid4()) for _ in range(30)]}, maxlen=1024
+        )
+        assert "..." in tr.task_kwargs
+        tr.save(update_fields=["task_kwargs"])
+        p_alive, p_revoke, p_app, mock_task = self._patches(alive=False)
+
+        with (
+            p_alive,
+            p_revoke,
+            p_app,
+            patch("tasks.jobs.orphan_recovery._recovery_attempt_count", return_value=1),
+        ):
+            result = _reconcile_task_results(
+                grace_minutes=2, max_attempts=3, window_hours=6, dry_run=False
+            )
+
+        assert tr.task_id in result["failed"]
+        mock_task.apply_async.assert_not_called()
+
+    def test_wrong_kwargs_shape_is_not_reenqueued(self, tenants_fixture):
+        tr = _orphan_result(
+            name="tenant-deletion",
+            kwargs={"tenant_id": str(tenants_fixture[0].id)},
+            worker="dead@gone",
+            created_minutes_ago=60,
+        )
+        tr.task_kwargs = "[]"
+        tr.save(update_fields=["task_kwargs"])
+        p_alive, p_revoke, p_app, mock_task = self._patches(alive=False)
+
+        with (
+            p_alive,
+            p_revoke,
+            p_app,
+            patch("tasks.jobs.orphan_recovery._recovery_attempt_count", return_value=1),
+        ):
+            result = _reconcile_task_results(
+                grace_minutes=2, max_attempts=3, window_hours=6, dry_run=False
+            )
+
+        assert tr.task_id in result["failed"]
+        mock_task.apply_async.assert_not_called()
+
     def test_external_integration_task_is_not_reenqueued_by_default(
         self, tenants_fixture
     ):

api/uv.lock

@@ -4673,8 +4673,8 @@ wheels = [
 
 [[package]]
 name = "prowler"
-version = "5.35.0"
-source = { git = "https://github.com/prowler-cloud/prowler.git?rev=master#f5ea116763aeffede9f399c8934fc280eaccd315" }
+version = "5.36.0"
+source = { git = "https://github.com/prowler-cloud/prowler.git?rev=v5.36#2298d4a3f881abe9e2266195f764b60d34af07eb" }
 dependencies = [
     { name = "alibabacloud-actiontrail20200706" },
     { name = "alibabacloud-credentials" },
@@ -4762,7 +4762,7 @@ dependencies = [
 
 [[package]]
 name = "prowler-api"
-version = "1.37.0"
+version = "1.37.1"
 source = { virtual = "." }
 dependencies = [
     { name = "cartography" },
@@ -4862,7 +4862,7 @@ requires-dist = [
     { name = "matplotlib", specifier = "==3.10.8" },
     { name = "neo4j", specifier = "==6.1.0" },
     { name = "openai", specifier = "==1.109.1" },
-    { name = "prowler", git = "https://github.com/prowler-cloud/prowler.git?rev=master" },
+    { name = "prowler", git = "https://github.com/prowler-cloud/prowler.git?rev=v5.36" },
     { name = "psycopg2-binary", specifier = "==2.9.9" },
     { name = "pytest-celery", extras = ["redis"], specifier = "==1.3.0" },
     { name = "reportlab", specifier = "==4.4.10" },

prowler/changelog.d/gcp-cloudfunction-gen2-iam-thread-safety.fixed.md

@@ -0,0 +1 @@
+GCP Cloud Functions gen2 IAM policy retrieval now uses a per-request HTTP client, preventing a process crash from concurrent thread-unsafe `httplib2` access when a project has several gen2 functions

prowler/changelog.d/gcp-firewall-multiport.fixed.md

@@ -0,0 +1 @@
+GCP firewall SSH and RDP checks now detect exposed target ports in any position within multi-port rules

prowler/changelog.d/jira-adf-metadata.fixed.md

@@ -0,0 +1 @@
+Jira descriptions with inline code nested in bold or italic Markdown now render as valid ADF

prowler/config/config.py

@@ -49,7 +49,7 @@ class _MutableTimestamp:
 
 timestamp = _MutableTimestamp(datetime.today())
 timestamp_utc = _MutableTimestamp(datetime.now(timezone.utc))
-prowler_version = "5.36.0"
+prowler_version = "5.36.1"
 html_logo_url = "https://github.com/prowler-cloud/prowler/"
 square_logo_img = "https://raw.githubusercontent.com/prowler-cloud/prowler/dc7d2d5aeb92fdf12e8604f42ef6472cd3e8e889/docs/img/prowler-logo-black.png"
 aws_logo = "https://user-images.githubusercontent.com/38561120/235953920-3e3fba08-0795-41dc-b480-9bea57db9f2e.png"

prowler/lib/outputs/jira/jira.py

@@ -203,7 +203,9 @@ class MarkdownToADFConverter:
             if token_type == "text":
                 result.extend(self._text_to_nodes(token.content, marks_stack))
             elif token_type == "code_inline":
-                marks = self._clone_marks(marks_stack)
+                marks = self._clone_marks(
+                    [mark for mark in marks_stack if mark["type"] == "link"]
+                )
                 marks.append({"type": "code"})
                 result.append(self._create_text_node(token.content, marks))
             elif token_type in {"softbreak", "hardbreak"}:

prowler/providers/gcp/services/cloudfunction/cloudfunction_service.py

@@ -108,7 +108,10 @@ class CloudFunction(GCPService):
                     .locations()
                     .services()
                     .getIamPolicy(resource=function.service)
-                    .execute(num_retries=DEFAULT_RETRY_ATTEMPTS)
+                    .execute(
+                        http=self.__get_AuthorizedHttp_client__(),
+                        num_retries=DEFAULT_RETRY_ATTEMPTS,
+                    )
                 )
             else:
                 response = (

prowler/providers/gcp/services/compute/compute_firewall_rdp_access_from_the_internet_allowed/compute_firewall_rdp_access_from_the_internet_allowed.py

@@ -31,7 +31,7 @@ class compute_firewall_rdp_access_from_the_internet_allowed(Check):
                                     break
                             elif int(port) == 3389:
                                 opened_port = True
-                            break
+                                break
             if (
                 "0.0.0.0/0" in firewall.source_ranges
                 and firewall.direction == "INGRESS"

prowler/providers/gcp/services/compute/compute_firewall_ssh_access_from_the_internet_allowed/compute_firewall_ssh_access_from_the_internet_allowed.py

@@ -31,7 +31,7 @@ class compute_firewall_ssh_access_from_the_internet_allowed(Check):
                                     break
                             elif int(port) == 22:
                                 opened_port = True
-                            break
+                                break
             if (
                 "0.0.0.0/0" in firewall.source_ranges
                 and firewall.direction == "INGRESS"

pyproject.toml

@@ -125,7 +125,7 @@ maintainers = [{name = "Prowler Engineering", email = "engineering@prowler.com"}
 name = "prowler"
 readme = "README.md"
 requires-python = ">=3.10,<3.14"
-version = "5.36.0"
+version = "5.36.1"
 
 [project.scripts]
 prowler = "prowler.__main__:prowler"

tests/lib/outputs/jira/jira_test.py

@@ -23,11 +23,55 @@ from prowler.lib.outputs.jira.exceptions.exceptions import (
     JiraSendFindingsResponseError,
     JiraTestConnectionError,
 )
-from prowler.lib.outputs.jira.jira import Jira
+from prowler.lib.outputs.jira.jira import Jira, MarkdownToADFConverter
 
 TEST_DATETIME = "2023-01-01T12:01:01+00:00"
 
 
+class TestMarkdownToADFConverter:
+    def setup_method(self):
+        self.converter = MarkdownToADFConverter()
+
+    def test_inline_code_nested_in_strong_has_only_code_mark(self):
+        result = self.converter.convert("**before `code` after**")
+
+        assert result[0]["content"] == [
+            {"type": "text", "text": "before ", "marks": [{"type": "strong"}]},
+            {"type": "text", "text": "code", "marks": [{"type": "code"}]},
+            {"type": "text", "text": " after", "marks": [{"type": "strong"}]},
+        ]
+
+    def test_inline_code_nested_in_emphasis_has_only_code_mark(self):
+        result = self.converter.convert("*before `code` after*")
+
+        assert result[0]["content"] == [
+            {"type": "text", "text": "before ", "marks": [{"type": "em"}]},
+            {"type": "text", "text": "code", "marks": [{"type": "code"}]},
+            {"type": "text", "text": " after", "marks": [{"type": "em"}]},
+        ]
+
+    def test_inline_code_in_link_preserves_link_and_code_marks(self):
+        result = self.converter.convert("[`code`](https://example.com)")
+
+        assert result[0]["content"][0]["marks"] == [
+            {"type": "link", "attrs": {"href": "https://example.com"}},
+            {"type": "code"},
+        ]
+
+    def test_inline_code_in_emphasized_link_preserves_link_and_code_marks(self):
+        result = self.converter.convert("*[`code`](https://example.com)*")
+
+        assert result[0]["content"][0]["marks"] == [
+            {"type": "link", "attrs": {"href": "https://example.com"}},
+            {"type": "code"},
+        ]
+
+    def test_standalone_inline_code_has_code_mark(self):
+        result = self.converter.convert("`code`")
+
+        assert result[0]["content"][0]["marks"] == [{"type": "code"}]
+
+
 class TestJiraIntegration:
     @pytest.fixture(autouse=True)
     @patch.object(Jira, "get_auth", return_value=None)

tests/providers/gcp/services/cloudfunction/cloudfunction_service_test.py

@@ -1,5 +1,7 @@
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, call, patch
 
+from prowler.providers.gcp.config import DEFAULT_RETRY_ATTEMPTS
+from prowler.providers.gcp.lib.service.service import GCPService
 from prowler.providers.gcp.services.cloudfunction.cloudfunction_service import (
     CloudFunction,
 )
@@ -148,6 +150,90 @@ class TestCloudFunctionService:
             assert fn.vpc_connector is None
             assert fn.publicly_accessible is False
 
+    def test_get_functions_iam_policy_gen2_uses_distinct_per_request_http(self):
+        """Regression: the gen2 IAM lookup must pass a per-request HTTP client.
+
+        _get_function_iam_policy runs once per function across a thread pool
+        (GCPService.__threading_call__), and httplib2 is not thread-safe. The
+        gen1 branch isolates each thread with its own AuthorizedHttp via
+        __get_AuthorizedHttp_client__; the gen2 branch must do the same. Sharing
+        the single self._run_client transport across threads corrupts the
+        process heap and aborts the scan (SIGABRT/SIGSEGV).
+        """
+        second_function_name = "second-function"
+        second_function_id = f"projects/{GCP_PROJECT_ID}/locations/{_LOCATION_ID}/functions/{second_function_name}"
+        second_run_service = f"projects/{GCP_PROJECT_ID}/locations/{_LOCATION_ID}/services/{second_function_name}"
+        first_http = object()
+        second_http = object()
+
+        first_request = MagicMock()
+        first_request.execute.return_value = {"bindings": []}
+        second_request = MagicMock()
+        second_request.execute.return_value = {"bindings": []}
+        run_client = MagicMock()
+        get_iam_policy = run_client.projects().locations().services().getIamPolicy
+        get_iam_policy.side_effect = [first_request, second_request]
+
+        def run_sequentially(self, callback, iterator):
+            for value in iterator:
+                callback(value)
+
+        def mock_api_client(*args, **kwargs):
+            return _make_cloudfunction_client(
+                functions_list=[
+                    {
+                        "name": _FUNCTION_ID,
+                        "state": "ACTIVE",
+                        "environment": "GEN_2",
+                        "serviceConfig": {"service": _RUN_SERVICE},
+                    },
+                    {
+                        "name": second_function_id,
+                        "state": "ACTIVE",
+                        "environment": "GEN_2",
+                        "serviceConfig": {"service": second_run_service},
+                    },
+                ]
+            )
+
+        with (
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__is_api_active__",
+                new=mock_is_api_active,
+            ),
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__generate_client__",
+                new=mock_api_client,
+            ),
+            patch(
+                "prowler.providers.gcp.services.cloudfunction.cloudfunction_service.discovery.build",
+                return_value=run_client,
+            ),
+            patch.object(
+                GCPService,
+                "__get_AuthorizedHttp_client__",
+                side_effect=[first_http, second_http],
+            ),
+            patch.object(
+                GCPService,
+                "__threading_call__",
+                new=run_sequentially,
+            ),
+        ):
+            CloudFunction(set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID]))
+
+        get_iam_policy.assert_has_calls(
+            [call(resource=_RUN_SERVICE), call(resource=second_run_service)]
+        )
+        first_request.execute.assert_called_once_with(
+            http=first_http,
+            num_retries=DEFAULT_RETRY_ATTEMPTS,
+        )
+        second_request.execute.assert_called_once_with(
+            http=second_http,
+            num_retries=DEFAULT_RETRY_ATTEMPTS,
+        )
+
     def test_get_functions_iam_policy_gen2_all_users(self):
         """Gen2 functions: allUsers binding lives on the Cloud Run service."""
 

tests/providers/gcp/services/compute/compute_firewall_rdp_access_from_the_internet_allowed/compute_firewall_rdp_access_from_the_internet_allowed_test.py

@@ -279,6 +279,48 @@ class Test_compute_firewall_rdp_access_from_the_internet_allowed:
             )
             assert result[0].resource_id == firewall.id
 
+    def test_one_non_compliant_rule_with_multiple_ports(self):
+        from prowler.providers.gcp.services.compute.compute_service import Firewall
+
+        firewall = Firewall(
+            name="test",
+            id="1234567890",
+            source_ranges=["0.0.0.0/0"],
+            direction="INGRESS",
+            allowed_rules=[{"IPProtocol": "tcp", "ports": ["80", "3389"]}],
+            project_id=GCP_PROJECT_ID,
+        )
+
+        compute_client = mock.MagicMock()
+        compute_client.project_ids = [GCP_PROJECT_ID]
+        compute_client.firewalls = [firewall]
+        compute_client.region = "global"
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.gcp.services.compute.compute_firewall_rdp_access_from_the_internet_allowed.compute_firewall_rdp_access_from_the_internet_allowed.compute_client",
+                new=compute_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.compute.compute_firewall_rdp_access_from_the_internet_allowed.compute_firewall_rdp_access_from_the_internet_allowed import (
+                compute_firewall_rdp_access_from_the_internet_allowed,
+            )
+
+            check = compute_firewall_rdp_access_from_the_internet_allowed()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert search(
+                f"Firewall {firewall.name} does exposes port 3389",
+                result[0].status_extended,
+            )
+            assert result[0].resource_id == firewall.id
+
     def test_one_non_compliant_rule_with_port_range(self):
         from prowler.providers.gcp.services.compute.compute_service import Firewall
 

tests/providers/gcp/services/compute/compute_firewall_ssh_access_from_the_internet_allowed/compute_firewall_ssh_access_from_the_internet_allowed_test.py

@@ -279,6 +279,48 @@ class Test_compute_firewall_ssh_access_from_the_internet_allowed:
             )
             assert result[0].resource_id == firewall.id
 
+    def test_one_non_compliant_rule_with_multiple_ports(self):
+        from prowler.providers.gcp.services.compute.compute_service import Firewall
+
+        firewall = Firewall(
+            name="test",
+            id="1234567890",
+            source_ranges=["0.0.0.0/0"],
+            direction="INGRESS",
+            allowed_rules=[{"IPProtocol": "tcp", "ports": ["80", "22"]}],
+            project_id=GCP_PROJECT_ID,
+        )
+
+        compute_client = mock.MagicMock()
+        compute_client.project_ids = [GCP_PROJECT_ID]
+        compute_client.firewalls = [firewall]
+        compute_client.region = "global"
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.gcp.services.compute.compute_firewall_ssh_access_from_the_internet_allowed.compute_firewall_ssh_access_from_the_internet_allowed.compute_client",
+                new=compute_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.compute.compute_firewall_ssh_access_from_the_internet_allowed.compute_firewall_ssh_access_from_the_internet_allowed import (
+                compute_firewall_ssh_access_from_the_internet_allowed,
+            )
+
+            check = compute_firewall_ssh_access_from_the_internet_allowed()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert search(
+                f"Firewall {firewall.name} does exposes port 22",
+                result[0].status_extended,
+            )
+            assert result[0].resource_id == firewall.id
+
     def test_one_non_compliant_rule_with_port_range(self):
         from prowler.providers.gcp.services.compute.compute_service import Firewall
 

uv.lock

@@ -3553,7 +3553,7 @@ wheels = [
 
 [[package]]
 name = "prowler"
-version = "5.36.0"
+version = "5.36.1"
 source = { editable = "." }
 dependencies = [
     { name = "alibabacloud-actiontrail20200706" },