fix(sdk): address Cloudflare review feedback

- Clarify the missing challenge TTL changelog behavior
- Make account fallback expectations explicit in provider tests

README.md

@@ -139,7 +139,7 @@ Every AWS provider scan will enqueue an Attack Paths ingestion job automatically
 | MongoDB Atlas | 10 | 3 | 1 | 8 | Official | UI, API, CLI |
 | LLM | [See `promptfoo` docs.](https://www.promptfoo.dev/docs/red-team/plugins/) | N/A | N/A | N/A | Official | CLI |
 | Image | N/A | N/A | N/A | N/A | Official | UI, API, CLI |
-| Google Workspace | 65 | 11 | 3 | 6 | Official | UI, API, CLI |
+| Google Workspace | 65 | 11 | 4 | 6 | Official | UI, API, CLI |
 | OpenStack | 34 | 5 | 1 | 9 | Official | UI, API, CLI |
 | Vercel | 26 | 6 | 1 | 8 | Official | UI, API, CLI |
 | Okta | 29 | 8 | 2 | 2 | Official | UI, API, CLI |

api/.env.example

@@ -59,5 +59,9 @@ DJANGO_GITHUB_OAUTH_CLIENT_ID=""
 DJANGO_GITHUB_OAUTH_CLIENT_SECRET=""
 DJANGO_GITHUB_OAUTH_CALLBACK_URL=""
 
+# Public base URL of the Prowler UI, used to link Jira issues back to findings.
+# Leave empty to omit the link.
+DJANGO_UI_BASE_URL=""
+
 # Deletion Task Batch Size
 DJANGO_DELETION_BATCH_SIZE=5000

api/changelog.d/jira-finding-labels-url.added.md

@@ -0,0 +1 @@
+Jira issues created from Prowler Cloud now carry the `prowler`, `prowler-<provider>`, `prowler-<severity>`, `prowler-<check-id>`, and `prowler-finding-<finding-uid>` labels, a link back to the finding when `DJANGO_UI_BASE_URL` is configured, and the tenant name

api/changelog.d/mute-rules-current-scans.fixed.md

@@ -0,0 +1 @@
+`POST /api/v1/mute-rules` now updates only each affected provider's latest completed scan and future scans, preventing historical reaggregation from flooding Celery queues

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

@@ -18333,19 +18333,14 @@ class TestMuteRuleViewSet:
         assert len(data) == 2
         assert data[0]["id"] == str(mute_rules_fixture[first_index].id)
 
-    @patch("api.v1.views.chain")
-    @patch("api.v1.views.reaggregate_all_finding_group_summaries_task.si")
-    @patch("api.v1.views.mute_historical_findings_task.si")
+    @patch("api.v1.views.mute_findings_in_latest_scans_task.apply_async")
     @patch("api.v1.views.transaction.on_commit", side_effect=lambda fn: fn())
     def test_mute_rules_create_valid(
         self,
         _mock_on_commit,
-        mock_mute_signature,
-        mock_reaggregate_signature,
-        mock_chain,
+        mock_mute_task,
         authenticated_client,
         findings_fixture,
-        create_test_user,
     ):
         """Test creating a valid mute rule."""
         finding_ids = [str(findings_fixture[0].id)]
@@ -18372,24 +18367,20 @@ class TestMuteRuleViewSet:
         assert response_data["attributes"]["name"] == "New Mute Rule"
         assert response_data["attributes"]["reason"] == "Security exception approved"
 
-        # Verify the finding was immediately muted
-        from api.models import Finding
-
         finding = Finding.objects.get(id=findings_fixture[0].id)
-        assert finding.muted is True
-        assert finding.muted_at is not None
-        assert finding.muted_reason == "Security exception approved"
+        assert finding.muted is False
+        assert finding.muted_at is None
+        assert finding.muted_reason is None
 
-        # Verify background task chain was called: mute → reaggregate all
-        mock_mute_signature.assert_called_once()
-        mock_reaggregate_signature.assert_called_once()
-        mock_chain.assert_called_once_with(
-            mock_mute_signature.return_value,
-            mock_reaggregate_signature.return_value,
+        mock_mute_task.assert_called_once_with(
+            kwargs={
+                "tenant_id": str(finding.tenant_id),
+                "mute_rule_id": response_data["id"],
+                "provider_ids": [str(finding.scan.provider_id)],
+            }
         )
-        mock_chain.return_value.apply_async.assert_called_once()
 
-    @patch("tasks.tasks.mute_historical_findings_task.apply_async")
+    @patch("api.v1.views.mute_findings_in_latest_scans_task.apply_async")
     def test_mute_rules_create_converts_finding_ids_to_uids(
         self,
         mock_task,
@@ -18425,7 +18416,7 @@ class TestMuteRuleViewSet:
         ]
         assert set(mute_rule.finding_uids) == set(expected_uids)
 
-    @patch("tasks.tasks.mute_historical_findings_task.apply_async")
+    @patch("api.v1.views.mute_findings_in_latest_scans_task.apply_async")
     def test_mute_rules_deduplicates_uids(
         self,
         mock_task,
@@ -18492,10 +18483,10 @@ class TestMuteRuleViewSet:
 
         finding1.refresh_from_db()
         finding2.refresh_from_db()
-        assert finding1.muted is True
-        assert finding2.muted is True
+        assert finding1.muted is False
+        assert finding2.muted is False
 
-    @patch("tasks.tasks.mute_historical_findings_task.apply_async")
+    @patch("api.v1.views.mute_findings_in_latest_scans_task.apply_async")
     def test_mute_rules_create_overlap_detection_active(
         self,
         mock_task,
@@ -18528,7 +18519,7 @@ class TestMuteRuleViewSet:
             "already muted" in error_detail.lower() or "overlap" in error_detail.lower()
         )
 
-    @patch("tasks.tasks.mute_historical_findings_task.apply_async")
+    @patch("api.v1.views.mute_findings_in_latest_scans_task.apply_async")
     def test_mute_rules_create_no_overlap_with_inactive(
         self,
         mock_task,
@@ -18584,7 +18575,7 @@ class TestMuteRuleViewSet:
             == "/data/attributes/finding_ids"
         )
 
-    @patch("tasks.tasks.mute_historical_findings_task.apply_async")
+    @patch("api.v1.views.mute_findings_in_latest_scans_task.apply_async")
     def test_mute_rules_create_invalid_finding_ids(
         self, mock_task, authenticated_client
     ):

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

@@ -244,7 +244,6 @@ from api.v1.serializers import (
     UserUpdateSerializer,
 )
 from botocore.exceptions import ClientError, NoCredentialsError, ParamValidationError
-from celery import chain
 from celery.result import AsyncResult
 from config.custom_logging import BackendLogger
 from config.env import env
@@ -342,8 +341,7 @@ from tasks.tasks import (
     enqueue_scan_execution_on_commit,
     get_active_provider_scan,
     jira_integration_task,
-    mute_historical_findings_task,
-    reaggregate_all_finding_group_summaries_task,
+    mute_findings_in_latest_scans_task,
     refresh_lighthouse_provider_models_task,
 )
 
@@ -7551,35 +7549,28 @@ class MuteRuleViewSet(BaseRLSViewSet):
         serializer = self.get_serializer(data=request.data)
         serializer.is_valid(raise_exception=True)
 
-        # Create the mute rule
+        tenant_id = str(request.tenant_id)
+        finding_ids = serializer.validated_data["finding_ids"]
+        provider_ids = list(
+            dict.fromkeys(
+                Finding.all_objects.filter(
+                    id__in=finding_ids, tenant_id=tenant_id
+                ).values_list("scan__provider_id", flat=True)
+            )
+        )
+
         mute_rule = serializer.save()
 
-        tenant_id = str(request.tenant_id)
-        finding_ids = request.data.get("finding_ids", [])
-
-        # Immediately mute the selected findings
-        Finding.all_objects.filter(
-            id__in=finding_ids, tenant_id=tenant_id, muted=False
-        ).update(
-            muted=True,
-            muted_at=mute_rule.inserted_at,
-            muted_reason=mute_rule.reason,
-        )
-
-        # Launch background task for historical muting + reaggregation
         transaction.on_commit(
-            lambda: chain(
-                mute_historical_findings_task.si(
-                    tenant_id=tenant_id,
-                    mute_rule_id=str(mute_rule.id),
-                ),
-                reaggregate_all_finding_group_summaries_task.si(
-                    tenant_id=tenant_id,
-                ),
-            ).apply_async()
+            lambda: mute_findings_in_latest_scans_task.apply_async(
+                kwargs={
+                    "tenant_id": tenant_id,
+                    "mute_rule_id": str(mute_rule.id),
+                    "provider_ids": [str(provider_id) for provider_id in provider_ids],
+                }
+            )
         )
 
-        # Return the created mute rule
         serializer = self.get_serializer(mute_rule)
         return Response(
             data=serializer.data,

api/src/backend/config/django/base.py

@@ -303,6 +303,11 @@ SECURE_REFERRER_POLICY = "strict-origin-when-cross-origin"
 
 DJANGO_DELETION_BATCH_SIZE = env.int("DJANGO_DELETION_BATCH_SIZE", 5000)
 
+# Public base URL of the Prowler UI (for example https://cloud.prowler.com). Used to
+# build links back to findings in outbound integrations such as Jira. Empty by
+# default, so self-hosted deployments emit no links unless they configure it.
+UI_BASE_URL = env.str("DJANGO_UI_BASE_URL", "").rstrip("/")
+
 # SAML requirement
 CSRF_COOKIE_SECURE = True
 SESSION_COOKIE_SECURE = True

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

@@ -2,13 +2,16 @@ import os
 import time
 from datetime import UTC, datetime
 from glob import glob
+from urllib.parse import quote
 
 from api.db_router import READ_REPLICA_ALIAS, MainRouter
 from api.db_utils import REPLICA_MAX_ATTEMPTS, REPLICA_RETRY_BASE_DELAY, rls_transaction
 from api.models import Finding, Integration, Provider
+from api.rls import Tenant
 from api.utils import initialize_prowler_integration, initialize_prowler_provider
 from celery.utils.log import get_task_logger
 from config.django.base import DJANGO_FINDINGS_BATCH_SIZE
+from django.conf import settings
 from django.db import OperationalError
 from prowler.lib.outputs.asff.asff import ASFF
 from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
@@ -16,6 +19,7 @@ from prowler.lib.outputs.csv.csv import CSV
 from prowler.lib.outputs.finding import Finding as FindingOutput
 from prowler.lib.outputs.html.html import HTML
 from prowler.lib.outputs.jira.exceptions.exceptions import JiraBaseException
+from prowler.lib.outputs.jira.jira import Jira
 from prowler.lib.outputs.ocsf.ocsf import OCSF
 from prowler.providers.aws.aws_provider import AwsProvider
 from prowler.providers.aws.lib.s3.s3 import S3
@@ -477,6 +481,55 @@ def upload_security_hub_integration(
         return False
 
 
+JIRA_LABEL_PREFIX = "prowler"
+
+
+def build_jira_finding_url(finding_uid: str) -> str:
+    """Build the Prowler UI link for a finding, or "" when no UI base URL is set.
+
+    The link filters by the finding ``uid`` rather than the per-scan record id so
+    it keeps resolving after the finding is seen again in later scans.
+    """
+    base_url = getattr(settings, "UI_BASE_URL", "")
+    if not base_url or not finding_uid:
+        return ""
+    return f"{base_url}/findings?filter[uid]={quote(finding_uid, safe='')}"
+
+
+def build_jira_issue_labels(
+    finding_uid: str, provider: str, severity: str, check_id: str
+) -> list[str]:
+    """Build the deterministic label set written to every Jira issue.
+
+    Labels are prefixed to avoid colliding with customer labels and sanitized so
+    Jira never rejects them; the finding-uid label is what lets a ticket be traced
+    back (or JQL-filtered) to its finding.
+    """
+    raw_labels = [
+        JIRA_LABEL_PREFIX,
+        f"{JIRA_LABEL_PREFIX}-{provider}" if provider else "",
+        f"{JIRA_LABEL_PREFIX}-{severity}" if severity else "",
+        f"{JIRA_LABEL_PREFIX}-{check_id}" if check_id else "",
+        Jira.build_finding_label(finding_uid),
+    ]
+    return Jira.sanitize_labels(raw_labels)
+
+
+def get_tenant_name(tenant_id: str) -> str:
+    """Return the tenant name for the Jira issue "Tenant Info" row, or "" if unknown.
+
+    The name is informational only, so a lookup failure must never block the send.
+    """
+    try:
+        return (
+            Tenant.objects.filter(id=tenant_id).values_list("name", flat=True).first()
+            or ""
+        )
+    except Exception:
+        logger.warning("Could not resolve tenant name for %s", tenant_id)
+        return ""
+
+
 def send_findings_to_jira(
     tenant_id: str,
     integration_id: str,
@@ -487,6 +540,7 @@ def send_findings_to_jira(
     with rls_transaction(tenant_id):
         integration = Integration.objects.get(id=integration_id)
         jira_integration = initialize_prowler_integration(integration)
+        tenant_info = get_tenant_name(tenant_id)
 
     num_tickets_created = 0
     error_messages = []
@@ -519,6 +573,15 @@ def send_findings_to_jira(
             recommendation = remediation.get("recommendation", {})
             remediation_code = remediation.get("code", {})
 
+            provider_type = finding_instance.scan.provider.provider
+            issue_labels = build_jira_issue_labels(
+                finding_uid=finding_instance.uid,
+                provider=provider_type,
+                severity=finding_instance.severity,
+                check_id=finding_instance.check_id,
+            )
+            finding_url = build_jira_finding_url(finding_instance.uid)
+
             try:
                 # Send the individual finding to Jira
                 result = jira_integration.send_finding(
@@ -527,7 +590,7 @@ def send_findings_to_jira(
                     severity=finding_instance.severity,
                     status=finding_instance.status,
                     status_extended=finding_instance.status_extended or "",
-                    provider=finding_instance.scan.provider.provider,
+                    provider=provider_type,
                     region=region,
                     resource_uid=resource_uid,
                     resource_name=resource_name,
@@ -542,6 +605,9 @@ def send_findings_to_jira(
                     compliance=finding_instance.compliance or {},
                     project_key=project_key,
                     issue_type=issue_type,
+                    issue_labels=issue_labels,
+                    finding_url=finding_url,
+                    tenant_info=tenant_info,
                 )
             except JiraBaseException as error:
                 error_message = error.message or JIRA_GENERIC_SEND_ERROR
@@ -557,6 +623,11 @@ def send_findings_to_jira(
 
             if result:
                 num_tickets_created += 1
+                logger.info(
+                    "Finding %s sent to Jira as %s",
+                    finding_id,
+                    result.get("key") if isinstance(result, dict) else result,
+                )
             else:
                 error_message = JIRA_GENERIC_SEND_ERROR
                 logger.error(error_message)

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

@@ -1,63 +1,104 @@
+from collections.abc import Iterable
+
 from api.db_utils import rls_transaction
-from api.models import Finding, MuteRule
+from api.models import Finding, MuteRule, Scan, StateChoices
 from celery.utils.log import get_task_logger
-from config.django.base import DJANGO_FINDINGS_BATCH_SIZE
-from tasks.utils import batched
 
 logger = get_task_logger(__name__)
 
 
-def mute_historical_findings(tenant_id: str, mute_rule_id: str):
-    """
-    Mute historical findings that match the given mute rule.
+def _mute_findings_for_rule(
+    *,
+    tenant_id: str,
+    scan_id: str,
+    finding_uids: Iterable[str],
+    muted_at,
+    muted_reason: str,
+) -> int:
+    finding_uids = list(finding_uids)
+    if not finding_uids:
+        return 0
 
-    This function processes findings in batches, updating their muted status
-    and adding the mute reason.
+    return Finding.all_objects.filter(
+        tenant_id=tenant_id,
+        scan_id=scan_id,
+        uid__in=finding_uids,
+        muted=False,
+    ).update(
+        muted=True,
+        muted_at=muted_at,
+        muted_reason=muted_reason,
+    )
 
-    Args:
-        tenant_id (str): The tenant ID for RLS context
-        mute_rule_id (str): The ID of the mute rule to apply
 
-    Returns:
-        dict: Summary of the muting operation with findings_muted count
-    """
-    findings_muted_count = 0
+def mute_findings_in_latest_scans(
+    tenant_id: str, mute_rule_id: str, provider_ids: list[str]
+) -> dict:
+    """Apply a mute rule to the latest completed scan of each provider."""
+    provider_ids = list(dict.fromkeys(provider_ids))
 
-    # Get the list of UIDs to mute and the reason
     with rls_transaction(tenant_id):
         mute_rule = MuteRule.objects.get(id=mute_rule_id, tenant_id=tenant_id)
-        finding_uids = mute_rule.finding_uids
-        mute_reason = mute_rule.reason
-        muted_at = mute_rule.inserted_at
-
-    # Query findings that match the UIDs and are not already muted
-    with rls_transaction(tenant_id):
-        findings_to_mute = Finding.objects.filter(
-            tenant_id=tenant_id, uid__in=finding_uids, muted=False
-        )
-        total_findings = findings_to_mute.count()
-
-        logger.info(
-            f"Processing {total_findings} findings for mute rule {mute_rule_id}"
+        latest_scans = list(
+            Scan.objects.filter(
+                tenant_id=tenant_id,
+                provider_id__in=provider_ids,
+                state=StateChoices.COMPLETED,
+                completed_at__isnull=False,
+            )
+            .order_by("provider_id", "-completed_at", "-inserted_at", "-id")
+            .distinct("provider_id")
+            .values_list("id", flat=True)
         )
 
-        if total_findings > 0:
-            for batch, is_last in batched(
-                findings_to_mute.iterator(), DJANGO_FINDINGS_BATCH_SIZE
-            ):
-                batch_ids = [f.id for f in batch]
-                updated_count = Finding.all_objects.filter(
-                    id__in=batch_ids, tenant_id=tenant_id
-                ).update(
-                    muted=True,
-                    muted_at=muted_at,
-                    muted_reason=mute_reason,
-                )
-                findings_muted_count += updated_count
-
-    logger.info(f"Muted {findings_muted_count} findings for rule {mute_rule_id}")
+        changed_scan_ids = []
+        findings_muted = 0
+        for scan_id in latest_scans:
+            updated = _mute_findings_for_rule(
+                tenant_id=tenant_id,
+                scan_id=str(scan_id),
+                finding_uids=mute_rule.finding_uids,
+                muted_at=mute_rule.inserted_at,
+                muted_reason=mute_rule.reason,
+            )
+            if updated:
+                findings_muted += updated
+                changed_scan_ids.append(str(scan_id))
 
+    logger.info(
+        "Muted %d findings in %d latest scans for rule %s",
+        findings_muted,
+        len(changed_scan_ids),
+        mute_rule_id,
+    )
     return {
-        "findings_muted": findings_muted_count,
+        "findings_muted": findings_muted,
         "rule_id": mute_rule_id,
+        "scan_ids": changed_scan_ids,
     }
+
+
+def reconcile_scan_mute_rules(tenant_id: str, scan_id: str) -> dict:
+    """Apply the current enabled mute rules to one completed scan."""
+    findings_muted = 0
+
+    with rls_transaction(tenant_id):
+        mute_rules = MuteRule.objects.filter(tenant_id=tenant_id, enabled=True).values(
+            "finding_uids", "reason", "inserted_at"
+        )
+
+        for mute_rule in mute_rules:
+            findings_muted += _mute_findings_for_rule(
+                tenant_id=tenant_id,
+                scan_id=scan_id,
+                finding_uids=mute_rule["finding_uids"],
+                muted_at=mute_rule["inserted_at"],
+                muted_reason=mute_rule["reason"],
+            )
+
+    logger.info(
+        "Reconciled mute rules for scan %s; muted %d findings",
+        scan_id,
+        findings_muted,
+    )
+    return {"findings_muted": findings_muted, "scan_id": str(scan_id)}

api/src/backend/tasks/tasks.py

@@ -73,7 +73,10 @@ from tasks.jobs.lighthouse_providers import (
     check_lighthouse_provider_connection,
     refresh_lighthouse_provider_models,
 )
-from tasks.jobs.muting import mute_historical_findings
+from tasks.jobs.muting import (
+    mute_findings_in_latest_scans,
+    reconcile_scan_mute_rules,
+)
 from tasks.jobs.orphan_recovery import reconcile_orphans
 from tasks.jobs.report import (
     STALE_TMP_OUTPUT_MAX_AGE_HOURS,
@@ -526,6 +529,7 @@ def perform_scan_task(
             provider_id=provider_id,
             checks_to_execute=checks_to_execute,
         )
+        reconcile_scan_mute_rules(tenant_id, scan_id)
         _perform_scan_complete_tasks(tenant_id, scan_id, provider_id)
         return result
     finally:
@@ -635,6 +639,7 @@ def perform_scheduled_scan_task(self, tenant_id: str, provider_id: str):
             scan_id=str(scan_instance.id),
             provider_id=provider_id,
         )
+        reconcile_scan_mute_rules(tenant_id, str(scan_instance.id))
         _perform_scan_complete_tasks(tenant_id, str(scan_instance.id), provider_id)
         return result
     finally:
@@ -1188,85 +1193,48 @@ def aggregate_finding_group_summaries_task(tenant_id: str, scan_id: str):
     return aggregate_finding_group_summaries(tenant_id=tenant_id, scan_id=scan_id)
 
 
-@shared_task(
-    base=RLSTask, name="reaggregate-all-finding-group-summaries", queue="overview"
-)
-@set_tenant(keep_tenant=True)
-def reaggregate_all_finding_group_summaries_task(tenant_id: str):
-    """Reaggregate every pre-aggregated summary table for this tenant.
+def _dispatch_scan_summary_reaggregation(tenant_id: str, scan_ids: list[str]) -> None:
+    if not scan_ids:
+        return
 
-    Mirrors the unbounded scope of `mute_historical_findings_task`: that task
-    rewrites every Finding row whose UID matches a mute rule, with no time
-    limit. To keep the pre-aggregated tables consistent with that update,
-    this task re-runs the same per-scan aggregation pipeline that scan
-    completion runs on the latest completed scan of every (provider, day)
-    pair, rebuilding the tables that power the read endpoints:
-
-      - `ScanSummary` and `DailySeveritySummary` -> `/overviews/findings`,
-        `/overviews/findings-severity`, `/overviews/services`.
-      - `FindingGroupDailySummary` -> `/finding-groups` and
-        `/finding-groups/latest`.
-      - `ScanGroupSummary` -> `/overviews/resource-groups` (resource
-        inventory).
-      - `ScanCategorySummary` -> `/overviews/categories`.
-      - `AttackSurfaceOverview` -> `/overviews/attack-surfaces`.
-
-    Per-scan pipelines are dispatched in parallel via a Celery group so
-    wallclock scales with the worker pool.
-    """
-    completed_scans = list(
-        Scan.objects.filter(
-            tenant_id=tenant_id,
-            state=StateChoices.COMPLETED,
-            completed_at__isnull=False,
-        )
-        .order_by("-completed_at")
-        .values("id", "completed_at", "provider_id")
+    logger.info(
+        "Reaggregating overview/finding summaries for %d latest scans",
+        len(scan_ids),
     )
-
-    # Keep the latest scan per (provider, day) pair so the daily summary row
-    # the aggregator writes is the most recent snapshot of that day for that
-    # provider. Iterating from most recent to oldest means the first scan we
-    # see for a given key wins.
-    latest_scans: dict[tuple, str] = {}
-    for scan in completed_scans:
-        key = (scan["provider_id"], scan["completed_at"].date())
-        if key not in latest_scans:
-            latest_scans[key] = str(scan["id"])
-
-    scan_ids = list(latest_scans.values())
-    if scan_ids:
-        logger.info(
-            "Reaggregating overview/finding summaries for %d scans (provider x day)",
-            len(scan_ids),
-        )
-        # DailySeveritySummary reads from ScanSummary, so ScanSummary must be
-        # recomputed first; the other aggregators read Finding directly and
-        # can run in parallel with the severity step.
-        group(
-            chain(
-                perform_scan_summary_task.si(tenant_id=tenant_id, scan_id=scan_id),
-                group(
-                    aggregate_daily_severity_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
-                    aggregate_finding_group_summaries_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
-                    aggregate_scan_resource_group_summaries_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
-                    aggregate_scan_category_summaries_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
-                    aggregate_attack_surface_task.si(
-                        tenant_id=tenant_id, scan_id=scan_id
-                    ),
+    group(
+        chain(
+            perform_scan_summary_task.si(tenant_id=tenant_id, scan_id=scan_id),
+            group(
+                aggregate_daily_severity_task.si(tenant_id=tenant_id, scan_id=scan_id),
+                aggregate_finding_group_summaries_task.si(
+                    tenant_id=tenant_id, scan_id=scan_id
                 ),
-            )
-            for scan_id in scan_ids
-        ).apply_async()
-    return {"scans_reaggregated": len(scan_ids)}
+                aggregate_scan_resource_group_summaries_task.si(
+                    tenant_id=tenant_id, scan_id=scan_id
+                ),
+                aggregate_scan_category_summaries_task.si(
+                    tenant_id=tenant_id, scan_id=scan_id
+                ),
+                aggregate_attack_surface_task.si(tenant_id=tenant_id, scan_id=scan_id),
+            ),
+        )
+        for scan_id in scan_ids
+    ).apply_async()
+
+
+@shared_task(base=RLSTask, name="findings-mute-latest-scans", queue="overview")
+@set_tenant(keep_tenant=True)
+def mute_findings_in_latest_scans_task(
+    tenant_id: str, mute_rule_id: str, provider_ids: list[str]
+):
+    """Apply a mute rule to current scans and rebuild only changed summaries."""
+    result = mute_findings_in_latest_scans(
+        tenant_id=tenant_id,
+        mute_rule_id=mute_rule_id,
+        provider_ids=provider_ids,
+    )
+    _dispatch_scan_summary_reaggregation(tenant_id, result["scan_ids"])
+    return result
 
 
 @shared_task(base=RLSTask, name="lighthouse-connection-check")
@@ -1467,25 +1435,3 @@ def generate_compliance_reports_task(tenant_id: str, scan_id: str, provider_id:
         generate_csa=True,
         generate_cis=True,
     )
-
-
-@shared_task(name="findings-mute-historical")
-def mute_historical_findings_task(tenant_id: str, mute_rule_id: str):
-    """
-    Background task to mute all historical findings matching a mute rule.
-
-    This task processes findings in batches to avoid memory issues with large datasets.
-    It updates the Finding.muted, Finding.muted_at, and Finding.muted_reason fields
-    for all findings whose UID is in the mute rule's finding_uids list.
-
-    Args:
-        tenant_id (str): The tenant ID for RLS context.
-        mute_rule_id (str): The primary key of the MuteRule to apply.
-
-    Returns:
-        dict: A dictionary containing:
-            - 'findings_muted' (int): Total number of findings muted.
-            - 'rule_id' (str): The mute rule ID.
-            - 'status' (str): Final status ('completed').
-    """
-    return mute_historical_findings(tenant_id, mute_rule_id)

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

@@ -6,15 +6,20 @@ from api.db_router import READ_REPLICA_ALIAS, MainRouter
 from api.models import Integration
 from api.utils import prowler_integration_connection_test
 from django.db import OperationalError
+from django.test import override_settings
 from prowler.lib.outputs.jira.exceptions.exceptions import (
     JiraRefreshTokenError,
     JiraRequiredCustomFieldsError,
 )
+from prowler.lib.outputs.jira.jira import Jira
 from prowler.providers.aws.lib.security_hub.security_hub import SecurityHubConnection
 from prowler.providers.common.models import Connection
 from tasks.jobs.integrations import (
+    build_jira_finding_url,
+    build_jira_issue_labels,
     get_s3_client_from_integration,
     get_security_hub_client_from_integration,
+    get_tenant_name,
     send_findings_to_jira,
     upload_s3_integration,
     upload_security_hub_integration,
@@ -1696,6 +1701,7 @@ class TestJiraIntegration:
 
         finding1 = MagicMock()
         finding1.id = "finding-1"
+        finding1.uid = "prowler-aws-check_001-123456789012-us-east-1-my bucket"
         finding1.check_id = "check_001"
         finding1.severity = "high"
         finding1.status = "FAIL"
@@ -1724,6 +1730,7 @@ class TestJiraIntegration:
 
         finding2 = MagicMock()
         finding2.id = "finding-2"
+        finding2.uid = "prowler-azure-check_002-sub/resource"
         finding2.check_id = "check_002"
         finding2.severity = "medium"
         finding2.status = "PASS"
@@ -1748,9 +1755,13 @@ class TestJiraIntegration:
         ]
 
         # Call the function
-        result = send_findings_to_jira(
-            tenant_id, integration_id, project_key, issue_type, finding_ids
-        )
+        with (
+            override_settings(UI_BASE_URL="https://cloud.example.com"),
+            patch("tasks.jobs.integrations.get_tenant_name", return_value="Acme"),
+        ):
+            result = send_findings_to_jira(
+                tenant_id, integration_id, project_key, issue_type, finding_ids
+            )
 
         # Assertions
         assert result == {"created_count": 2, "failed_count": 0}
@@ -1773,12 +1784,36 @@ class TestJiraIntegration:
         assert first_call.kwargs["provider"] == "aws"
         assert first_call.kwargs["project_key"] == project_key
         assert first_call.kwargs["issue_type"] == issue_type
+        # Finding reference: labels, link back and tenant info
+        assert first_call.kwargs["issue_labels"] == [
+            "prowler",
+            "prowler-aws",
+            "prowler-high",
+            "prowler-check_001",
+            "prowler-finding-prowler-aws-check_001-123456789012-us-east-1-my_bucket",
+        ]
+        assert first_call.kwargs["finding_url"] == (
+            "https://cloud.example.com/findings?filter[uid]="
+            "prowler-aws-check_001-123456789012-us-east-1-my%20bucket"
+        )
+        assert first_call.kwargs["tenant_info"] == "Acme"
 
         # Verify second call
         second_call = mock_jira_integration.send_finding.call_args_list[1]
         assert second_call.kwargs["check_id"] == "check_002"
         assert second_call.kwargs["severity"] == "medium"
         assert second_call.kwargs["status"] == "PASS"
+        assert second_call.kwargs["issue_labels"] == [
+            "prowler",
+            "prowler-azure",
+            "prowler-medium",
+            "prowler-check_002",
+            "prowler-finding-prowler-azure-check_002-sub/resource",
+        ]
+        assert second_call.kwargs["finding_url"] == (
+            "https://cloud.example.com/findings?filter[uid]="
+            "prowler-azure-check_002-sub%2Fresource"
+        )
 
     @patch("tasks.jobs.integrations.rls_transaction")
     @patch("tasks.jobs.integrations.Finding")
@@ -2200,3 +2235,101 @@ class TestJiraIntegration:
         assert call_kwargs["remediation_code_cli"] == ""
         assert call_kwargs["remediation_code_other"] == ""
         assert call_kwargs["compliance"] == {}
+
+
+class TestJiraFindingReference:
+    """Helpers that give Jira issues a stable reference back to the finding."""
+
+    def test_build_jira_issue_labels(self):
+        assert build_jira_issue_labels(
+            finding_uid="prowler-aws-check-123-eu-west-1-hub/unknown",
+            provider="aws",
+            severity="critical",
+            check_id="iam_root_mfa",
+        ) == [
+            "prowler",
+            "prowler-aws",
+            "prowler-critical",
+            "prowler-iam_root_mfa",
+            "prowler-finding-prowler-aws-check-123-eu-west-1-hub/unknown",
+        ]
+
+    def test_build_jira_issue_labels_skips_empty_parts(self):
+        assert build_jira_issue_labels(
+            finding_uid="", provider="", severity="", check_id=""
+        ) == ["prowler"]
+
+    def test_build_jira_issue_labels_sanitizes_metadata(self):
+        assert build_jira_issue_labels(
+            finding_uid=" uid\x00 with spaces ",
+            provider="aws cloud",
+            severity="high severity",
+            check_id="check id",
+        ) == [
+            "prowler",
+            "prowler-aws_cloud",
+            "prowler-high_severity",
+            "prowler-check_id",
+            "prowler-finding-uid_with_spaces",
+        ]
+
+    def test_build_jira_issue_labels_preserves_maximum_length_uid(self):
+        finding_uid = "u" * (Jira.LABEL_MAX_LENGTH - len(Jira.FINDING_LABEL_PREFIX) - 1)
+        finding_label = build_jira_issue_labels(
+            finding_uid=finding_uid,
+            provider="gcp",
+            severity="low",
+            check_id="check",
+        )[-1]
+
+        assert finding_label == f"{Jira.FINDING_LABEL_PREFIX}-{finding_uid}"
+        assert len(finding_label) == Jira.LABEL_MAX_LENGTH
+
+    def test_build_jira_issue_labels_distinguishes_long_uids(self):
+        common_prefix = "u" * 300
+        first_uid = f"{common_prefix}-first"
+        second_uid = f"{common_prefix}-second"
+
+        first_label = build_jira_issue_labels(
+            finding_uid=first_uid,
+            provider="gcp",
+            severity="low",
+            check_id="check",
+        )[-1]
+        second_label = build_jira_issue_labels(
+            finding_uid=second_uid,
+            provider="gcp",
+            severity="low",
+            check_id="check",
+        )[-1]
+
+        assert first_label == Jira.build_finding_label(first_uid)
+        assert second_label == Jira.build_finding_label(second_uid)
+        assert first_label != second_label
+        assert len(first_label) == Jira.LABEL_MAX_LENGTH
+        assert len(second_label) == Jira.LABEL_MAX_LENGTH
+
+    @override_settings(UI_BASE_URL="")
+    def test_build_jira_finding_url_without_base_url(self):
+        assert build_jira_finding_url("prowler-aws-check-1") == ""
+
+    @override_settings(UI_BASE_URL="https://cloud.example.com")
+    def test_build_jira_finding_url_with_base_url(self):
+        assert build_jira_finding_url("prowler-aws-check-1") == (
+            "https://cloud.example.com/findings?filter[uid]=prowler-aws-check-1"
+        )
+        # uid characters that would break the query string are encoded
+        assert build_jira_finding_url("a/b c&d") == (
+            "https://cloud.example.com/findings?filter[uid]=a%2Fb%20c%26d"
+        )
+        assert build_jira_finding_url("") == ""
+
+    @pytest.mark.django_db
+    def test_get_tenant_name(self, tenants_fixture):
+        tenant = tenants_fixture[0]
+        assert get_tenant_name(str(tenant.id)) == tenant.name
+
+    @pytest.mark.django_db
+    def test_get_tenant_name_unknown_or_invalid(self):
+        assert get_tenant_name("00000000-0000-0000-0000-000000000000") == ""
+        assert get_tenant_name("not-a-uuid") == ""

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

@@ -1,531 +1,205 @@
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from uuid import uuid4
 
 import pytest
-from api.models import Finding, MuteRule
-from django.core.exceptions import ObjectDoesNotExist
+from api.models import Finding, MuteRule, Scan, StateChoices
 from prowler.lib.check.models import Severity
 from prowler.lib.outputs.finding import Status
-from tasks.jobs.muting import mute_historical_findings
+from tasks.jobs.muting import (
+    mute_findings_in_latest_scans,
+    reconcile_scan_mute_rules,
+)
+
+
+def _create_finding(scan: Scan, uid: str) -> Finding:
+    return Finding.objects.create(
+        tenant_id=scan.tenant_id,
+        uid=uid,
+        scan=scan,
+        status=Status.FAIL,
+        status_extended="Test finding",
+        impact=Severity.high,
+        severity=Severity.high,
+        raw_result={},
+        check_id="test_check",
+        check_metadata={"CheckId": "test_check"},
+        muted=False,
+    )
+
+
+def _create_mute_rule(tenant_id, user, finding_uids, *, enabled=True) -> MuteRule:
+    return MuteRule.objects.create(
+        tenant_id=tenant_id,
+        name=f"Mute rule {uuid4()}",
+        reason="Approved exception",
+        enabled=enabled,
+        created_by=user,
+        finding_uids=finding_uids,
+    )
 
 
 @pytest.mark.django_db
-class TestMuteHistoricalFindings:
-    """
-    Test suite for the mute_historical_findings function.
+class TestMuteFindingsInLatestScans:
+    def test_mutes_latest_scan_and_leaves_older_scan_unchanged(
+        self, scans_fixture, create_test_user
+    ):
+        latest_scan = scans_fixture[0]
+        older_scan = Scan.objects.create(
+            tenant_id=latest_scan.tenant_id,
+            provider=latest_scan.provider,
+            name="Older scan",
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            started_at=datetime.now(UTC) - timedelta(days=1),
+            completed_at=datetime.now(UTC) - timedelta(days=1),
+        )
+        uid = "latest-scan-only"
+        older_finding = _create_finding(older_scan, uid)
+        latest_finding = _create_finding(latest_scan, uid)
+        mute_rule = _create_mute_rule(latest_scan.tenant_id, create_test_user, [uid])
 
-    This class tests the batch processing of findings to update their muted status
-    based on MuteRule criteria.
-    """
+        result = mute_findings_in_latest_scans(
+            str(latest_scan.tenant_id),
+            str(mute_rule.id),
+            [str(latest_scan.provider_id)],
+        )
 
-    @pytest.fixture(scope="function")
-    def test_user(self, create_test_user):
-        """Create a test user for mute rule creation."""
-        return create_test_user
+        older_finding.refresh_from_db()
+        latest_finding.refresh_from_db()
+        assert older_finding.muted is False
+        assert latest_finding.muted is True
+        assert latest_finding.muted_at == mute_rule.inserted_at
+        assert latest_finding.muted_reason == mute_rule.reason
+        assert result == {
+            "findings_muted": 1,
+            "rule_id": str(mute_rule.id),
+            "scan_ids": [str(latest_scan.id)],
+        }
 
-    @pytest.fixture(scope="function")
-    def mute_rule_with_findings(self, tenants_fixture, findings_fixture, test_user):
-        """
-        Create a mute rule that targets the first finding in the fixture.
-        """
+    def test_mutes_one_latest_scan_per_provider(self, scans_fixture, create_test_user):
+        first_scan, second_scan, _ = scans_fixture
+        uid = "shared-selected-uid"
+        first_finding = _create_finding(first_scan, uid)
+        second_finding = _create_finding(second_scan, uid)
+        mute_rule = _create_mute_rule(first_scan.tenant_id, create_test_user, [uid])
+
+        result = mute_findings_in_latest_scans(
+            str(first_scan.tenant_id),
+            str(mute_rule.id),
+            [str(first_scan.provider_id), str(second_scan.provider_id)],
+        )
+
+        first_finding.refresh_from_db()
+        second_finding.refresh_from_db()
+        assert first_finding.muted is True
+        assert second_finding.muted is True
+        assert result["findings_muted"] == 2
+        assert set(result["scan_ids"]) == {str(first_scan.id), str(second_scan.id)}
+
+    def test_provider_without_completed_scan_does_nothing(
+        self, tenants_fixture, provider_factory, create_test_user
+    ):
         tenant = tenants_fixture[0]
-        finding = findings_fixture[0]
-
-        mute_rule = MuteRule.objects.create(
-            tenant_id=tenant.id,
-            name="Test Mute Rule",
-            reason="Testing mute functionality",
-            enabled=True,
-            created_by=test_user,
-            finding_uids=[finding.uid],
+        provider = provider_factory()
+        mute_rule = _create_mute_rule(
+            tenant.id, create_test_user, ["future-scan-finding"]
         )
 
-        return mute_rule
+        result = mute_findings_in_latest_scans(
+            str(tenant.id), str(mute_rule.id), [str(provider.id)]
+        )
 
-    @pytest.fixture(scope="function")
-    def mute_rule_multiple_findings(self, scans_fixture, test_user):
-        """
-        Create multiple unmuted findings and a mute rule targeting all of them.
-        """
+        assert result == {
+            "findings_muted": 0,
+            "rule_id": str(mute_rule.id),
+            "scan_ids": [],
+        }
+
+    def test_retry_does_not_report_changed_scans_twice(
+        self, scans_fixture, create_test_user
+    ):
         scan = scans_fixture[0]
-        tenant_id = scan.tenant_id
-
-        # Create 5 unmuted findings
-        finding_uids = []
-        for i in range(5):
-            finding = Finding.objects.create(
-                tenant_id=tenant_id,
-                uid=f"test_finding_uid_mute_{i}",
-                scan=scan,
-                status=Status.FAIL,
-                status_extended=f"Test status {i}",
-                impact=Severity.high,
-                severity=Severity.high,
-                raw_result={
-                    "status": Status.FAIL,
-                    "impact": Severity.high,
-                    "severity": Severity.high,
-                },
-                check_id=f"test_check_id_{i}",
-                check_metadata={
-                    "CheckId": f"test_check_id_{i}",
-                    "Description": f"Test description {i}",
-                },
-                muted=False,
-            )
-            finding_uids.append(finding.uid)
-
-        # Create mute rule targeting all findings
-        mute_rule = MuteRule.objects.create(
-            tenant_id=tenant_id,
-            name="Test Multiple Findings Mute Rule",
-            reason="Testing batch muting",
-            enabled=True,
-            created_by=test_user,
-            finding_uids=finding_uids,
+        finding = _create_finding(scan, "idempotent-mute")
+        mute_rule = _create_mute_rule(scan.tenant_id, create_test_user, [finding.uid])
+        args = (
+            str(scan.tenant_id),
+            str(mute_rule.id),
+            [str(scan.provider_id)],
         )
 
-        return mute_rule, finding_uids
+        first_result = mute_findings_in_latest_scans(*args)
+        second_result = mute_findings_in_latest_scans(*args)
 
-    @pytest.fixture(scope="function")
-    def mute_rule_already_muted(self, findings_fixture, test_user):
-        """
-        Create a mute rule that targets an already-muted finding.
-        """
-        tenant_id = findings_fixture[1].tenant_id
-        already_muted_finding = findings_fixture[1]
+        assert first_result["scan_ids"] == [str(scan.id)]
+        assert second_result["findings_muted"] == 0
+        assert second_result["scan_ids"] == []
 
-        mute_rule = MuteRule.objects.create(
-            tenant_id=tenant_id,
-            name="Test Already Muted Rule",
-            reason="Testing already muted findings",
-            enabled=True,
-            created_by=test_user,
-            finding_uids=[already_muted_finding.uid],
+    def test_does_not_cross_tenant_boundary(
+        self, tenants_fixture, provider_factory, create_test_user
+    ):
+        tenant = tenants_fixture[0]
+        other_tenant = tenants_fixture[2]
+        other_provider = provider_factory(tenant=other_tenant)
+        other_scan = Scan.objects.create(
+            tenant_id=other_tenant.id,
+            provider=other_provider,
+            name="Other tenant scan",
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            started_at=datetime.now(UTC),
+            completed_at=datetime.now(UTC),
+        )
+        other_finding = _create_finding(other_scan, "tenant-isolated-uid")
+        mute_rule = _create_mute_rule(tenant.id, create_test_user, [other_finding.uid])
+
+        result = mute_findings_in_latest_scans(
+            str(tenant.id), str(mute_rule.id), [str(other_provider.id)]
         )
 
-        return mute_rule
+        other_finding.refresh_from_db()
+        assert other_finding.muted is False
+        assert result["scan_ids"] == []
 
-    @pytest.fixture(scope="function")
-    def mute_rule_mixed_findings(self, scans_fixture, test_user):
-        """
-        Create a mute rule with a mix of muted and unmuted findings.
-        """
+    def test_nonexistent_rule_raises(self, tenants_fixture):
+        with pytest.raises(MuteRule.DoesNotExist):
+            mute_findings_in_latest_scans(str(tenants_fixture[0].id), str(uuid4()), [])
+
+
+@pytest.mark.django_db
+class TestReconcileScanMuteRules:
+    def test_applies_only_enabled_rules_to_requested_scan(
+        self, scans_fixture, create_test_user
+    ):
         scan = scans_fixture[0]
-        tenant_id = scan.tenant_id
-
-        # Create 3 unmuted findings
-        unmuted_uids = []
-        for i in range(3):
-            finding = Finding.objects.create(
-                tenant_id=tenant_id,
-                uid=f"unmuted_finding_{i}",
-                scan=scan,
-                status=Status.FAIL,
-                status_extended=f"Unmuted status {i}",
-                impact=Severity.medium,
-                severity=Severity.medium,
-                raw_result={
-                    "status": Status.FAIL,
-                    "impact": Severity.medium,
-                    "severity": Severity.medium,
-                },
-                check_id=f"unmuted_check_{i}",
-                check_metadata={
-                    "CheckId": f"unmuted_check_{i}",
-                    "Description": f"Unmuted description {i}",
-                },
-                muted=False,
-            )
-            unmuted_uids.append(finding.uid)
-
-        # Create 2 already muted findings
-        muted_uids = []
-        for i in range(2):
-            finding = Finding.objects.create(
-                tenant_id=tenant_id,
-                uid=f"muted_finding_{i}",
-                scan=scan,
-                status=Status.FAIL,
-                status_extended=f"Muted status {i}",
-                impact=Severity.low,
-                severity=Severity.low,
-                raw_result={
-                    "status": Status.FAIL,
-                    "impact": Severity.low,
-                    "severity": Severity.low,
-                },
-                check_id=f"muted_check_{i}",
-                check_metadata={
-                    "CheckId": f"muted_check_{i}",
-                    "Description": f"Muted description {i}",
-                },
-                muted=True,
-                muted_at=datetime.now(UTC),
-                muted_reason="Already muted",
-            )
-            muted_uids.append(finding.uid)
-
-        # Create mute rule targeting all findings
-        all_uids = unmuted_uids + muted_uids
-        mute_rule = MuteRule.objects.create(
-            tenant_id=tenant_id,
-            name="Test Mixed Findings Rule",
-            reason="Testing mixed muted/unmuted findings",
-            enabled=True,
-            created_by=test_user,
-            finding_uids=all_uids,
+        active_finding = _create_finding(scan, "active-rule-uid")
+        disabled_finding = _create_finding(scan, "disabled-rule-uid")
+        active_rule = _create_mute_rule(
+            scan.tenant_id, create_test_user, [active_finding.uid]
         )
-
-        return mute_rule, unmuted_uids, muted_uids
-
-    @pytest.fixture(scope="function")
-    def mute_rule_batch_test(self, scans_fixture, test_user):
-        """
-        Create enough findings to test batch processing (>1000 for default batch size).
-        """
-        scan = scans_fixture[0]
-        tenant_id = scan.tenant_id
-
-        # Create 1500 findings to exceed default batch size of 1000
-        finding_uids = []
-        for i in range(1500):
-            finding = Finding.objects.create(
-                tenant_id=tenant_id,
-                uid=f"batch_test_finding_{i}",
-                scan=scan,
-                status=Status.FAIL,
-                status_extended=f"Batch test status {i}",
-                impact=Severity.critical,
-                severity=Severity.critical,
-                raw_result={
-                    "status": Status.FAIL,
-                    "impact": Severity.critical,
-                    "severity": Severity.critical,
-                },
-                check_id=f"batch_test_check_{i}",
-                check_metadata={
-                    "CheckId": f"batch_test_check_{i}",
-                    "Description": f"Batch test description {i}",
-                },
-                muted=False,
-            )
-            finding_uids.append(finding.uid)
-
-        # Create mute rule targeting all findings
-        mute_rule = MuteRule.objects.create(
-            tenant_id=tenant_id,
-            name="Test Batch Processing Rule",
-            reason="Testing batch processing functionality",
-            enabled=True,
-            created_by=test_user,
-            finding_uids=finding_uids,
+        _create_mute_rule(
+            scan.tenant_id,
+            create_test_user,
+            [disabled_finding.uid],
+            enabled=False,
         )
-
-        return mute_rule, finding_uids
-
-    def test_mute_historical_findings_single_finding(
-        self, mute_rule_with_findings, findings_fixture
-    ):
-        """
-        Test muting a single historical finding.
-        """
-        mute_rule = mute_rule_with_findings
-        tenant_id = str(mute_rule.tenant_id)
-        finding = findings_fixture[0]
-
-        # Ensure the finding is not muted before execution
-        finding.refresh_from_db()
-        assert finding.muted is False
-        assert finding.muted_at is None
-        assert finding.muted_reason is None
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify return value
-        assert result["findings_muted"] == 1
-        assert result["rule_id"] == str(mute_rule.id)
-
-        # Verify the finding was muted
-        finding.refresh_from_db()
-        assert finding.muted is True
-        assert finding.muted_at == mute_rule.inserted_at
-        assert finding.muted_reason == mute_rule.reason
-
-    def test_mute_historical_findings_multiple_findings(
-        self, mute_rule_multiple_findings
-    ):
-        """
-        Test muting multiple historical findings.
-        """
-        mute_rule, finding_uids = mute_rule_multiple_findings
-        tenant_id = str(mute_rule.tenant_id)
-
-        # Verify all findings are unmuted
-        findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=finding_uids)
-        assert findings.count() == 5
-        for finding in findings:
-            assert finding.muted is False
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify return value
-        assert result["findings_muted"] == 5
-        assert result["rule_id"] == str(mute_rule.id)
-
-        # Verify all findings were muted
-        findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=finding_uids)
-        for finding in findings:
-            assert finding.muted is True
-            assert finding.muted_at == mute_rule.inserted_at
-            assert finding.muted_reason == mute_rule.reason
-
-    def test_mute_historical_findings_already_muted(
-        self, mute_rule_already_muted, findings_fixture
-    ):
-        """
-        Test that already-muted findings are not counted or updated.
-        """
-        mute_rule = mute_rule_already_muted
-        tenant_id = str(mute_rule.tenant_id)
-        finding = findings_fixture[1]
-
-        # Verify the finding is already muted
-        finding.refresh_from_db()
-        assert finding.muted is True
-        original_muted_at = finding.muted_at
-        original_muted_reason = finding.muted_reason
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify no findings were muted
-        assert result["findings_muted"] == 0
-        assert result["rule_id"] == str(mute_rule.id)
-
-        # Verify the finding's mute status did not change
-        finding.refresh_from_db()
-        assert finding.muted is True
-        assert finding.muted_at == original_muted_at
-        assert finding.muted_reason == original_muted_reason
-
-    def test_mute_historical_findings_mixed_status(self, mute_rule_mixed_findings):
-        """
-        Test muting when some findings are already muted and others are not.
-        """
-        mute_rule, unmuted_uids, muted_uids = mute_rule_mixed_findings
-        tenant_id = str(mute_rule.tenant_id)
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify only unmuted findings were counted
-        assert result["findings_muted"] == 3
-        assert result["rule_id"] == str(mute_rule.id)
-
-        # Verify unmuted findings are now muted
-        unmuted_findings = Finding.objects.filter(
-            tenant_id=tenant_id, uid__in=unmuted_uids
+        older_scan = Scan.objects.create(
+            tenant_id=scan.tenant_id,
+            provider=scan.provider,
+            name="Older matching scan",
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            started_at=datetime.now(UTC) - timedelta(days=1),
+            completed_at=datetime.now(UTC) - timedelta(days=1),
         )
-        for finding in unmuted_findings:
-            assert finding.muted is True
-            assert finding.muted_at == mute_rule.inserted_at
-            assert finding.muted_reason == mute_rule.reason
+        older_finding = _create_finding(older_scan, active_finding.uid)
 
-        # Verify already-muted findings remained unchanged
-        already_muted_findings = Finding.objects.filter(
-            tenant_id=tenant_id, uid__in=muted_uids
-        )
-        for finding in already_muted_findings:
-            assert finding.muted is True
-            assert finding.muted_reason == "Already muted"
+        result = reconcile_scan_mute_rules(str(scan.tenant_id), str(scan.id))
 
-    def test_mute_historical_findings_nonexistent_rule(self, tenants_fixture):
-        """
-        Test that a nonexistent mute rule raises ObjectDoesNotExist.
-        """
-        tenant_id = str(tenants_fixture[0].id)
-        nonexistent_rule_id = str(uuid4())
-
-        with pytest.raises(ObjectDoesNotExist):
-            mute_historical_findings(tenant_id, nonexistent_rule_id)
-
-    def test_mute_historical_findings_no_matching_findings(
-        self, tenants_fixture, test_user
-    ):
-        """
-        Test muting when no findings match the rule's UIDs.
-        """
-        tenant_id = str(tenants_fixture[0].id)
-
-        # Create a mute rule with non-existent finding UIDs
-        mute_rule = MuteRule.objects.create(
-            tenant_id=tenant_id,
-            name="Test No Match Rule",
-            reason="Testing no matching findings",
-            enabled=True,
-            created_by=test_user,
-            finding_uids=[
-                "nonexistent_uid_1",
-                "nonexistent_uid_2",
-                "nonexistent_uid_3",
-            ],
-        )
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify no findings were muted
-        assert result["findings_muted"] == 0
-        assert result["rule_id"] == str(mute_rule.id)
-
-    def test_mute_historical_findings_batch_processing(self, mute_rule_batch_test):
-        """
-        Test that large numbers of findings are processed in batches correctly.
-        """
-        mute_rule, finding_uids = mute_rule_batch_test
-        tenant_id = str(mute_rule.tenant_id)
-
-        # Verify all findings exist and are unmuted
-        findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=finding_uids)
-        assert findings.count() == 1500
-        for finding in findings:
-            assert finding.muted is False
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify return value
-        assert result["findings_muted"] == 1500
-        assert result["rule_id"] == str(mute_rule.id)
-
-        # Verify all findings were muted
-        findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=finding_uids)
-        for finding in findings:
-            assert finding.muted is True
-            assert finding.muted_at == mute_rule.inserted_at
-            assert finding.muted_reason == mute_rule.reason
-
-    def test_mute_historical_findings_preserves_muted_at_timestamp(
-        self, mute_rule_with_findings, findings_fixture
-    ):
-        """
-        Test that muted_at is set to the rule's inserted_at, not the current time.
-        """
-        mute_rule = mute_rule_with_findings
-        tenant_id = str(mute_rule.tenant_id)
-        finding = findings_fixture[0]
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify the finding was muted
-        assert result["findings_muted"] == 1
-
-        # Verify muted_at matches the rule's inserted_at timestamp
-        finding.refresh_from_db()
-        assert finding.muted_at == mute_rule.inserted_at
-        assert finding.muted_at is not None
-
-    def test_mute_historical_findings_partial_match(self, scans_fixture, test_user):
-        """
-        Test muting when only some of the rule's UIDs exist as findings.
-        """
-        scan = scans_fixture[0]
-        tenant_id = str(scan.tenant_id)
-
-        # Create 3 findings
-        existing_uids = []
-        for i in range(3):
-            finding = Finding.objects.create(
-                tenant_id=tenant_id,
-                uid=f"partial_match_finding_{i}",
-                scan=scan,
-                status=Status.FAIL,
-                status_extended=f"Partial match status {i}",
-                impact=Severity.high,
-                severity=Severity.high,
-                raw_result={
-                    "status": Status.FAIL,
-                    "impact": Severity.high,
-                    "severity": Severity.high,
-                },
-                check_id=f"partial_match_check_{i}",
-                check_metadata={
-                    "CheckId": f"partial_match_check_{i}",
-                    "Description": f"Partial match description {i}",
-                },
-                muted=False,
-            )
-            existing_uids.append(finding.uid)
-
-        # Create a mute rule with both existing and non-existing UIDs
-        all_uids = existing_uids + [
-            "nonexistent_uid_1",
-            "nonexistent_uid_2",
-        ]
-        mute_rule = MuteRule.objects.create(
-            tenant_id=tenant_id,
-            name="Test Partial Match Rule",
-            reason="Testing partial matching",
-            enabled=True,
-            created_by=test_user,
-            finding_uids=all_uids,
-        )
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify only existing findings were muted
-        assert result["findings_muted"] == 3
-        assert result["rule_id"] == str(mute_rule.id)
-
-        # Verify the existing findings were muted
-        findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=existing_uids)
-        assert findings.count() == 3
-        for finding in findings:
-            assert finding.muted is True
-            assert finding.muted_at == mute_rule.inserted_at
-            assert finding.muted_reason == mute_rule.reason
-
-    def test_mute_historical_findings_empty_uids(self, tenants_fixture, test_user):
-        """
-        Test muting when the rule has an empty finding_uids array.
-        """
-        tenant_id = str(tenants_fixture[0].id)
-
-        # Create a mute rule with empty finding_uids
-        mute_rule = MuteRule.objects.create(
-            tenant_id=tenant_id,
-            name="Test Empty UIDs Rule",
-            reason="Testing empty UIDs",
-            enabled=True,
-            created_by=test_user,
-            finding_uids=[],
-        )
-
-        # Execute the muting function
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify no findings were muted
-        assert result["findings_muted"] == 0
-        assert result["rule_id"] == str(mute_rule.id)
-
-    def test_mute_historical_findings_return_format(self, mute_rule_with_findings):
-        """
-        Test that the return value has the correct format and fields.
-        """
-        mute_rule = mute_rule_with_findings
-        tenant_id = str(mute_rule.tenant_id)
-
-        result = mute_historical_findings(tenant_id, str(mute_rule.id))
-
-        # Verify return value structure
-        assert isinstance(result, dict)
-        assert "findings_muted" in result
-        assert "rule_id" in result
-        assert isinstance(result["findings_muted"], int)
-        assert isinstance(result["rule_id"], str)
-        assert result["rule_id"] == str(mute_rule.id)
+        active_finding.refresh_from_db()
+        disabled_finding.refresh_from_db()
+        older_finding.refresh_from_db()
+        assert active_finding.muted is True
+        assert active_finding.muted_at == active_rule.inserted_at
+        assert disabled_finding.muted is False
+        assert older_finding.muted is False
+        assert result == {"findings_muted": 1, "scan_id": str(scan.id)}

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

@@ -1,6 +1,6 @@
 import uuid
 from contextlib import contextmanager
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime
 from unittest.mock import MagicMock, patch
 
 import httpx
@@ -33,10 +33,10 @@ from tasks.tasks import (
     check_integrations_task,
     check_lighthouse_provider_connection_task,
     generate_outputs_task,
+    mute_findings_in_latest_scans_task,
     perform_attack_paths_scan_task,
     perform_scan_task,
     perform_scheduled_scan_task,
-    reaggregate_all_finding_group_summaries_task,
     refresh_lighthouse_provider_models_task,
     s3_integration_task,
     security_hub_integration_task,
@@ -2959,6 +2959,7 @@ class TestPerformScheduledScanTask:
         with (
             patch("tasks.tasks.perform_prowler_scan", side_effect=_complete_scan),
             patch("tasks.tasks._perform_scan_complete_tasks"),
+            patch("tasks.tasks.reconcile_scan_mute_rules") as mock_reconcile,
             self._override_task_request(perform_scheduled_scan_task, id=task_id),
         ):
             perform_scheduled_scan_task.run(
@@ -2982,6 +2983,13 @@ class TestPerformScheduledScanTask:
             ).count()
             == 1
         )
+        completed_scan = Scan.objects.get(
+            tenant_id=tenant.id,
+            provider=provider,
+            trigger=Scan.TriggerChoices.SCHEDULED,
+            state=StateChoices.COMPLETED,
+        )
+        mock_reconcile.assert_called_once_with(str(tenant.id), str(completed_scan.id))
         assert (
             Scan.objects.filter(
                 tenant_id=tenant.id,
@@ -3176,7 +3184,10 @@ class TestPerformScanTask:
             task=queued_task,
         )
 
+        events = []
+
         def _complete_scan(tenant_id, scan_id, provider_id, checks_to_execute=None):
+            events.append("scan")
             scan_instance = Scan.objects.get(id=scan_id)
             scan_instance.state = StateChoices.COMPLETED
             scan_instance.save()
@@ -3184,7 +3195,14 @@ class TestPerformScanTask:
 
         with (
             patch("tasks.tasks.perform_prowler_scan", side_effect=_complete_scan),
-            patch("tasks.tasks._perform_scan_complete_tasks"),
+            patch(
+                "tasks.tasks.reconcile_scan_mute_rules",
+                side_effect=lambda *_args: events.append("reconcile"),
+            ),
+            patch(
+                "tasks.tasks._perform_scan_complete_tasks",
+                side_effect=lambda *_args: events.append("summaries"),
+            ),
             patch("tasks.tasks.perform_scan_task.apply_async") as mock_apply_async,
         ):
             with django_capture_on_commit_callbacks(execute=True):
@@ -3196,6 +3214,7 @@ class TestPerformScanTask:
 
         queued_task_result.refresh_from_db()
         assert result == {"status": "ok"}
+        assert events == ["scan", "reconcile", "summaries"]
         assert queued_task_result.status == states.PENDING
         mock_apply_async.assert_called_once_with(
             kwargs={
@@ -3241,10 +3260,7 @@ class TestPerformScanTask:
 
 
 @pytest.mark.django_db
-class TestReaggregateAllFindingGroupSummaries:
-    def setup_method(self):
-        self.tenant_id = str(uuid.uuid4())
-
+class TestMuteFindingsInLatestScansTask:
     @patch("tasks.tasks.chain")
     @patch("tasks.tasks.group")
     @patch("tasks.tasks.aggregate_attack_surface_task")
@@ -3253,10 +3269,10 @@ class TestReaggregateAllFindingGroupSummaries:
     @patch("tasks.tasks.aggregate_finding_group_summaries_task")
     @patch("tasks.tasks.aggregate_daily_severity_task")
     @patch("tasks.tasks.perform_scan_summary_task")
-    @patch("tasks.tasks.Scan.objects.filter")
-    def test_dispatches_subtasks_for_each_provider_per_day(
+    @patch("tasks.tasks.mute_findings_in_latest_scans")
+    def test_reaggregates_only_changed_scans(
         self,
-        mock_scan_filter,
+        mock_mute_findings,
         mock_scan_summary_task,
         mock_daily_severity_task,
         mock_finding_group_task,
@@ -3265,119 +3281,36 @@ class TestReaggregateAllFindingGroupSummaries:
         mock_attack_surface_task,
         mock_group,
         mock_chain,
+        tenants_fixture,
     ):
-        provider_id_1 = uuid.uuid4()
-        provider_id_2 = uuid.uuid4()
-        scan_id_today_p1 = uuid.uuid4()
-        scan_id_yesterday_p1 = uuid.uuid4()
-        scan_id_today_p2 = uuid.uuid4()
-        today = datetime.now(tz=UTC)
-        yesterday = today - timedelta(days=1)
-
-        mock_outer_group_result = MagicMock()
-        # The first `group()` call wraps the inner parallel step; subsequent
-        # calls wrap the outer per-scan generator.
-        mock_group.side_effect = lambda *args, **kwargs: (
-            list(args[0]) if args and hasattr(args[0], "__iter__") else None,
-            mock_outer_group_result,
-        )[1]
-
-        mock_scan_filter.return_value.order_by.return_value.values.return_value = [
-            {
-                "id": scan_id_today_p1,
-                "completed_at": today,
-                "provider_id": provider_id_1,
-            },
-            {
-                "id": scan_id_today_p2,
-                "completed_at": today,
-                "provider_id": provider_id_2,
-            },
-            {
-                "id": scan_id_yesterday_p1,
-                "completed_at": yesterday,
-                "provider_id": provider_id_1,
-            },
-        ]
-
-        result = reaggregate_all_finding_group_summaries_task(tenant_id=self.tenant_id)
-
-        assert result == {"scans_reaggregated": 3}
-        expected_scan_ids = {
-            str(scan_id_today_p1),
-            str(scan_id_today_p2),
-            str(scan_id_yesterday_p1),
+        tenant_id = str(tenants_fixture[0].id)
+        mute_rule_id = str(uuid.uuid4())
+        provider_ids = [str(uuid.uuid4()), str(uuid.uuid4())]
+        scan_ids = [str(uuid.uuid4()), str(uuid.uuid4())]
+        result = {
+            "findings_muted": 2,
+            "rule_id": mute_rule_id,
+            "scan_ids": scan_ids,
         }
-        for task_mock in (
-            mock_scan_summary_task,
-            mock_daily_severity_task,
-            mock_finding_group_task,
-            mock_resource_group_task,
-            mock_category_task,
-            mock_attack_surface_task,
-        ):
-            assert task_mock.si.call_count == 3
-            dispatched = {
-                call.kwargs["scan_id"] for call in task_mock.si.call_args_list
-            }
-            assert dispatched == expected_scan_ids
-            for call in task_mock.si.call_args_list:
-                assert call.kwargs["tenant_id"] == self.tenant_id
-        assert mock_chain.call_count == 3
-        mock_outer_group_result.apply_async.assert_called_once()
-
-    @patch("tasks.tasks.chain")
-    @patch("tasks.tasks.group")
-    @patch("tasks.tasks.aggregate_attack_surface_task")
-    @patch("tasks.tasks.aggregate_scan_category_summaries_task")
-    @patch("tasks.tasks.aggregate_scan_resource_group_summaries_task")
-    @patch("tasks.tasks.aggregate_finding_group_summaries_task")
-    @patch("tasks.tasks.aggregate_daily_severity_task")
-    @patch("tasks.tasks.perform_scan_summary_task")
-    @patch("tasks.tasks.Scan.objects.filter")
-    def test_dedupes_scans_to_latest_per_provider_per_day(
-        self,
-        mock_scan_filter,
-        mock_scan_summary_task,
-        mock_daily_severity_task,
-        mock_finding_group_task,
-        mock_resource_group_task,
-        mock_category_task,
-        mock_attack_surface_task,
-        mock_group,
-        mock_chain,
-    ):
-        """When several scans run on the same day for the same provider, only
-        the latest one is dispatched (matching the daily summary unique key)."""
-        provider_id = uuid.uuid4()
-        latest_scan_today = uuid.uuid4()
-        earlier_scan_today = uuid.uuid4()
-        today_late = datetime.now(tz=UTC)
-        today_early = today_late - timedelta(hours=4)
-
+        mock_mute_findings.return_value = result
         mock_outer_group_result = MagicMock()
         mock_group.side_effect = lambda *args, **kwargs: (
             list(args[0]) if args and hasattr(args[0], "__iter__") else None,
             mock_outer_group_result,
         )[1]
 
-        # Returned ordered by `-completed_at`, so the most recent comes first.
-        mock_scan_filter.return_value.order_by.return_value.values.return_value = [
-            {
-                "id": latest_scan_today,
-                "completed_at": today_late,
-                "provider_id": provider_id,
-            },
-            {
-                "id": earlier_scan_today,
-                "completed_at": today_early,
-                "provider_id": provider_id,
-            },
-        ]
+        task_result = mute_findings_in_latest_scans_task(
+            tenant_id=tenant_id,
+            mute_rule_id=mute_rule_id,
+            provider_ids=provider_ids,
+        )
 
-        result = reaggregate_all_finding_group_summaries_task(tenant_id=self.tenant_id)
-
-        assert result == {"scans_reaggregated": 1}
+        assert task_result == result
+        mock_mute_findings.assert_called_once_with(
+            tenant_id=tenant_id,
+            mute_rule_id=mute_rule_id,
+            provider_ids=provider_ids,
+        )
         for task_mock in (
             mock_scan_summary_task,
             mock_daily_severity_task,
@@ -3386,23 +3319,35 @@ class TestReaggregateAllFindingGroupSummaries:
             mock_category_task,
             mock_attack_surface_task,
         ):
-            task_mock.si.assert_called_once_with(
-                tenant_id=self.tenant_id, scan_id=str(latest_scan_today)
-            )
-        mock_chain.assert_called_once()
+            assert task_mock.si.call_count == 2
+            assert {
+                call.kwargs["scan_id"] for call in task_mock.si.call_args_list
+            } == set(scan_ids)
+        assert mock_chain.call_count == 2
         mock_outer_group_result.apply_async.assert_called_once()
 
     @patch("tasks.tasks.chain")
     @patch("tasks.tasks.group")
-    @patch("tasks.tasks.Scan.objects.filter")
-    def test_no_completed_scans_skips_dispatch(
-        self, mock_scan_filter, mock_group, mock_chain
+    @patch("tasks.tasks.mute_findings_in_latest_scans")
+    def test_skips_reaggregation_when_no_scan_changed(
+        self, mock_mute_findings, mock_group, mock_chain, tenants_fixture
     ):
-        mock_scan_filter.return_value.order_by.return_value.values.return_value = []
+        tenant_id = str(tenants_fixture[0].id)
+        mute_rule_id = str(uuid.uuid4())
+        result = {
+            "findings_muted": 0,
+            "rule_id": mute_rule_id,
+            "scan_ids": [],
+        }
+        mock_mute_findings.return_value = result
 
-        result = reaggregate_all_finding_group_summaries_task(tenant_id=self.tenant_id)
+        task_result = mute_findings_in_latest_scans_task(
+            tenant_id=tenant_id,
+            mute_rule_id=mute_rule_id,
+            provider_ids=[],
+        )
 
-        assert result == {"scans_reaggregated": 0}
+        assert task_result == result
         mock_group.assert_not_called()
         mock_chain.assert_not_called()
 

api/uv.lock

@@ -4835,8 +4835,8 @@ wheels = [
 
 [[package]]
 name = "prowler"
-version = "5.40.0"
-source = { git = "https://github.com/prowler-cloud/prowler.git?rev=master#b6e9967da6bebd6c7b8b237317a2a95e2e0c65bc" }
+version = "5.41.0"
+source = { git = "https://github.com/prowler-cloud/prowler.git?rev=master#f05a490cd74a2c0f11a5d66d8ce29d03fa5c64a2" }
 dependencies = [
     { name = "alibabacloud-actiontrail20200706" },
     { name = "alibabacloud-credentials" },
@@ -4928,9 +4928,12 @@ dependencies = [
     { name = "stackit-iaas" },
     { name = "stackit-objectstorage" },
     { name = "stackit-resourcemanager" },
+    { name = "stackit-ske" },
     { name = "tabulate" },
+    { name = "truststore" },
     { name = "tzlocal" },
     { name = "uuid6" },
+    { name = "zstandard" },
 ]
 
 [[package]]
@@ -6117,6 +6120,21 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/c7/9c/38a74d0f7a89b4320f6d2366fb660638bda8860daa08748b12c713d84381/stackit_resourcemanager-0.8.0-py3-none-any.whl", hash = "sha256:dd04bb8353d041a137c4dcba190beabded7acfaff1bc98b218fce20a99389ebc", size = 81288, upload-time = "2026-05-13T09:43:07.81Z" },
 ]
 
+[[package]]
+name = "stackit-ske"
+version = "1.12.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pydantic" },
+    { name = "python-dateutil" },
+    { name = "requests" },
+    { name = "stackit-core" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/cd/9e/df3ad585cb96d028354f4253568e9879d81bb9395d5ebfa268fa9350e2df/stackit_ske-1.12.0.tar.gz", hash = "sha256:62814279f3b7fb2387648f92d14453a8905ad60115c07579f2741ddb7d1fcc94", size = 37239, upload-time = "2026-06-30T11:18:49.39Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/00/37/dc54fb7185a2d4da37308322ea1a7b992312030b2e37262de4eb4003f5c7/stackit_ske-1.12.0-py3-none-any.whl", hash = "sha256:45bd8084d87f14f818b3d7e824450248c8784ed204ca1b2dc108f491dcbdb1a3", size = 93142, upload-time = "2026-06-30T11:18:48.233Z" },
+]
+
 [[package]]
 name = "statsd"
 version = "4.0.1"
@@ -6225,6 +6243,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/d0/30/dc54f88dd4a2b5dc8a0279bdd7270e735851848b762aeb1c1184ed1f6b14/tqdm-4.67.1-py3-none-any.whl", hash = "sha256:26445eca388f82e72884e0d580d5464cd801a3ea01e63e5601bdff9ba6a48de2", size = 78540, upload-time = "2024-11-24T20:12:19.698Z" },
 ]
 
+[[package]]
+name = "truststore"
+version = "0.10.4"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/a3/1585216310e344e8102c22482f6060c7a6ea0322b63e026372e6dcefcfd6/truststore-0.10.4.tar.gz", hash = "sha256:9d91bd436463ad5e4ee4aba766628dd6cd7010cf3e2461756b3303710eebc301", size = 26169, upload-time = "2025-08-12T18:49:02.73Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/19/97/56608b2249fe206a67cd573bc93cd9896e1efb9e98bce9c163bcdc704b88/truststore-0.10.4-py3-none-any.whl", hash = "sha256:adaeaecf1cbb5f4de3b1959b42d41f6fab57b2b1666adb59e89cb0b53361d981", size = 18660, upload-time = "2025-08-12T18:49:01.46Z" },
+]
+
 [[package]]
 name = "typer"
 version = "0.21.1"
@@ -6621,6 +6648,48 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/4a/81/2f171fbc4222066957e6b9220c4fb9146792540102c37e6d94e5d14aad97/zope_interface-8.2-cp312-cp312-win_amd64.whl", hash = "sha256:845d14e580220ae4544bd4d7eb800f0b6034fe5585fc2536806e0a26c2ee6640", size = 212444, upload-time = "2026-01-09T08:05:25.148Z" },
 ]
 
+[[package]]
+name = "zstandard"
+version = "0.25.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/fd/aa/3e0508d5a5dd96529cdc5a97011299056e14c6505b678fd58938792794b1/zstandard-0.25.0.tar.gz", hash = "sha256:7713e1179d162cf5c7906da876ec2ccb9c3a9dcbdffef0cc7f70c3667a205f0b", size = 711513, upload-time = "2025-09-14T22:15:54.002Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/2a/83/c3ca27c363d104980f1c9cee1101cc8ba724ac8c28a033ede6aab89585b1/zstandard-0.25.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:933b65d7680ea337180733cf9e87293cc5500cc0eb3fc8769f4d3c88d724ec5c", size = 795254, upload-time = "2025-09-14T22:16:26.137Z" },
+    { url = "https://files.pythonhosted.org/packages/ac/4d/e66465c5411a7cf4866aeadc7d108081d8ceba9bc7abe6b14aa21c671ec3/zstandard-0.25.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:a3f79487c687b1fc69f19e487cd949bf3aae653d181dfb5fde3bf6d18894706f", size = 640559, upload-time = "2025-09-14T22:16:27.973Z" },
+    { url = "https://files.pythonhosted.org/packages/12/56/354fe655905f290d3b147b33fe946b0f27e791e4b50a5f004c802cb3eb7b/zstandard-0.25.0-cp311-cp311-manylinux2010_i686.manylinux2014_i686.manylinux_2_12_i686.manylinux_2_17_i686.whl", hash = "sha256:0bbc9a0c65ce0eea3c34a691e3c4b6889f5f3909ba4822ab385fab9057099431", size = 5348020, upload-time = "2025-09-14T22:16:29.523Z" },
+    { url = "https://files.pythonhosted.org/packages/3b/13/2b7ed68bd85e69a2069bcc72141d378f22cae5a0f3b353a2c8f50ef30c1b/zstandard-0.25.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:01582723b3ccd6939ab7b3a78622c573799d5d8737b534b86d0e06ac18dbde4a", size = 5058126, upload-time = "2025-09-14T22:16:31.811Z" },
+    { url = "https://files.pythonhosted.org/packages/c9/dd/fdaf0674f4b10d92cb120ccff58bbb6626bf8368f00ebfd2a41ba4a0dc99/zstandard-0.25.0-cp311-cp311-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:5f1ad7bf88535edcf30038f6919abe087f606f62c00a87d7e33e7fc57cb69fcc", size = 5405390, upload-time = "2025-09-14T22:16:33.486Z" },
+    { url = "https://files.pythonhosted.org/packages/0f/67/354d1555575bc2490435f90d67ca4dd65238ff2f119f30f72d5cde09c2ad/zstandard-0.25.0-cp311-cp311-manylinux2014_s390x.manylinux_2_17_s390x.whl", hash = "sha256:06acb75eebeedb77b69048031282737717a63e71e4ae3f77cc0c3b9508320df6", size = 5452914, upload-time = "2025-09-14T22:16:35.277Z" },
+    { url = "https://files.pythonhosted.org/packages/bb/1f/e9cfd801a3f9190bf3e759c422bbfd2247db9d7f3d54a56ecde70137791a/zstandard-0.25.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:9300d02ea7c6506f00e627e287e0492a5eb0371ec1670ae852fefffa6164b072", size = 5559635, upload-time = "2025-09-14T22:16:37.141Z" },
+    { url = "https://files.pythonhosted.org/packages/21/88/5ba550f797ca953a52d708c8e4f380959e7e3280af029e38fbf47b55916e/zstandard-0.25.0-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:bfd06b1c5584b657a2892a6014c2f4c20e0db0208c159148fa78c65f7e0b0277", size = 5048277, upload-time = "2025-09-14T22:16:38.807Z" },
+    { url = "https://files.pythonhosted.org/packages/46/c0/ca3e533b4fa03112facbe7fbe7779cb1ebec215688e5df576fe5429172e0/zstandard-0.25.0-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:f373da2c1757bb7f1acaf09369cdc1d51d84131e50d5fa9863982fd626466313", size = 5574377, upload-time = "2025-09-14T22:16:40.523Z" },
+    { url = "https://files.pythonhosted.org/packages/12/9b/3fb626390113f272abd0799fd677ea33d5fc3ec185e62e6be534493c4b60/zstandard-0.25.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:6c0e5a65158a7946e7a7affa6418878ef97ab66636f13353b8502d7ea03c8097", size = 4961493, upload-time = "2025-09-14T22:16:43.3Z" },
+    { url = "https://files.pythonhosted.org/packages/cb/d3/23094a6b6a4b1343b27ae68249daa17ae0651fcfec9ed4de09d14b940285/zstandard-0.25.0-cp311-cp311-musllinux_1_2_i686.whl", hash = "sha256:c8e167d5adf59476fa3e37bee730890e389410c354771a62e3c076c86f9f7778", size = 5269018, upload-time = "2025-09-14T22:16:45.292Z" },
+    { url = "https://files.pythonhosted.org/packages/8c/a7/bb5a0c1c0f3f4b5e9d5b55198e39de91e04ba7c205cc46fcb0f95f0383c1/zstandard-0.25.0-cp311-cp311-musllinux_1_2_ppc64le.whl", hash = "sha256:98750a309eb2f020da61e727de7d7ba3c57c97cf6213f6f6277bb7fb42a8e065", size = 5443672, upload-time = "2025-09-14T22:16:47.076Z" },
+    { url = "https://files.pythonhosted.org/packages/27/22/503347aa08d073993f25109c36c8d9f029c7d5949198050962cb568dfa5e/zstandard-0.25.0-cp311-cp311-musllinux_1_2_s390x.whl", hash = "sha256:22a086cff1b6ceca18a8dd6096ec631e430e93a8e70a9ca5efa7561a00f826fa", size = 5822753, upload-time = "2025-09-14T22:16:49.316Z" },
+    { url = "https://files.pythonhosted.org/packages/e2/be/94267dc6ee64f0f8ba2b2ae7c7a2df934a816baaa7291db9e1aa77394c3c/zstandard-0.25.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:72d35d7aa0bba323965da807a462b0966c91608ef3a48ba761678cb20ce5d8b7", size = 5366047, upload-time = "2025-09-14T22:16:51.328Z" },
+    { url = "https://files.pythonhosted.org/packages/7b/a3/732893eab0a3a7aecff8b99052fecf9f605cf0fb5fb6d0290e36beee47a4/zstandard-0.25.0-cp311-cp311-win32.whl", hash = "sha256:f5aeea11ded7320a84dcdd62a3d95b5186834224a9e55b92ccae35d21a8b63d4", size = 436484, upload-time = "2025-09-14T22:16:55.005Z" },
+    { url = "https://files.pythonhosted.org/packages/43/a3/c6155f5c1cce691cb80dfd38627046e50af3ee9ddc5d0b45b9b063bfb8c9/zstandard-0.25.0-cp311-cp311-win_amd64.whl", hash = "sha256:daab68faadb847063d0c56f361a289c4f268706b598afbf9ad113cbe5c38b6b2", size = 506183, upload-time = "2025-09-14T22:16:52.753Z" },
+    { url = "https://files.pythonhosted.org/packages/8c/3e/8945ab86a0820cc0e0cdbf38086a92868a9172020fdab8a03ac19662b0e5/zstandard-0.25.0-cp311-cp311-win_arm64.whl", hash = "sha256:22a06c5df3751bb7dc67406f5374734ccee8ed37fc5981bf1ad7041831fa1137", size = 462533, upload-time = "2025-09-14T22:16:53.878Z" },
+    { url = "https://files.pythonhosted.org/packages/82/fc/f26eb6ef91ae723a03e16eddb198abcfce2bc5a42e224d44cc8b6765e57e/zstandard-0.25.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:7b3c3a3ab9daa3eed242d6ecceead93aebbb8f5f84318d82cee643e019c4b73b", size = 795738, upload-time = "2025-09-14T22:16:56.237Z" },
+    { url = "https://files.pythonhosted.org/packages/aa/1c/d920d64b22f8dd028a8b90e2d756e431a5d86194caa78e3819c7bf53b4b3/zstandard-0.25.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:913cbd31a400febff93b564a23e17c3ed2d56c064006f54efec210d586171c00", size = 640436, upload-time = "2025-09-14T22:16:57.774Z" },
+    { url = "https://files.pythonhosted.org/packages/53/6c/288c3f0bd9fcfe9ca41e2c2fbfd17b2097f6af57b62a81161941f09afa76/zstandard-0.25.0-cp312-cp312-manylinux2010_i686.manylinux2014_i686.manylinux_2_12_i686.manylinux_2_17_i686.whl", hash = "sha256:011d388c76b11a0c165374ce660ce2c8efa8e5d87f34996aa80f9c0816698b64", size = 5343019, upload-time = "2025-09-14T22:16:59.302Z" },
+    { url = "https://files.pythonhosted.org/packages/1e/15/efef5a2f204a64bdb5571e6161d49f7ef0fffdbca953a615efbec045f60f/zstandard-0.25.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:6dffecc361d079bb48d7caef5d673c88c8988d3d33fb74ab95b7ee6da42652ea", size = 5063012, upload-time = "2025-09-14T22:17:01.156Z" },
+    { url = "https://files.pythonhosted.org/packages/b7/37/a6ce629ffdb43959e92e87ebdaeebb5ac81c944b6a75c9c47e300f85abdf/zstandard-0.25.0-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:7149623bba7fdf7e7f24312953bcf73cae103db8cae49f8154dd1eadc8a29ecb", size = 5394148, upload-time = "2025-09-14T22:17:03.091Z" },
+    { url = "https://files.pythonhosted.org/packages/e3/79/2bf870b3abeb5c070fe2d670a5a8d1057a8270f125ef7676d29ea900f496/zstandard-0.25.0-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.whl", hash = "sha256:6a573a35693e03cf1d67799fd01b50ff578515a8aeadd4595d2a7fa9f3ec002a", size = 5451652, upload-time = "2025-09-14T22:17:04.979Z" },
+    { url = "https://files.pythonhosted.org/packages/53/60/7be26e610767316c028a2cbedb9a3beabdbe33e2182c373f71a1c0b88f36/zstandard-0.25.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:5a56ba0db2d244117ed744dfa8f6f5b366e14148e00de44723413b2f3938a902", size = 5546993, upload-time = "2025-09-14T22:17:06.781Z" },
+    { url = "https://files.pythonhosted.org/packages/85/c7/3483ad9ff0662623f3648479b0380d2de5510abf00990468c286c6b04017/zstandard-0.25.0-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:10ef2a79ab8e2974e2075fb984e5b9806c64134810fac21576f0668e7ea19f8f", size = 5046806, upload-time = "2025-09-14T22:17:08.415Z" },
+    { url = "https://files.pythonhosted.org/packages/08/b3/206883dd25b8d1591a1caa44b54c2aad84badccf2f1de9e2d60a446f9a25/zstandard-0.25.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:aaf21ba8fb76d102b696781bddaa0954b782536446083ae3fdaa6f16b25a1c4b", size = 5576659, upload-time = "2025-09-14T22:17:10.164Z" },
+    { url = "https://files.pythonhosted.org/packages/9d/31/76c0779101453e6c117b0ff22565865c54f48f8bd807df2b00c2c404b8e0/zstandard-0.25.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:1869da9571d5e94a85a5e8d57e4e8807b175c9e4a6294e3b66fa4efb074d90f6", size = 4953933, upload-time = "2025-09-14T22:17:11.857Z" },
+    { url = "https://files.pythonhosted.org/packages/18/e1/97680c664a1bf9a247a280a053d98e251424af51f1b196c6d52f117c9720/zstandard-0.25.0-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:809c5bcb2c67cd0ed81e9229d227d4ca28f82d0f778fc5fea624a9def3963f91", size = 5268008, upload-time = "2025-09-14T22:17:13.627Z" },
+    { url = "https://files.pythonhosted.org/packages/1e/73/316e4010de585ac798e154e88fd81bb16afc5c5cb1a72eeb16dd37e8024a/zstandard-0.25.0-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:f27662e4f7dbf9f9c12391cb37b4c4c3cb90ffbd3b1fb9284dadbbb8935fa708", size = 5433517, upload-time = "2025-09-14T22:17:16.103Z" },
+    { url = "https://files.pythonhosted.org/packages/5b/60/dd0f8cfa8129c5a0ce3ea6b7f70be5b33d2618013a161e1ff26c2b39787c/zstandard-0.25.0-cp312-cp312-musllinux_1_2_s390x.whl", hash = "sha256:99c0c846e6e61718715a3c9437ccc625de26593fea60189567f0118dc9db7512", size = 5814292, upload-time = "2025-09-14T22:17:17.827Z" },
+    { url = "https://files.pythonhosted.org/packages/fc/5f/75aafd4b9d11b5407b641b8e41a57864097663699f23e9ad4dbb91dc6bfe/zstandard-0.25.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:474d2596a2dbc241a556e965fb76002c1ce655445e4e3bf38e5477d413165ffa", size = 5360237, upload-time = "2025-09-14T22:17:19.954Z" },
+    { url = "https://files.pythonhosted.org/packages/ff/8d/0309daffea4fcac7981021dbf21cdb2e3427a9e76bafbcdbdf5392ff99a4/zstandard-0.25.0-cp312-cp312-win32.whl", hash = "sha256:23ebc8f17a03133b4426bcc04aabd68f8236eb78c3760f12783385171b0fd8bd", size = 436922, upload-time = "2025-09-14T22:17:24.398Z" },
+    { url = "https://files.pythonhosted.org/packages/79/3b/fa54d9015f945330510cb5d0b0501e8253c127cca7ebe8ba46a965df18c5/zstandard-0.25.0-cp312-cp312-win_amd64.whl", hash = "sha256:ffef5a74088f1e09947aecf91011136665152e0b4b359c42be3373897fb39b01", size = 506276, upload-time = "2025-09-14T22:17:21.429Z" },
+    { url = "https://files.pythonhosted.org/packages/ea/6b/8b51697e5319b1f9ac71087b0af9a40d8a6288ff8025c36486e0c12abcc4/zstandard-0.25.0-cp312-cp312-win_arm64.whl", hash = "sha256:181eb40e0b6a29b3cd2849f825e0fa34397f649170673d385f3598ae17cca2e9", size = 462679, upload-time = "2025-09-14T22:17:23.147Z" },
+]
+
 [[package]]
 name = "zstd"
 version = "1.5.7.2"

docs/user-guide/tutorials/prowler-app-jira-integration.mdx

@@ -124,6 +124,18 @@ To manually send individual Findings to Jira:
 
     ![Send to Jira modal](/images/prowler-app/jira/send-to-jira-modal.png)
 
+### Finding Reference in the Jira Issue
+
+<VersionBadge version="5.41.0" />
+
+Every Jira issue created from a single Finding carries a stable reference back to that Finding, so issues can be filtered, searched with Jira Query Language (JQL), or matched by automation:
+
+* **Labels**: `prowler`, `prowler-<provider>`, `prowler-<severity>`, `prowler-<check-id>` and `prowler-finding-<finding-uid>`. Labels are sanitized deterministically: whitespace becomes `_`, control characters are removed, and values are truncated to Jira's 255-character label limit.
+* **Finding URL**: a link that opens the Finding in Prowler, filtered by its unique identifier (UID) so it keeps working after later scans.
+* **Tenant Info**: the name of the Prowler organization that sent the Finding.
+
+Prowler Cloud always includes the Finding URL. In Prowler Local Server, set `DJANGO_UI_BASE_URL` in the API environment (for example, `https://prowler.example.com`) to enable it. When the variable is empty, the issue is created without the link.
+
 ## Integration Status
 
 Monitor and manage your Jira integrations through the management interface:

docs/user-guide/tutorials/prowler-app-slack-integration.mdx

@@ -58,12 +58,14 @@ Two of these read more broadly than they behave, and both are worth understandin
 
 On the consent screen, `chat:write.public` reads as permission to post in any public channel. Prowler never uses it that way: **Prowler only ever posts to the channels authorized on the integration.** The scope exists so that authorizing a public channel does not also require someone to invite the Prowler app to it first.
 
+{/* The Prowler UI deep-links to this heading's anchor, so rewording the heading breaks that link. */}
+
 ### Why a Private Channel Is Missing From the Channel List
 
-`groups:read` reveals only the private channels the Prowler app is already a member of. A private channel therefore appears in the channel list only after someone invites `@Prowler` to it in Slack:
+`groups:read` reveals only the private channels the Prowler app is already a member of. A private channel therefore appears in the channel list only after someone invites `@Prowler Cloud` to it in Slack:
 
 ```text
-/invite @Prowler
+/invite @Prowler Cloud
 ```
 
 That invite is issued in Slack, by that channel's own members, and **the invite itself is the permission grant** — no scope bypasses it. Prowler ships no in-product flow to get the app invited, because the decision belongs to the channel's members. After inviting the app, click **Refresh channels** to re-read the list.
@@ -100,14 +102,14 @@ Prowler posts to the channels authorized on the integration. Several channels ca
 
     ![Destination channels selection listing public channels and an invited private channel marked Private](/images/prowler-app/slack/channel-picker.png)
 
-2. Select one or more channels. A selected private channel keeps its lock and **Private** identification with the list closed, so the authorized set stays readable at a glance.
+2. Select one or more channels. A selected private channel keeps the same **Private** marking with the list closed, so the authorized set stays readable at a glance.
 3. Click **Save channels**.
 
 Prowler validates the selection against Slack and derives each channel name itself, so a recorded name can never drift from the channel it belongs to. Once the set is saved, the page reports where Prowler posts and runs the connection check over it.
 
-If the selection reports that no channels are available, the workspace exposes nothing Prowler can see. Create a public channel, or invite `@Prowler` to a private one, then click **Refresh channels**.
+If the selection reports that no channels are available, the workspace exposes nothing Prowler can see. Create a public channel, or invite `@Prowler Cloud` to a private one, then click **Refresh channels**.
 
-A workspace can hold more channels than Prowler reads in one go. When that happens, the selection says so and lists what was read: every listed channel is usable, and a channel missing from a partial list is not necessarily one `@Prowler` has to be invited to. Only listed channels can be selected: **Refresh channels** repeats the same bounded read rather than reading further, and the selection's search filters what was already read, so neither surfaces a channel the read left out.
+A workspace can hold more channels than Prowler reads in one go. When that happens, the selection says so and lists what was read: every listed channel is usable, and a channel missing from a partial list is not necessarily one `@Prowler Cloud` has to be invited to. Only listed channels can be selected: **Refresh channels** repeats the same bounded read rather than reading further, and the selection's search filters what was already read, so neither surfaces a channel the read left out.
 
 Saving a new selection replaces the authorized set: channels left out of it stop being authorized, and channels added to it are authorized but not yet confirmed. Changing which channels are in the set also resets the integration's connection state, so the check runs again over the new set — reordering the same channels does not. Saving an empty selection leaves the integration with no authorized channels, and **Test connection** cannot be run again until at least one channel is authorized.
 
@@ -159,7 +161,7 @@ The Slack management page reports the state of the connection and offers these a
 | Button | Purpose | Notes |
 |--------|---------|-------|
 | **Test connection** | Verify the credential and every authorized channel, and confirm the ones not confirmed yet | Posts the confirmation message once per channel and updates the last-checked time. Cannot be run until at least one channel is authorized |
-| **Refresh channels** | Re-read the workspace's channel list | Use after inviting `@Prowler` to a private channel |
+| **Refresh channels** | Re-read the workspace's channel list | Use after inviting `@Prowler Cloud` to a private channel |
 | **Save channels** | Record the selected channels as the integration's authorized set | Enabled once the selection differs from the authorized set |
 | **Disconnect** | Remove the integration and attempt to revoke access at Slack | ⚠️ **Cannot be undone** — confirm before disconnecting |
 
@@ -171,7 +173,7 @@ The Prowler Slack app is not configured for the deployment being used, so no wor
 
 ### A Private Channel Does Not Appear in the Channel List
 
-The Prowler app has not been invited to it. In Slack, run `/invite @Prowler` in that channel, then click **Refresh channels**. Membership is the permission: no scope reveals a private channel the app is not in.
+The Prowler app has not been invited to it. In Slack, run `/invite @Prowler Cloud` in that channel, then click **Refresh channels**. Membership is the permission: no scope reveals a private channel the app is not in.
 
 ### Connection Test Fails
 

mcp_server/changelog.d/mcp-docs-search-swallowed-errors.fixed.md

@@ -0,0 +1 @@
+`prowler_docs_search` no longer reports a failed search as zero matches or an unreadable answer as a bad search term, and `prowler_docs_get_document` no longer reports a failed fetch as a missing page

mcp_server/prowler_mcp_server/prowler_documentation/search_engine.py

@@ -2,6 +2,7 @@ import httpx
 from pydantic import BaseModel, Field
 
 from prowler_mcp_server import __version__
+from prowler_mcp_server.lib.errors import parse_json_response
 
 
 class SearchResult(BaseModel):
@@ -58,8 +59,7 @@ class ProwlerDocsSearchEngine:
         )
 
     def search(self, query: str, page_size: int = 5) -> list[SearchResult]:
-        """
-        Search documentation using Mintlify API.
+        """Search documentation using Mintlify API.
 
         Args:
             query: Search query string
@@ -69,82 +69,85 @@ class ProwlerDocsSearchEngine:
 
         Returns:
             list of search results
+
+        Raises:
+            httpx.HTTPError: If the search request failed, which is not the same
+                answer as no matches
+            UpstreamInvalidResponse: If the answer is not JSON, which is the
+                documentation site's fault and not the search term's
         """
-        try:
-            # Make request to Mintlify API
-            response = self.mintlify_client.post(
-                self.api_base_url,
-                json={"query": query, "filters": {}},
-            )
-            response.raise_for_status()
-            data = response.json()
+        # Make request to Mintlify API
+        response = self.mintlify_client.post(
+            self.api_base_url,
+            json={"query": query, "filters": {}},
+        )
+        response.raise_for_status()
+        # Not `response.json()`: the decode error it raises is a ValueError, which
+        # the shared classifier reads as a malformed argument and answers by
+        # telling the caller to fix a search term that was never the problem.
+        data = parse_json_response(response)
 
-            # Parse results
-            results = []
-            for match in data.get("results", [])[:page_size]:
-                metadata = match.get("metadata", {})
-                breadcrumbs = metadata.get("breadcrumbs", [])
-                doc_path = match.get("page", "")
+        # Parse results
+        results = []
+        for match in data.get("results", [])[:page_size]:
+            metadata = match.get("metadata", {})
+            breadcrumbs = metadata.get("breadcrumbs", [])
+            doc_path = match.get("page", "")
 
-                # A match is one section of a page rather than the page: the
-                # heading it was found under is its header, and the page's own
-                # title is the last step of its breadcrumb trail.
-                section = match.get("header", "")
-                title = breadcrumbs[-1] if breadcrumbs else section
+            # A match is one section of a page rather than the page: the
+            # heading it was found under is its header, and the page's own
+            # title is the last step of its breadcrumb trail.
+            section = match.get("header", "")
+            title = breadcrumbs[-1] if breadcrumbs else section
 
-                # Sent as "" for the section a page opens with and as null for
-                # the pages that have no anchors at all; both mean the page.
-                anchor = metadata.get("hash")
-                url = f"{self.docs_base_url}/{doc_path}"
-                if anchor:
-                    url = f"{url}#{anchor}"
+            # Sent as "" for the section a page opens with and as null for
+            # the pages that have no anchors at all; both mean the page.
+            anchor = metadata.get("hash")
+            url = f"{self.docs_base_url}/{doc_path}"
+            if anchor:
+                url = f"{url}#{anchor}"
 
-                results.append(
-                    SearchResult(
-                        path=doc_path,
-                        title=title,
-                        section=section,
-                        breadcrumbs=breadcrumbs,
-                        url=url,
-                        excerpt=match.get("content", ""),
-                        score=match.get("score", 0.0),
-                    )
+            results.append(
+                SearchResult(
+                    path=doc_path,
+                    title=title,
+                    section=section,
+                    breadcrumbs=breadcrumbs,
+                    url=url,
+                    excerpt=match.get("content", ""),
+                    score=match.get("score", 0.0),
                 )
+            )
 
-            return results
-
-        except Exception as e:
-            # Return empty list on error
-            print(f"Search error: {e}")
-            return []
+        return results
 
     def get_document(self, doc_path: str) -> str | None:
-        """
-        Get full document content from Mintlify documentation.
+        """Get full document content from Mintlify documentation.
 
         Args:
             doc_path: Path to the documentation file (e.g., "getting-started/installation")
 
         Returns:
-            Full markdown content of the documentation, or None if not found
+            Full markdown content of the documentation, or None if there is no
+            page at that path
+
+        Raises:
+            httpx.HTTPError: If the fetch failed for any reason other than a 404
         """
-        try:
-            # Clean up the path
-            doc_path = doc_path.rstrip("/")
+        # Clean up the path
+        doc_path = doc_path.rstrip("/")
 
-            # Add .md extension if not present (Mintlify serves both .md and .mdx)
-            if not doc_path.endswith(".md"):
-                doc_path = f"{doc_path}.md"
+        # Add .md extension if not present (Mintlify serves both .md and .mdx)
+        if not doc_path.endswith(".md"):
+            doc_path = f"{doc_path}.md"
 
-            # Construct Mintlify URL
-            url = f"{self.docs_base_url}/{doc_path}"
+        # Construct Mintlify URL
+        url = f"{self.docs_base_url}/{doc_path}"
 
-            # Fetch the documentation page
-            response = self.docs_client.get(url)
-            response.raise_for_status()
-
-            return response.text
-
-        except Exception as e:
-            print(f"Error fetching document: {e}")
+        # Fetch the documentation page
+        response = self.docs_client.get(url)
+        if response.status_code == 404:
             return None
+        response.raise_for_status()
+
+        return response.text

mcp_server/prowler_mcp_server/prowler_documentation/server.py

@@ -1,6 +1,7 @@
 from typing import Any
 
 from fastmcp import FastMCP
+from fastmcp.exceptions import ToolError
 from pydantic import Field
 
 from prowler_mcp_server.lib.types import NonBlankStr
@@ -9,7 +10,7 @@ from prowler_mcp_server.prowler_documentation.search_engine import (
 )
 
 # Initialize FastMCP server
-docs_mcp_server = FastMCP("prowler-docs")
+docs_mcp_server = FastMCP("prowler-docs", mask_error_details=True)
 prowler_docs_search_engine = ProwlerDocsSearchEngine()
 
 
@@ -56,6 +57,10 @@ def get_document(
     """
     content: str | None = prowler_docs_search_engine.get_document(doc_path)
     if content is None:
-        return {"error": f"Document '{doc_path}' not found."}
-    else:
-        return {"content": content}
+        # No `from`: this names the path asked for and the tool that produces a
+        # valid one, neither of which the shared classifier can know.
+        raise ToolError(
+            f"The Prowler documentation has no page at '{doc_path}'. Use "
+            "prowler_docs_search and pass the 'path' field of a result verbatim."
+        )
+    return {"content": content}

mcp_server/tests/prowler_documentation/test_server.py

@@ -1,7 +1,9 @@
-"""Tests for the Prowler documentation search tool.
+"""Tests for the Prowler documentation tools.
 
 Mintlify moved the docs search to a new endpoint that answers with page
-sections, so a result is a part of a page and has to read as one.
+sections, so a result is a part of a page and has to read as one. And a failed
+request must not reach an agent as "the documentation has nothing on this",
+which is an answer it would act on, confidently and wrongly.
 """
 
 import json
@@ -9,6 +11,7 @@ import json
 from fastmcp import Client
 
 SEARCH = "/api/search/prowler"
+DOC = "/getting-started/installation.md"
 
 
 def search_match(
@@ -107,3 +110,65 @@ async def test_page_size_caps_a_response_the_api_did_not_size(
         )
 
     assert len(result.data) == 2
+
+
+async def test_a_search_that_failed_is_not_reported_as_no_matches(
+    mcp_root_server, docs_router
+):
+    """An empty list is an answer. A failed request is not, and must not look like one."""
+    docs_router.add("POST", SEARCH, status=500, text="upstream error")
+
+    async with Client(mcp_root_server) as client:
+        result = await client.call_tool_mcp("prowler_docs_search", {"term": "install"})
+
+    assert result.isError is True
+    assert result.structuredContent is None
+
+
+async def test_a_missing_page_fails_and_names_the_tool_that_finds_a_valid_path(
+    mcp_root_server, docs_router
+):
+    """A 404 answers the question, and still reaches the agent as an error."""
+    docs_router.add("GET", DOC, status=404, text="Not Found")
+
+    async with Client(mcp_root_server) as client:
+        result = await client.call_tool_mcp(
+            "prowler_docs_get_document", {"doc_path": "getting-started/installation"}
+        )
+
+    assert result.isError is True
+    assert "prowler_docs_search" in result.content[0].text
+
+
+async def test_a_fetch_that_failed_is_not_reported_as_a_missing_page(
+    mcp_root_server, docs_router
+):
+    """Only a 404 answers the question; every other status left it unanswered."""
+    docs_router.add("GET", DOC, status=503, text="upstream error")
+
+    async with Client(mcp_root_server) as client:
+        result = await client.call_tool_mcp(
+            "prowler_docs_get_document", {"doc_path": "getting-started/installation"}
+        )
+
+    assert result.isError is True
+    assert "no page at" not in result.content[0].text
+
+
+async def test_an_unreadable_body_is_not_reported_as_a_bad_search_term(
+    mcp_root_server, docs_router
+):
+    """An edge serving HTML is the site's fault; the caller has no term to fix."""
+    docs_router.add("POST", SEARCH, status=200, text="<html>edge error page</html>")
+
+    async with Client(mcp_root_server) as client:
+        result = await client.call_tool_mcp("prowler_docs_search", {"term": "install"})
+
+    assert result.isError is True
+    message = result.content[0].text
+    # Named as the upstream at fault, and explicitly not the caller's arguments,
+    # which is the story the shared ValueError branch would otherwise tell.
+    assert "leaves.mintlify.com" in message
+    assert "changing them will not help" in message
+    # The body it choked on is upstream text, which this server never relays.
+    assert "edge error page" not in message

prowler/changelog.d/aws-partition-env-var.added.md

@@ -0,0 +1 @@
+`PROWLER_AWS_PARTITION` environment variable to select the AWS partition used for STS credential validation and scan bootstrap, with a clear error when the account belongs to a different partition

prowler/changelog.d/cis-1.4-googleworkspace.added.md

@@ -0,0 +1 @@
+CIS Google Workspace Foundations Benchmark v1.4.0 compliance framework

prowler/changelog.d/cloudflare-challenge-ttl-none.fixed.md

@@ -0,0 +1 @@
+Zones without a challenge passage TTL are reported as non-compliant by `zone_challenge_passage_configured`

prowler/changelog.d/googleworkspace-2sv-policy-overrides.fixed.md

@@ -0,0 +1 @@
+`security_2sv_enforced` and `security_2sv_hardware_keys_admins` report MANUAL instead of judging domain-wide values that a group or a sub-organizational unit overrides, or that were dropped because the root organizational unit could not be resolved; a domain-wide failure is still reported as such, with the override noted

prowler/changelog.d/googleworkspace-alert-center-delivery.fixed.md

@@ -0,0 +1 @@
+`rules_*_alert_configured` checks no longer pass a rule whose delivery to the alert center is disabled, the setting behind the benchmark's "Ensure that Alerts is set to On"

prowler/changelog.d/googleworkspace-cis-full-audit-procedures.changed.md

@@ -0,0 +1 @@
+Google Workspace checks mapped to CIS evaluate the full audit procedure instead of a single condition, so Gmail spoofing actions, 2-Step Verification, password expiration and alert severity left on Google's defaults no longer pass

prowler/changelog.d/googleworkspace-compliance-mappings.changed.md

@@ -0,0 +1 @@
+`security_login_challenges_configured` and `security_2sv_enforced` unmapped from CIS Google Workspace 4.1.4.1 and CISA SCuBA 0.6 `GWS.COMMONCONTROLS.1.1`, whose Post-SSO verification and phishing-resistant MFA requirements neither check can prove

prowler/changelog.d/googleworkspace-password-strength-default.fixed.md

@@ -0,0 +1 @@
+`security_password_policy_strong` no longer fails a domain that never touched the password strength setting: Google enforces strong passwords by default, so an unset value is the secure default and not a missing configuration

prowler/compliance/googleworkspace/cis_1.3_googleworkspace.json

@@ -1498,9 +1498,7 @@
     {
       "Id": "4.1.4.1",
       "Description": "Ensure login challenges are enforced",
-      "Checks": [
-        "security_login_challenges_configured"
-      ],
+      "Checks": [],
       "Attributes": [
         {
           "Section": "4 Security",

prowler/compliance/googleworkspace/cisa_scuba_0.6_googleworkspace.json

@@ -9,7 +9,6 @@
       "Id": "GWS.COMMONCONTROLS.1.1",
       "Description": "Phishing-resistant MFA SHALL be required for all users",
       "Checks": [
-        "security_2sv_enforced",
         "security_2sv_hardware_keys_admins"
       ],
       "Attributes": [

prowler/providers/aws/aws_provider.py

@@ -1,6 +1,7 @@
 import os
 import pathlib
 from datetime import datetime
+from functools import lru_cache
 from re import fullmatch
 from typing import Optional
 
@@ -671,7 +672,12 @@ class AwsProvider(Provider):
             if mfa:
                 session = Session(**session_arguments)
                 session._session.set_default_client_config(session_config)
-                sts_client = session.client("sts")
+                sts_region = (
+                    get_env_partition_bootstrap_region()
+                    or session.region_name
+                    or AWS_STS_GLOBAL_ENDPOINT_REGION
+                )
+                sts_client = AwsProvider.create_sts_session(session, sts_region)
 
                 # TODO: pass values from the input
                 mfa_info = AwsProvider.input_role_mfa_token_and_code()
@@ -1352,7 +1358,7 @@ class AwsProvider(Provider):
     @staticmethod
     def test_connection(
         profile: str = None,
-        aws_region: str = AWS_STS_GLOBAL_ENDPOINT_REGION,
+        aws_region: str = None,
         role_arn: str = None,
         role_session_name: str = ROLE_SESSION_NAME,
         session_duration: int = 3600,
@@ -1369,7 +1375,9 @@ class AwsProvider(Provider):
 
         Args:
             profile (str): The AWS profile to use for the session.
-            aws_region (str): The AWS region to validate the credentials in.
+            aws_region (str): The AWS region to validate the credentials in. When not
+                provided, it defaults to the bootstrap region of the partition set in
+                the PROWLER_AWS_PARTITION environment variable or, if unset, to us-east-1.
             role_arn (str): The ARN of the IAM role to assume.
             role_session_name (str): The name of the role session.
             session_duration (int): The duration of the assumed role session in seconds.
@@ -1412,6 +1420,12 @@ class AwsProvider(Provider):
             Connection(is_connected=True, Error=None))
         """
         try:
+            if aws_region is None:
+                aws_region = (
+                    get_env_partition_bootstrap_region()
+                    or AWS_STS_GLOBAL_ENDPOINT_REGION
+                )
+
             session = AwsProvider.setup_session(
                 mfa=mfa_enabled,
                 profile=profile,
@@ -1430,6 +1444,7 @@ class AwsProvider(Provider):
                     external_id=external_id,
                     mfa_enabled=mfa_enabled,
                     role_session_name=role_session_name,
+                    sts_region=aws_region,
                 )
                 assumed_role_credentials = AwsProvider.assume_role(
                     session,
@@ -1451,6 +1466,13 @@ class AwsProvider(Provider):
             if provider_id and caller_identity.account != provider_id:
                 raise AWSInvalidProviderIdError(file=pathlib.Path(__file__).name)
 
+            # Validate that the account belongs to the configured partition, if any
+            env_partition = os.environ.get("PROWLER_AWS_PARTITION", "").strip()
+            if env_partition and caller_identity.arn.partition != env_partition:
+                raise AWSInvalidPartitionError(
+                    message=f"The AWS account is in the {caller_identity.arn.partition} partition, but this deployment is configured for the {env_partition} partition via PROWLER_AWS_PARTITION"
+                )
+
             return Connection(
                 is_connected=True,
             )
@@ -1591,6 +1613,14 @@ class AwsProvider(Provider):
                 raise session_token_expired
             return Connection(error=session_token_expired)
 
+        except AWSInvalidPartitionError as invalid_partition_error:
+            logger.error(
+                f"{invalid_partition_error.__class__.__name__}[{invalid_partition_error.__traceback__.tb_lineno}]: {invalid_partition_error}"
+            )
+            if raise_on_exception:
+                raise invalid_partition_error
+            return Connection(error=invalid_partition_error)
+
         except Exception as error:
             logger.critical(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -1618,14 +1648,9 @@ class AwsProvider(Provider):
             sts_client = create_sts_session(session, 'us-west-2')
         """
         try:
-            if os.environ.get("AWS_ENDPOINT_URL"):
-                sts_endpoint_url = os.environ["AWS_ENDPOINT_URL"]
-            elif aws_region.startswith("cn-"):
-                sts_endpoint_url = f"https://sts.{aws_region}.amazonaws.com.cn"
-            elif aws_region.startswith("eusc-"):
-                sts_endpoint_url = f"https://sts.{aws_region}.amazonaws.eu"
-            else:
-                sts_endpoint_url = f"https://sts.{aws_region}.amazonaws.com"
+            # Botocore resolves the regional STS endpoint for every partition
+            # (China, EUSC, GovCloud, ISO); AWS_ENDPOINT_URL overrides it
+            sts_endpoint_url = os.environ.get("AWS_ENDPOINT_URL") or None
             return session.client("sts", aws_region, endpoint_url=sts_endpoint_url)
         except Exception as error:
             logger.critical(
@@ -1702,6 +1727,80 @@ def read_aws_regions_file() -> dict:
     return data
 
 
+@lru_cache(maxsize=1)
+def get_botocore_partition_regions() -> dict:
+    """
+    Get the AWS partitions and their bootstrap region candidates from the
+    botocore endpoints data.
+
+    The region of the partition's global STS endpoint, when declared, is moved
+    to the front since it is never an opt-in region; the rest are sorted
+    alphabetically.
+
+    Returns:
+        dict: A dictionary mapping each partition name to its list of regions.
+    """
+    endpoints_data = BotocoreSession().get_data("endpoints")
+    partition_regions = {}
+    for partition in endpoints_data["partitions"]:
+        regions = sorted(partition.get("regions", {}))
+        sts_service = partition.get("services", {}).get("sts", {})
+        global_endpoint = sts_service.get("partitionEndpoint")
+        global_region = (
+            sts_service.get("endpoints", {})
+            .get(global_endpoint, {})
+            .get("credentialScope", {})
+            .get("region")
+        )
+        if global_region in regions:
+            regions.remove(global_region)
+            regions.insert(0, global_region)
+        partition_regions[partition["partition"]] = regions
+    return partition_regions
+
+
+def get_env_partition_regions() -> Optional[list]:
+    """
+    Get the bootstrap region candidates for the partition set in the
+    PROWLER_AWS_PARTITION environment variable.
+
+    Returns:
+        Optional[list]: The regions of the configured partition, preferred
+            bootstrap region first, or None when the environment variable is
+            not set.
+
+    Raises:
+        AWSInvalidPartitionError: If the value is not a partition known to botocore.
+    """
+    raw_partition = os.environ.get("PROWLER_AWS_PARTITION", "").strip()
+    if not raw_partition:
+        return None
+
+    partition_regions = get_botocore_partition_regions()
+    regions = partition_regions.get(raw_partition)
+    if not regions:
+        raise AWSInvalidPartitionError(
+            message=f"Invalid partition: {raw_partition} set in PROWLER_AWS_PARTITION. Valid partitions: {', '.join(sorted(partition_regions))}"
+        )
+    return regions
+
+
+def get_env_partition_bootstrap_region() -> Optional[str]:
+    """
+    Get the STS bootstrap region for the partition set in the
+    PROWLER_AWS_PARTITION environment variable.
+
+    Returns:
+        Optional[str]: The preferred bootstrap region of the configured
+            partition, or None when the environment variable is not set.
+
+    Raises:
+        AWSInvalidPartitionError: If the value is not a partition known to botocore.
+    """
+    regions = get_env_partition_regions()
+    return regions[0] if regions else None
+
+
 # TODO: This can be moved to another class since it doesn't need self
 def get_aws_region_for_sts(
     session_region: str,
@@ -1711,6 +1810,10 @@ def get_aws_region_for_sts(
     """
     Get the AWS region for the STS Assume Role operation.
 
+    The precedence is: explicit regions, the partition set in the
+    PROWLER_AWS_PARTITION environment variable, the session region and,
+    finally, the bootstrap region candidates.
+
     Args:
         - session_region (str): The region configured in the AWS session.
         - regions (set[str]): The regions passed with the -f/--region/--filter-region option.
@@ -1730,6 +1833,15 @@ def get_aws_region_for_sts(
             if region not in excluded_regions:
                 return region
 
+    env_partition_regions = get_env_partition_regions()
+    if env_partition_regions:
+        # The configured partition constrains the whole fallback chain: prefer
+        # a non-excluded region, but never leave the partition
+        for region in env_partition_regions:
+            if region not in excluded_regions:
+                return region
+        return env_partition_regions[0]
+
     if session_region and session_region not in excluded_regions:
         return session_region
 

prowler/providers/cloudflare/services/zone/zone_challenge_passage_configured/zone_challenge_passage_configured.py

@@ -29,17 +29,23 @@ class zone_challenge_passage_configured(Check):
                 metadata=self.metadata(),
                 resource=zone,
             )
-            # API returns seconds, convert to minutes
-            challenge_ttl_minutes = zone.settings.challenge_ttl // 60
-
-            if min_minutes <= challenge_ttl_minutes <= max_minutes:
-                report.status = "PASS"
-                report.status_extended = f"Challenge Passage is set to {challenge_ttl_minutes} minutes for zone {zone.name}."
-            else:
+            challenge_ttl = zone.settings.challenge_ttl
+            if challenge_ttl is None:
                 report.status = "FAIL"
                 report.status_extended = (
-                    f"Challenge Passage is set to {challenge_ttl_minutes} minutes for zone {zone.name} "
-                    f"(recommended: between {min_minutes} and {max_minutes} minutes)."
+                    f"Challenge Passage is not configured for zone {zone.name}."
                 )
+            else:
+                # API returns seconds, convert to minutes
+                challenge_ttl_minutes = challenge_ttl // 60
+                if min_minutes <= challenge_ttl_minutes <= max_minutes:
+                    report.status = "PASS"
+                    report.status_extended = f"Challenge Passage is set to {challenge_ttl_minutes} minutes for zone {zone.name}."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = (
+                        f"Challenge Passage is set to {challenge_ttl_minutes} minutes for zone {zone.name} "
+                        f"(recommended: between {min_minutes} and {max_minutes} minutes)."
+                    )
             findings.append(report)
         return findings

prowler/providers/googleworkspace/lib/service/service.py

@@ -6,6 +6,11 @@ from prowler.providers.googleworkspace.googleworkspace_provider import (
     GoogleworkspaceProvider,
 )
 
+# How far a Cloud Identity policy reaches.
+CUSTOMER_SCOPE = "customer"
+OVERRIDE_SCOPE = "override"
+UNKNOWN_SCOPE = "unknown"
+
 
 class GoogleWorkspaceService:
     def __init__(
@@ -42,26 +47,29 @@ class GoogleWorkspaceService:
             )
             return None
 
-    def _is_customer_level_policy(self, policy: dict) -> bool:
-        """Check if a policy applies at the customer (domain-wide) level.
+    def _policy_scope(self, policy: dict) -> str:
+        """Return how far a policy reaches: CUSTOMER_SCOPE, OVERRIDE_SCOPE or UNKNOWN_SCOPE.
 
-        The Cloud Identity Policy API typically scopes all policies to an OU;
-        absence of orgUnit is treated as customer-level as a safety net.
-        The root OU is equivalent to customer-level. This method accepts
-        policies with no orgUnit or policies targeting the root OU,
-        and rejects group-targeted and sub-OU policies.
+        The Cloud Identity Policy API typically scopes every policy to an OU,
+        and the root OU is equivalent to customer-level, so telling them apart
+        needs the root OU id. That id is fetched on a best-effort basis, and
+        without it a root-OU policy is indistinguishable from a sub-OU one:
+        that is UNKNOWN_SCOPE, which callers must not read as either.
         """
-        policy_query = policy.get("policyQuery", {})
+        policy_query = policy.get("policyQuery") or {}
         if policy_query.get("group"):
-            return False
+            return OVERRIDE_SCOPE
         org_unit = policy_query.get("orgUnit")
         if not org_unit:
-            return True
-        # Accept root OU as customer-level
+            return CUSTOMER_SCOPE
         root_id = getattr(self.provider.identity, "root_org_unit_id", None)
-        if root_id and org_unit == f"orgUnits/{root_id}":
-            return True
-        return False
+        if not root_id:
+            return UNKNOWN_SCOPE
+        return CUSTOMER_SCOPE if org_unit == f"orgUnits/{root_id}" else OVERRIDE_SCOPE
+
+    def _is_customer_level_policy(self, policy: dict) -> bool:
+        """Whether a policy applies to the whole domain."""
+        return self._policy_scope(policy) == CUSTOMER_SCOPE
 
     def _handle_api_error(self, error, context: str, resource_name: str = ""):
         """

prowler/providers/googleworkspace/services/gmail/gmail_domain_spoofing_protection_enabled/gmail_domain_spoofing_protection_enabled.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "collaboration",
-  "Description": "Verifies that Gmail is configured to take a protective action (such as moving to spam, quarantining, or showing a warning) when emails appear to come from domain names that look similar to the organization's domain. Lookalike domains are a common phishing technique used to trick users into trusting malicious messages.",
+  "Description": "Verifies that Gmail moves out of the inbox, to spam or quarantine, the emails that appear to come from domain names that look similar to the organization's domain. An action that only shows a warning is not enough. Lookalike domains are a common phishing technique used to trick users into trusting malicious messages.",
   "Risk": "Without protection against domain spoofing based on similar domain names, users may receive **phishing emails from lookalike domains** (e.g., examp1e.com instead of example.com) that appear legitimate. This enables **credential theft, malware delivery, and business email compromise** attacks.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/gmail/gmail_domain_spoofing_protection_enabled/gmail_domain_spoofing_protection_enabled.py

@@ -2,6 +2,10 @@ from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
 from prowler.providers.googleworkspace.services.gmail.gmail_client import gmail_client
+from prowler.providers.googleworkspace.services.gmail.lib.spoofing import (
+    describe_consequence,
+    is_protective,
+)
 
 
 class gmail_domain_spoofing_protection_enabled(Check):
@@ -9,7 +13,9 @@ class gmail_domain_spoofing_protection_enabled(Check):
 
     This check verifies that Gmail is configured to take action on
     emails that appear to come from similar-looking domain names,
-    helping prevent phishing via domain impersonation.
+    helping prevent phishing via domain impersonation. CIS requires the
+    configured action to move the message to spam or quarantine it, so an action
+    that only shows a warning is reported as a failure.
     """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
@@ -26,38 +32,30 @@ class gmail_domain_spoofing_protection_enabled(Check):
 
             enabled = gmail_client.policies.detect_domain_name_spoofing
             consequence = gmail_client.policies.domain_spoofing_consequence
+            domain = gmail_client.provider.identity.domain
 
             if enabled is False:
                 report.status = "FAIL"
                 report.status_extended = (
-                    f"Protection against domain spoofing based on similar "
-                    f"domain names is disabled in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"Enable the protection and configure a protective action."
+                    f"Protection against domain spoofing based on similar domain names "
+                    f"is disabled in domain {domain}. "
+                    f"Enable the protection and set the action to move the "
+                    f"email to spam."
                 )
-            elif consequence == "NO_ACTION":
+            elif not is_protective(consequence):
                 report.status = "FAIL"
                 report.status_extended = (
-                    f"Protection against domain spoofing based on similar "
-                    f"domain names is set to take no action in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"A protective action should be configured."
-                )
-            elif consequence is None:
-                report.status = "PASS"
-                report.status_extended = (
-                    f"Protection against domain spoofing based on similar "
-                    f"domain names uses Google's secure default configuration "
-                    f"(enabled) in domain "
-                    f"{gmail_client.provider.identity.domain}."
+                    f"Protection against domain spoofing based on similar domain names "
+                    f"{describe_consequence(consequence)} in domain {domain}. "
+                    f"The action should move the email to spam."
                 )
             else:
                 report.status = "PASS"
+                state = "is enabled" if enabled else "uses Google's default (enabled)"
                 report.status_extended = (
-                    f"Protection against domain spoofing based on similar "
-                    f"domain names is enabled with consequence "
-                    f"'{consequence}' in domain "
-                    f"{gmail_client.provider.identity.domain}."
+                    f"Protection against domain spoofing based on similar domain names "
+                    f"{state} with action '{consequence}' in domain "
+                    f"{domain}."
                 )
 
             findings.append(report)

prowler/providers/googleworkspace/services/gmail/gmail_employee_name_spoofing_protection_enabled/gmail_employee_name_spoofing_protection_enabled.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "collaboration",
-  "Description": "Verifies that Gmail is configured to take a protective action (such as moving to spam, quarantining, or showing a warning) when the sender's display name matches an employee's name but the email comes from an external address. This is a common social engineering technique where attackers impersonate colleagues or executives.",
+  "Description": "Verifies that Gmail moves out of the inbox, to spam or quarantine, the emails whose sender display name matches an employee's name but come from an external address. An action that only shows a warning is not enough. This is a common social engineering technique where attackers impersonate colleagues or executives.",
   "Risk": "Without protection against employee name spoofing, users may receive **emails that appear to come from colleagues or executives** but are actually from external attackers. This enables **business email compromise (BEC)**, **wire fraud**, and **social engineering attacks** that exploit trust relationships.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/gmail/gmail_employee_name_spoofing_protection_enabled/gmail_employee_name_spoofing_protection_enabled.py

@@ -2,6 +2,10 @@ from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
 from prowler.providers.googleworkspace.services.gmail.gmail_client import gmail_client
+from prowler.providers.googleworkspace.services.gmail.lib.spoofing import (
+    describe_consequence,
+    is_protective,
+)
 
 
 class gmail_employee_name_spoofing_protection_enabled(Check):
@@ -9,7 +13,9 @@ class gmail_employee_name_spoofing_protection_enabled(Check):
 
     This check verifies that Gmail is configured to take action on
     emails where the sender name matches an employee name but comes
-    from an external address, helping prevent social engineering attacks.
+    from an external address, helping prevent social engineering attacks. CIS requires the configured
+    action to move the message to spam or quarantine it, so an action that
+    only shows a warning is reported as a failure.
     """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
@@ -26,36 +32,30 @@ class gmail_employee_name_spoofing_protection_enabled(Check):
 
             enabled = gmail_client.policies.detect_employee_name_spoofing
             consequence = gmail_client.policies.employee_name_spoofing_consequence
+            domain = gmail_client.provider.identity.domain
 
             if enabled is False:
                 report.status = "FAIL"
                 report.status_extended = (
-                    f"Protection against spoofing of employee names is "
-                    f"disabled in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"Enable the protection and configure a protective action."
+                    f"Protection against spoofing of employee names "
+                    f"is disabled in domain {domain}. "
+                    f"Enable the protection and set the action to move the "
+                    f"email to spam."
                 )
-            elif consequence == "NO_ACTION":
+            elif not is_protective(consequence):
                 report.status = "FAIL"
                 report.status_extended = (
-                    f"Protection against spoofing of employee names is set "
-                    f"to take no action in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"A protective action should be configured."
-                )
-            elif consequence is None:
-                report.status = "PASS"
-                report.status_extended = (
-                    f"Protection against spoofing of employee names uses "
-                    f"Google's secure default configuration (enabled) "
-                    f"in domain {gmail_client.provider.identity.domain}."
+                    f"Protection against spoofing of employee names "
+                    f"{describe_consequence(consequence)} in domain {domain}. "
+                    f"The action should move the email to spam."
                 )
             else:
                 report.status = "PASS"
+                state = "is enabled" if enabled else "uses Google's default (enabled)"
                 report.status_extended = (
-                    f"Protection against spoofing of employee names is "
-                    f"enabled with consequence '{consequence}' in domain "
-                    f"{gmail_client.provider.identity.domain}."
+                    f"Protection against spoofing of employee names "
+                    f"{state} with action '{consequence}' in domain "
+                    f"{domain}."
                 )
 
             findings.append(report)

prowler/providers/googleworkspace/services/gmail/gmail_groups_spoofing_protection_enabled/gmail_groups_spoofing_protection_enabled.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "collaboration",
-  "Description": "Verifies that Gmail is configured to take a protective action (such as moving to spam, quarantining, or showing a warning) when groups receive inbound emails that spoof the organization's domain. Google Groups are a high-value target because a single spoofed message can reach many recipients at once.",
+  "Description": "Verifies that Gmail moves out of the inbox, to spam or quarantine, the inbound emails to groups that spoof the organization's domain. An action that only shows a warning is not enough. Google Groups are a high-value target because a single spoofed message can reach many recipients at once.",
   "Risk": "Without protection of groups from domain-spoofing emails, attackers can send **spoofed messages to group mailboxes** that appear to originate from the organization. Since groups distribute to many recipients, a single spoofed email can enable **mass phishing, social engineering, or misinformation** campaigns across the organization.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/gmail/gmail_groups_spoofing_protection_enabled/gmail_groups_spoofing_protection_enabled.py

@@ -2,6 +2,10 @@ from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
 from prowler.providers.googleworkspace.services.gmail.gmail_client import gmail_client
+from prowler.providers.googleworkspace.services.gmail.lib.spoofing import (
+    describe_consequence,
+    is_protective,
+)
 
 
 class gmail_groups_spoofing_protection_enabled(Check):
@@ -10,6 +14,9 @@ class gmail_groups_spoofing_protection_enabled(Check):
     This check verifies that Gmail is configured to take action on
     inbound emails to groups that spoof the organization's domain,
     helping prevent impersonation attacks targeting group mailboxes.
+    CIS requires the configured action to move the message to spam or
+    quarantine it, so an action that only shows a warning is reported as a
+    failure.
     """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
@@ -27,56 +34,44 @@ class gmail_groups_spoofing_protection_enabled(Check):
             enabled = gmail_client.policies.detect_groups_spoofing
             consequence = gmail_client.policies.groups_spoofing_consequence
             visibility_type = gmail_client.policies.groups_spoofing_visibility_type
+            domain = gmail_client.provider.identity.domain
+            scope = (
+                "private groups only"
+                if visibility_type == "PRIVATE_GROUPS_ONLY"
+                else "all groups"
+            )
 
             if enabled is False:
                 report.status = "FAIL"
                 report.status_extended = (
                     f"Protection of groups from inbound emails spoofing your "
-                    f"domain is disabled in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"Enable the protection and configure a protective action."
+                    f"domain is disabled in domain {domain}. "
+                    f"Enable the protection and set the action to move the "
+                    f"email to spam."
                 )
             elif enabled is None:
                 report.status = "FAIL"
                 report.status_extended = (
                     f"Protection of groups from inbound emails spoofing your "
                     f"domain is not configured and uses Google's insecure "
-                    f"default (disabled) in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"Enable the protection and configure a protective action."
+                    f"default (disabled) in domain {domain}. "
+                    f"Enable the protection and set the action to move the "
+                    f"email to spam."
                 )
-            elif consequence == "NO_ACTION":
+            elif not is_protective(consequence):
                 report.status = "FAIL"
                 report.status_extended = (
                     f"Protection of groups from inbound emails spoofing your "
-                    f"domain is set to take no action in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"A protective action should be configured."
-                )
-            elif consequence is None:
-                report.status = "PASS"
-                scope = (
-                    "private groups only"
-                    if visibility_type == "PRIVATE_GROUPS_ONLY"
-                    else "all groups"
-                )
-                report.status_extended = (
-                    f"Protection of groups from inbound emails spoofing your "
-                    f"domain is enabled for {scope} in domain "
-                    f"{gmail_client.provider.identity.domain}."
+                    f"domain is enabled for {scope} but "
+                    f"{describe_consequence(consequence)} in domain {domain}. "
+                    f"The action should move the email to spam."
                 )
             else:
                 report.status = "PASS"
-                scope = (
-                    "private groups only"
-                    if visibility_type == "PRIVATE_GROUPS_ONLY"
-                    else "all groups"
-                )
                 report.status_extended = (
                     f"Protection of groups from inbound emails spoofing your "
-                    f"domain is enabled for {scope} with consequence "
-                    f"'{consequence}' in domain "
-                    f"{gmail_client.provider.identity.domain}."
+                    f"domain is enabled for {scope} with action "
+                    f"'{consequence}' in domain {domain}."
                 )
 
             findings.append(report)

prowler/providers/googleworkspace/services/gmail/gmail_inbound_domain_spoofing_protection_enabled/gmail_inbound_domain_spoofing_protection_enabled.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "collaboration",
-  "Description": "Verifies that Gmail is configured to take a protective action (such as moving to spam, quarantining, or showing a warning) when inbound emails spoof the organization's own domain. This protects against attackers sending emails that appear to originate from within the organization but are actually external.",
+  "Description": "Verifies that Gmail moves out of the inbox, to spam or quarantine, the inbound emails that spoof the organization's own domain. An action that only shows a warning is not enough. This protects against attackers sending emails that appear to originate from within the organization but are actually external.",
   "Risk": "Without protection against inbound domain spoofing, users may receive **emails that appear to come from their own organization** but are sent by external attackers. This enables **internal impersonation**, **phishing**, and **business email compromise** attacks that exploit trust in internal communications.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/gmail/gmail_inbound_domain_spoofing_protection_enabled/gmail_inbound_domain_spoofing_protection_enabled.py

@@ -2,6 +2,10 @@ from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
 from prowler.providers.googleworkspace.services.gmail.gmail_client import gmail_client
+from prowler.providers.googleworkspace.services.gmail.lib.spoofing import (
+    describe_consequence,
+    is_protective,
+)
 
 
 class gmail_inbound_domain_spoofing_protection_enabled(Check):
@@ -9,7 +13,9 @@ class gmail_inbound_domain_spoofing_protection_enabled(Check):
 
     This check verifies that Gmail is configured to take action on
     inbound emails that spoof the organization's own domain, helping
-    prevent impersonation of internal senders.
+    prevent impersonation of internal senders. CIS requires the configured
+    action to move the message to spam or quarantine it, so an action that
+    only shows a warning is reported as a failure.
     """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
@@ -26,36 +32,30 @@ class gmail_inbound_domain_spoofing_protection_enabled(Check):
 
             enabled = gmail_client.policies.detect_inbound_domain_spoofing
             consequence = gmail_client.policies.inbound_domain_spoofing_consequence
+            domain = gmail_client.provider.identity.domain
 
             if enabled is False:
                 report.status = "FAIL"
                 report.status_extended = (
                     f"Protection against inbound emails spoofing your domain "
-                    f"is disabled in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"Enable the protection and configure a protective action."
+                    f"is disabled in domain {domain}. "
+                    f"Enable the protection and set the action to move the "
+                    f"email to spam."
                 )
-            elif consequence == "NO_ACTION":
+            elif not is_protective(consequence):
                 report.status = "FAIL"
                 report.status_extended = (
                     f"Protection against inbound emails spoofing your domain "
-                    f"is set to take no action in domain "
-                    f"{gmail_client.provider.identity.domain}. "
-                    f"A protective action should be configured."
-                )
-            elif consequence is None:
-                report.status = "PASS"
-                report.status_extended = (
-                    f"Protection against inbound emails spoofing your domain "
-                    f"uses Google's secure default configuration (enabled) "
-                    f"in domain {gmail_client.provider.identity.domain}."
+                    f"{describe_consequence(consequence)} in domain {domain}. "
+                    f"The action should move the email to spam."
                 )
             else:
                 report.status = "PASS"
+                state = "is enabled" if enabled else "uses Google's default (enabled)"
                 report.status_extended = (
                     f"Protection against inbound emails spoofing your domain "
-                    f"is enabled with consequence '{consequence}' "
-                    f"in domain {gmail_client.provider.identity.domain}."
+                    f"{state} with action '{consequence}' in domain "
+                    f"{domain}."
                 )
 
             findings.append(report)

prowler/providers/googleworkspace/services/gmail/lib/spoofing.py

@@ -0,0 +1,33 @@
+"""Helpers to evaluate the action configured for Gmail spoofing protections."""
+
+from typing import Optional
+
+# Actions that actually keep the message away from the inbox. CIS Google
+# Workspace 3.1.3.4.3.1, 3.1.3.4.3.2, 3.1.3.4.3.3 and 3.1.3.4.3.5 all require
+# the action to be "Move email to spam"; quarantining is stricter and also
+# satisfies the recommendation.
+PROTECTIVE_CONSEQUENCES = {"SPAM_FOLDER", "QUARANTINE"}
+
+# Google leaves these protections enabled but set to "Keep email in inbox and
+# show warning", which the benchmark does not accept, so an unset action is
+# evaluated as the insecure default rather than as a secure one.
+UNSET_CONSEQUENCE_DESCRIPTION = (
+    "uses Google's default action (keep email in inbox and show a warning)"
+)
+
+CONSEQUENCE_DESCRIPTIONS = {
+    "NO_ACTION": "is set to take no action",
+    "WARNING": "is set to keep the email in the inbox and show a warning",
+}
+
+
+def describe_consequence(consequence: Optional[str]) -> str:
+    """Return a human-readable description of a non-protective action."""
+    if consequence is None:
+        return UNSET_CONSEQUENCE_DESCRIPTION
+    return CONSEQUENCE_DESCRIPTIONS.get(consequence, f"is set to '{consequence}'")
+
+
+def is_protective(consequence: Optional[str]) -> bool:
+    """Whether the configured action moves the message out of the inbox."""
+    return consequence in PROTECTIVE_CONSEQUENCES

prowler/providers/googleworkspace/services/rules/lib/alerts.py

@@ -0,0 +1,127 @@
+"""Shared evaluation of the system-defined alert rules audited by CIS section 6."""
+
+from typing import TYPE_CHECKING, List
+
+from prowler.lib.check.models import CheckReportGoogleWorkspace
+
+if TYPE_CHECKING:
+    from prowler.providers.googleworkspace.services.rules.rules_service import Rules
+
+# A rule classified above what the benchmark asks for is stricter, not weaker,
+# so severities are compared by rank instead of by equality.
+SEVERITY_RANK = {"LOW": 1, "MEDIUM": 2, "HIGH": 3}
+
+# The rule can be active while its delivery to the alert center is switched
+# off, which is what the audit's "Ensure that Alerts is set to On" checks.
+ALERT_CENTER_DISABLED = "DISABLED"
+
+
+def _severity_issue(severity: str, minimum_severity: str) -> str:
+    """Return why a severity does not meet the benchmark, or an empty string."""
+    if severity is None:
+        return f"severity is not configured (should be at least {minimum_severity})"
+    rank = SEVERITY_RANK.get(severity)
+    if rank is None:
+        return (
+            f"severity is {severity}, which is not one of "
+            f"{', '.join(SEVERITY_RANK)}, so it could not be compared against "
+            f"the {minimum_severity} the benchmark asks for"
+        )
+    if rank < SEVERITY_RANK[minimum_severity]:
+        return f"severity is {severity} (should be at least {minimum_severity})"
+    return ""
+
+
+def evaluate_system_defined_alert(
+    client: "Rules",
+    metadata: dict,
+    rule_name: str,
+    minimum_severity: str,
+) -> List[CheckReportGoogleWorkspace]:
+    """Report on one system-defined alert rule against the CIS audit procedure.
+
+    Every recommendation in CIS section 6 asks for the rule to be on, to notify
+    by email, to include all super administrators as recipients and to carry a
+    minimum severity. Returns no finding at all when the policies could not be
+    fetched or the rule is not among the ones the client collected.
+    """
+    findings = []
+
+    if not client.policies_fetched:
+        return findings
+
+    for alert in client.system_defined_alerts:
+        if alert.display_name != rule_name:
+            continue
+
+        domain = client.provider.identity.domain
+        report = CheckReportGoogleWorkspace(
+            metadata=metadata,
+            resource=alert,
+            resource_id=f"systemDefinedAlert/{rule_name}",
+            resource_name=rule_name,
+            customer_id=client.provider.identity.customer_id,
+        )
+
+        if alert.from_default:
+            # Nothing was observed: the state below is Google's documented
+            # default and the severity has no documented default at all.
+            if alert.state != "ACTIVE":
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"System-defined alert rule '{rule_name}' was not returned "
+                    f"by the API in domain {domain} and Google's default for it "
+                    f"is OFF."
+                )
+            else:
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"System-defined alert rule '{rule_name}' was not returned "
+                    f"by the API in domain {domain}, so its configuration could "
+                    f"not be verified. Review it in the Admin console: it should "
+                    f"be ON, notify all super administrators by email and be set "
+                    f"to {minimum_severity} severity or higher."
+                )
+            findings.append(report)
+            continue
+
+        issues = []
+
+        if alert.state != "ACTIVE":
+            issues.append("alert is OFF")
+
+        # Only an explicit DISABLED fails. The API does not return this field
+        # even for a rule that is ON and has a severity set, and a severity
+        # cannot be configured for the alert center while delivery is off, so
+        # treating its absence as unverified would leave every one of these
+        # checks permanently MANUAL.
+        if alert.alert_center_status == ALERT_CENTER_DISABLED:
+            issues.append("the alert is not sent to the alert center")
+
+        if not alert.email_notifications_enabled:
+            issues.append("email notifications are disabled")
+        elif not alert.all_super_admins:
+            issues.append("email recipients do not include all super administrators")
+
+        severity = _severity_issue(alert.severity, minimum_severity)
+        if severity:
+            issues.append(severity)
+
+        if issues:
+            report.status = "FAIL"
+            report.status_extended = (
+                f"System-defined alert rule '{rule_name}' is not properly "
+                f"configured in domain {domain}: {', '.join(issues)}."
+            )
+        else:
+            report.status = "PASS"
+            report.status_extended = (
+                f"System-defined alert rule '{rule_name}' is properly "
+                f"configured in domain {domain}: alert is ON, email "
+                f"notifications are enabled, recipients include all super "
+                f"administrators and severity is {alert.severity}."
+            )
+
+        findings.append(report)
+
+    return findings

prowler/providers/googleworkspace/services/rules/rules_admin_privilege_granted_alert_configured/rules_admin_privilege_granted_alert_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "monitoring",
-  "Description": "The **User granted Admin privilege** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, and recipients set to all super administrators. This ensures administrators are notified when a user is given elevated admin privileges.",
+  "Description": "The **User granted Admin privilege** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, recipients set to all super administrators, and severity set to Medium or higher. This ensures administrators are notified when a user is given elevated admin privileges.",
   "Risk": "Without this alert enabled, administrators will not be notified when users receive **elevated admin privileges**. Unauthorized privilege escalation could indicate account compromise or insider threats and requires immediate verification.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/rules/rules_admin_privilege_granted_alert_configured/rules_admin_privilege_granted_alert_configured.py

@@ -1,61 +1,25 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
 from prowler.providers.googleworkspace.services.rules.rules_client import (
     rules_client,
 )
 
 RULE_NAME = "User granted Admin privilege"
+MINIMUM_SEVERITY = "MEDIUM"
 
 
 class rules_admin_privilege_granted_alert_configured(Check):
-    """Check that the User granted Admin privilege system-defined alert rule is fully configured."""
+    """Check that the User granted Admin privilege system-defined alert rule is fully configured.
+
+    CIS 6.4 requires the rule to be on, to notify by email, to include all
+    super administrators as recipients and to be set to MEDIUM severity or higher.
+    """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
-        findings = []
-
-        if rules_client.policies_fetched:
-            for alert in rules_client.system_defined_alerts:
-                if alert.display_name != RULE_NAME:
-                    continue
-
-                domain = rules_client.provider.identity.domain
-                report = CheckReportGoogleWorkspace(
-                    metadata=self.metadata(),
-                    resource=alert,
-                    resource_id=f"systemDefinedAlert/{RULE_NAME}",
-                    resource_name=RULE_NAME,
-                    customer_id=rules_client.provider.identity.customer_id,
-                )
-
-                is_active = alert.state == "ACTIVE"
-                has_recipients = alert.email_notifications_enabled
-                all_super_admins = alert.all_super_admins
-
-                if is_active and has_recipients and all_super_admins:
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is properly "
-                        f"configured in domain {domain}: alert is ON, email "
-                        f"notifications are enabled, and recipients include "
-                        f"all super administrators."
-                    )
-                else:
-                    report.status = "FAIL"
-                    issues = []
-                    if not is_active:
-                        issues.append("alert is OFF")
-                    if not has_recipients:
-                        issues.append("email notifications are disabled")
-                    elif not all_super_admins:
-                        issues.append(
-                            "email recipients do not include all super administrators"
-                        )
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is not properly "
-                        f"configured in domain {domain}: {', '.join(issues)}."
-                    )
-
-                findings.append(report)
-
-        return findings
+        return evaluate_system_defined_alert(
+            rules_client, self.metadata(), RULE_NAME, MINIMUM_SEVERITY
+        )

prowler/providers/googleworkspace/services/rules/rules_gmail_employee_spoofing_alert_configured/rules_gmail_employee_spoofing_alert_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "monitoring",
-  "Description": "The **Gmail potential employee spoofing** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, and recipients set to all super administrators. This ensures administrators are notified when incoming messages have a sender name matching the directory but from an external domain.",
+  "Description": "The **Gmail potential employee spoofing** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, recipients set to all super administrators, and severity set to Medium or higher. This ensures administrators are notified when incoming messages have a sender name matching the directory but from an external domain.",
   "Risk": "Without this alert enabled, administrators will not be notified of potential **employee spoofing via email**. Attackers may impersonate internal employees using external email addresses to conduct phishing attacks against the organization.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/rules/rules_gmail_employee_spoofing_alert_configured/rules_gmail_employee_spoofing_alert_configured.py

@@ -1,61 +1,25 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
 from prowler.providers.googleworkspace.services.rules.rules_client import (
     rules_client,
 )
 
 RULE_NAME = "Gmail potential employee spoofing"
+MINIMUM_SEVERITY = "MEDIUM"
 
 
 class rules_gmail_employee_spoofing_alert_configured(Check):
-    """Check that the Gmail potential employee spoofing system-defined alert rule is fully configured."""
+    """Check that the Gmail potential employee spoofing system-defined alert rule is fully configured.
+
+    CIS 6.8 requires the rule to be on, to notify by email, to include all
+    super administrators as recipients and to be set to MEDIUM severity or higher.
+    """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
-        findings = []
-
-        if rules_client.policies_fetched:
-            for alert in rules_client.system_defined_alerts:
-                if alert.display_name != RULE_NAME:
-                    continue
-
-                domain = rules_client.provider.identity.domain
-                report = CheckReportGoogleWorkspace(
-                    metadata=self.metadata(),
-                    resource=alert,
-                    resource_id=f"systemDefinedAlert/{RULE_NAME}",
-                    resource_name=RULE_NAME,
-                    customer_id=rules_client.provider.identity.customer_id,
-                )
-
-                is_active = alert.state == "ACTIVE"
-                has_recipients = alert.email_notifications_enabled
-                all_super_admins = alert.all_super_admins
-
-                if is_active and has_recipients and all_super_admins:
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is properly "
-                        f"configured in domain {domain}: alert is ON, email "
-                        f"notifications are enabled, and recipients include "
-                        f"all super administrators."
-                    )
-                else:
-                    report.status = "FAIL"
-                    issues = []
-                    if not is_active:
-                        issues.append("alert is OFF")
-                    if not has_recipients:
-                        issues.append("email notifications are disabled")
-                    elif not all_super_admins:
-                        issues.append(
-                            "email recipients do not include all super administrators"
-                        )
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is not properly "
-                        f"configured in domain {domain}: {', '.join(issues)}."
-                    )
-
-                findings.append(report)
-
-        return findings
+        return evaluate_system_defined_alert(
+            rules_client, self.metadata(), RULE_NAME, MINIMUM_SEVERITY
+        )

prowler/providers/googleworkspace/services/rules/rules_government_backed_attacks_alert_configured/rules_government_backed_attacks_alert_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "high",
   "ResourceType": "NotDefined",
   "ResourceGroup": "monitoring",
-  "Description": "The **Government-backed attacks** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, and recipients set to all super administrators. This ensures administrators are notified when Google believes users are being targeted by a government-backed attacker.",
+  "Description": "The **Government-backed attacks** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, recipients set to all super administrators, and severity set to High or higher. This ensures administrators are notified when Google believes users are being targeted by a government-backed attacker.",
   "Risk": "Without this alert enabled, administrators will not be notified of potential **government-backed attacks** targeting their users. These attacks are sophisticated and require immediate response to protect affected accounts and investigate the threat.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/rules/rules_government_backed_attacks_alert_configured/rules_government_backed_attacks_alert_configured.py

@@ -1,61 +1,25 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
 from prowler.providers.googleworkspace.services.rules.rules_client import (
     rules_client,
 )
 
 RULE_NAME = "Government-backed attacks"
+MINIMUM_SEVERITY = "HIGH"
 
 
 class rules_government_backed_attacks_alert_configured(Check):
-    """Check that the Government-backed attacks system-defined alert rule is fully configured."""
+    """Check that the Government-backed attacks system-defined alert rule is fully configured.
+
+    CIS 6.2 requires the rule to be on, to notify by email, to include all
+    super administrators as recipients and to be set to HIGH severity or higher.
+    """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
-        findings = []
-
-        if rules_client.policies_fetched:
-            for alert in rules_client.system_defined_alerts:
-                if alert.display_name != RULE_NAME:
-                    continue
-
-                domain = rules_client.provider.identity.domain
-                report = CheckReportGoogleWorkspace(
-                    metadata=self.metadata(),
-                    resource=alert,
-                    resource_id=f"systemDefinedAlert/{RULE_NAME}",
-                    resource_name=RULE_NAME,
-                    customer_id=rules_client.provider.identity.customer_id,
-                )
-
-                is_active = alert.state == "ACTIVE"
-                has_recipients = alert.email_notifications_enabled
-                all_super_admins = alert.all_super_admins
-
-                if is_active and has_recipients and all_super_admins:
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is properly "
-                        f"configured in domain {domain}: alert is ON, email "
-                        f"notifications are enabled, and recipients include "
-                        f"all super administrators."
-                    )
-                else:
-                    report.status = "FAIL"
-                    issues = []
-                    if not is_active:
-                        issues.append("alert is OFF")
-                    if not has_recipients:
-                        issues.append("email notifications are disabled")
-                    elif not all_super_admins:
-                        issues.append(
-                            "email recipients do not include all super administrators"
-                        )
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is not properly "
-                        f"configured in domain {domain}: {', '.join(issues)}."
-                    )
-
-                findings.append(report)
-
-        return findings
+        return evaluate_system_defined_alert(
+            rules_client, self.metadata(), RULE_NAME, MINIMUM_SEVERITY
+        )

prowler/providers/googleworkspace/services/rules/rules_leaked_password_alert_configured/rules_leaked_password_alert_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "monitoring",
-  "Description": "The **Leaked password** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, and recipients set to all super administrators. This ensures administrators are notified when Google detects compromised credentials requiring a password reset.",
+  "Description": "The **Leaked password** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, recipients set to all super administrators, and severity set to Medium or higher. This ensures administrators are notified when Google detects compromised credentials requiring a password reset.",
   "Risk": "Without this alert enabled, administrators will not be notified when Google detects that a user's **credentials have been compromised** in a publicized breach. The user likely reused their password at another site that was breached, and their account requires an immediate password change.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/rules/rules_leaked_password_alert_configured/rules_leaked_password_alert_configured.py

@@ -1,61 +1,25 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
 from prowler.providers.googleworkspace.services.rules.rules_client import (
     rules_client,
 )
 
 RULE_NAME = "Leaked password"
+MINIMUM_SEVERITY = "MEDIUM"
 
 
 class rules_leaked_password_alert_configured(Check):
-    """Check that the Leaked password system-defined alert rule is fully configured."""
+    """Check that the Leaked password system-defined alert rule is fully configured.
+
+    CIS 6.7 requires the rule to be on, to notify by email, to include all
+    super administrators as recipients and to be set to MEDIUM severity or higher.
+    """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
-        findings = []
-
-        if rules_client.policies_fetched:
-            for alert in rules_client.system_defined_alerts:
-                if alert.display_name != RULE_NAME:
-                    continue
-
-                domain = rules_client.provider.identity.domain
-                report = CheckReportGoogleWorkspace(
-                    metadata=self.metadata(),
-                    resource=alert,
-                    resource_id=f"systemDefinedAlert/{RULE_NAME}",
-                    resource_name=RULE_NAME,
-                    customer_id=rules_client.provider.identity.customer_id,
-                )
-
-                is_active = alert.state == "ACTIVE"
-                has_recipients = alert.email_notifications_enabled
-                all_super_admins = alert.all_super_admins
-
-                if is_active and has_recipients and all_super_admins:
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is properly "
-                        f"configured in domain {domain}: alert is ON, email "
-                        f"notifications are enabled, and recipients include "
-                        f"all super administrators."
-                    )
-                else:
-                    report.status = "FAIL"
-                    issues = []
-                    if not is_active:
-                        issues.append("alert is OFF")
-                    if not has_recipients:
-                        issues.append("email notifications are disabled")
-                    elif not all_super_admins:
-                        issues.append(
-                            "email recipients do not include all super administrators"
-                        )
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is not properly "
-                        f"configured in domain {domain}: {', '.join(issues)}."
-                    )
-
-                findings.append(report)
-
-        return findings
+        return evaluate_system_defined_alert(
+            rules_client, self.metadata(), RULE_NAME, MINIMUM_SEVERITY
+        )

prowler/providers/googleworkspace/services/rules/rules_password_changed_alert_configured/rules_password_changed_alert_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "monitoring",
-  "Description": "The **User's password changed** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, and recipients set to all super administrators. This ensures administrators are promptly notified when user passwords are changed.",
+  "Description": "The **User's password changed** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, recipients set to all super administrators, and severity set to Medium or higher. This ensures administrators are promptly notified when user passwords are changed.",
   "Risk": "Without this alert enabled, administrators will not be notified when user passwords are changed. This could allow **credential compromise and account takeover** to go undetected, giving attackers time to establish persistence.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/rules/rules_password_changed_alert_configured/rules_password_changed_alert_configured.py

@@ -1,61 +1,25 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
 from prowler.providers.googleworkspace.services.rules.rules_client import (
     rules_client,
 )
 
 RULE_NAME = "User's password changed"
+MINIMUM_SEVERITY = "MEDIUM"
 
 
 class rules_password_changed_alert_configured(Check):
-    """Check that the User's password changed system-defined alert rule is fully configured."""
+    """Check that the User's password changed system-defined alert rule is fully configured.
+
+    CIS 6.1 requires the rule to be on, to notify by email, to include all
+    super administrators as recipients and to be set to MEDIUM severity or higher.
+    """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
-        findings = []
-
-        if rules_client.policies_fetched:
-            for alert in rules_client.system_defined_alerts:
-                if alert.display_name != RULE_NAME:
-                    continue
-
-                domain = rules_client.provider.identity.domain
-                report = CheckReportGoogleWorkspace(
-                    metadata=self.metadata(),
-                    resource=alert,
-                    resource_id=f"systemDefinedAlert/{RULE_NAME}",
-                    resource_name=RULE_NAME,
-                    customer_id=rules_client.provider.identity.customer_id,
-                )
-
-                is_active = alert.state == "ACTIVE"
-                has_recipients = alert.email_notifications_enabled
-                all_super_admins = alert.all_super_admins
-
-                if is_active and has_recipients and all_super_admins:
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is properly "
-                        f"configured in domain {domain}: alert is ON, email "
-                        f"notifications are enabled, and recipients include "
-                        f"all super administrators."
-                    )
-                else:
-                    report.status = "FAIL"
-                    issues = []
-                    if not is_active:
-                        issues.append("alert is OFF")
-                    if not has_recipients:
-                        issues.append("email notifications are disabled")
-                    elif not all_super_admins:
-                        issues.append(
-                            "email recipients do not include all super administrators"
-                        )
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is not properly "
-                        f"configured in domain {domain}: {', '.join(issues)}."
-                    )
-
-                findings.append(report)
-
-        return findings
+        return evaluate_system_defined_alert(
+            rules_client, self.metadata(), RULE_NAME, MINIMUM_SEVERITY
+        )

prowler/providers/googleworkspace/services/rules/rules_service.py

@@ -90,6 +90,7 @@ class Rules(GoogleWorkspaceService):
                         state=default_state,
                         email_notifications_enabled=is_active_default,
                         all_super_admins=is_active_default,
+                        from_default=True,
                     )
                     logger.debug(
                         f"System-defined alert rule (default): {rule_name} "
@@ -119,7 +120,12 @@ class Rules(GoogleWorkspaceService):
         state = value.get("state", "INACTIVE")
 
         alert_center_action = value.get("action", {}).get("alertCenterAction", {})
-        severity = alert_center_action.get("alertCenterConfig", {}).get("severity")
+        alert_center_config = alert_center_action.get("alertCenterConfig", {})
+        severity = alert_center_config.get("severity")
+        # CIS remediation step 6: "Select Send to alert center (This will result
+        # in the alert being set to On)", so this is the toggle the audit's
+        # "Ensure that Alerts is set to On" refers to.
+        alert_center_status = alert_center_config.get("status")
         recipients = alert_center_action.get("recipients", [])
 
         all_super_admins = any(r.get("allSuperAdmins") is True for r in recipients)
@@ -128,6 +134,7 @@ class Rules(GoogleWorkspaceService):
             display_name=display_name,
             state=state,
             severity=severity,
+            alert_center_status=alert_center_status,
             email_notifications_enabled=len(recipients) > 0,
             all_super_admins=all_super_admins,
         )
@@ -139,5 +146,10 @@ class SystemDefinedAlert(BaseModel):
     display_name: str
     state: str = "INACTIVE"
     severity: Optional[str] = None
+    # rule.system_defined_alerts action.alertCenterAction.alertCenterConfig.status
+    alert_center_status: Optional[str] = None
     email_notifications_enabled: bool = False
     all_super_admins: bool = False
+    # True when the API returned no policy for the rule and the values above
+    # were inferred from Google's documented defaults rather than observed.
+    from_default: bool = False

prowler/providers/googleworkspace/services/rules/rules_suspicious_activity_suspension_alert_configured/rules_suspicious_activity_suspension_alert_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "high",
   "ResourceType": "NotDefined",
   "ResourceGroup": "monitoring",
-  "Description": "The **User suspended due to suspicious activity** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, and recipients set to all super administrators. This ensures administrators are notified when Google suspends an account due to a potential compromise.",
+  "Description": "The **User suspended due to suspicious activity** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, recipients set to all super administrators, and severity set to High or higher. This ensures administrators are notified when Google suspends an account due to a potential compromise.",
   "Risk": "Without this alert enabled, administrators will not be promptly notified when Google **suspends a user account** due to detected compromise. The suspended user cannot work, and the underlying security incident requires immediate investigation.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/rules/rules_suspicious_activity_suspension_alert_configured/rules_suspicious_activity_suspension_alert_configured.py

@@ -1,61 +1,25 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
 from prowler.providers.googleworkspace.services.rules.rules_client import (
     rules_client,
 )
 
 RULE_NAME = "User suspended due to suspicious activity"
+MINIMUM_SEVERITY = "HIGH"
 
 
 class rules_suspicious_activity_suspension_alert_configured(Check):
-    """Check that the User suspended due to suspicious activity system-defined alert rule is fully configured."""
+    """Check that the User suspended due to suspicious activity system-defined alert rule is fully configured.
+
+    CIS 6.3 requires the rule to be on, to notify by email, to include all
+    super administrators as recipients and to be set to HIGH severity or higher.
+    """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
-        findings = []
-
-        if rules_client.policies_fetched:
-            for alert in rules_client.system_defined_alerts:
-                if alert.display_name != RULE_NAME:
-                    continue
-
-                domain = rules_client.provider.identity.domain
-                report = CheckReportGoogleWorkspace(
-                    metadata=self.metadata(),
-                    resource=alert,
-                    resource_id=f"systemDefinedAlert/{RULE_NAME}",
-                    resource_name=RULE_NAME,
-                    customer_id=rules_client.provider.identity.customer_id,
-                )
-
-                is_active = alert.state == "ACTIVE"
-                has_recipients = alert.email_notifications_enabled
-                all_super_admins = alert.all_super_admins
-
-                if is_active and has_recipients and all_super_admins:
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is properly "
-                        f"configured in domain {domain}: alert is ON, email "
-                        f"notifications are enabled, and recipients include "
-                        f"all super administrators."
-                    )
-                else:
-                    report.status = "FAIL"
-                    issues = []
-                    if not is_active:
-                        issues.append("alert is OFF")
-                    if not has_recipients:
-                        issues.append("email notifications are disabled")
-                    elif not all_super_admins:
-                        issues.append(
-                            "email recipients do not include all super administrators"
-                        )
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is not properly "
-                        f"configured in domain {domain}: {', '.join(issues)}."
-                    )
-
-                findings.append(report)
-
-        return findings
+        return evaluate_system_defined_alert(
+            rules_client, self.metadata(), RULE_NAME, MINIMUM_SEVERITY
+        )

prowler/providers/googleworkspace/services/rules/rules_suspicious_login_alert_configured/rules_suspicious_login_alert_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "low",
   "ResourceType": "NotDefined",
   "ResourceGroup": "monitoring",
-  "Description": "The **Suspicious login** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, and recipients set to all super administrators. This ensures administrators are notified when Google detects a sign-in attempt that does not match a user's normal behavior.",
+  "Description": "The **Suspicious login** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, recipients set to all super administrators, and severity set to Low or higher. This ensures administrators are notified when Google detects a sign-in attempt that does not match a user's normal behavior.",
   "Risk": "Without this alert enabled, administrators will not be notified of **suspicious login attempts** such as sign-ins from unusual locations. This could indicate an active attack using previously obtained credentials.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/rules/rules_suspicious_login_alert_configured/rules_suspicious_login_alert_configured.py

@@ -1,61 +1,25 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
 from prowler.providers.googleworkspace.services.rules.rules_client import (
     rules_client,
 )
 
 RULE_NAME = "Suspicious login"
+MINIMUM_SEVERITY = "LOW"
 
 
 class rules_suspicious_login_alert_configured(Check):
-    """Check that the Suspicious login system-defined alert rule is fully configured."""
+    """Check that the Suspicious login system-defined alert rule is fully configured.
+
+    CIS 6.6 requires the rule to be on, to notify by email, to include all
+    super administrators as recipients and to be set to LOW severity or higher.
+    """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
-        findings = []
-
-        if rules_client.policies_fetched:
-            for alert in rules_client.system_defined_alerts:
-                if alert.display_name != RULE_NAME:
-                    continue
-
-                domain = rules_client.provider.identity.domain
-                report = CheckReportGoogleWorkspace(
-                    metadata=self.metadata(),
-                    resource=alert,
-                    resource_id=f"systemDefinedAlert/{RULE_NAME}",
-                    resource_name=RULE_NAME,
-                    customer_id=rules_client.provider.identity.customer_id,
-                )
-
-                is_active = alert.state == "ACTIVE"
-                has_recipients = alert.email_notifications_enabled
-                all_super_admins = alert.all_super_admins
-
-                if is_active and has_recipients and all_super_admins:
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is properly "
-                        f"configured in domain {domain}: alert is ON, email "
-                        f"notifications are enabled, and recipients include "
-                        f"all super administrators."
-                    )
-                else:
-                    report.status = "FAIL"
-                    issues = []
-                    if not is_active:
-                        issues.append("alert is OFF")
-                    if not has_recipients:
-                        issues.append("email notifications are disabled")
-                    elif not all_super_admins:
-                        issues.append(
-                            "email recipients do not include all super administrators"
-                        )
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is not properly "
-                        f"configured in domain {domain}: {', '.join(issues)}."
-                    )
-
-                findings.append(report)
-
-        return findings
+        return evaluate_system_defined_alert(
+            rules_client, self.metadata(), RULE_NAME, MINIMUM_SEVERITY
+        )

prowler/providers/googleworkspace/services/rules/rules_suspicious_programmatic_login_alert_configured/rules_suspicious_programmatic_login_alert_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "low",
   "ResourceType": "NotDefined",
   "ResourceGroup": "monitoring",
-  "Description": "The **Suspicious programmatic login** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, and recipients set to all super administrators. This ensures administrators are notified when Google detects suspicious login attempts from applications or programs.",
+  "Description": "The **Suspicious programmatic login** system-defined alert rule should be enabled with alerts sent to the alert center, email notifications turned on, recipients set to all super administrators, and severity set to Low or higher. This ensures administrators are notified when Google detects suspicious login attempts from applications or programs.",
   "Risk": "Without this alert enabled, administrators will not be notified of **suspicious programmatic login attempts**. This could indicate automated credential stuffing or unauthorized API access using compromised credentials.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/rules/rules_suspicious_programmatic_login_alert_configured/rules_suspicious_programmatic_login_alert_configured.py

@@ -1,61 +1,25 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
 from prowler.providers.googleworkspace.services.rules.rules_client import (
     rules_client,
 )
 
 RULE_NAME = "Suspicious programmatic login"
+MINIMUM_SEVERITY = "LOW"
 
 
 class rules_suspicious_programmatic_login_alert_configured(Check):
-    """Check that the Suspicious programmatic login system-defined alert rule is fully configured."""
+    """Check that the Suspicious programmatic login system-defined alert rule is fully configured.
+
+    CIS 6.5 requires the rule to be on, to notify by email, to include all
+    super administrators as recipients and to be set to LOW severity or higher.
+    """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
-        findings = []
-
-        if rules_client.policies_fetched:
-            for alert in rules_client.system_defined_alerts:
-                if alert.display_name != RULE_NAME:
-                    continue
-
-                domain = rules_client.provider.identity.domain
-                report = CheckReportGoogleWorkspace(
-                    metadata=self.metadata(),
-                    resource=alert,
-                    resource_id=f"systemDefinedAlert/{RULE_NAME}",
-                    resource_name=RULE_NAME,
-                    customer_id=rules_client.provider.identity.customer_id,
-                )
-
-                is_active = alert.state == "ACTIVE"
-                has_recipients = alert.email_notifications_enabled
-                all_super_admins = alert.all_super_admins
-
-                if is_active and has_recipients and all_super_admins:
-                    report.status = "PASS"
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is properly "
-                        f"configured in domain {domain}: alert is ON, email "
-                        f"notifications are enabled, and recipients include "
-                        f"all super administrators."
-                    )
-                else:
-                    report.status = "FAIL"
-                    issues = []
-                    if not is_active:
-                        issues.append("alert is OFF")
-                    if not has_recipients:
-                        issues.append("email notifications are disabled")
-                    elif not all_super_admins:
-                        issues.append(
-                            "email recipients do not include all super administrators"
-                        )
-                    report.status_extended = (
-                        f"System-defined alert rule '{RULE_NAME}' is not properly "
-                        f"configured in domain {domain}: {', '.join(issues)}."
-                    )
-
-                findings.append(report)
-
-        return findings
+        return evaluate_system_defined_alert(
+            rules_client, self.metadata(), RULE_NAME, MINIMUM_SEVERITY
+        )

prowler/providers/googleworkspace/services/security/lib/durations.py

@@ -0,0 +1,89 @@
+"""Helpers for the duration and timestamp values of the Cloud Identity security policies."""
+
+import re
+from datetime import datetime, timezone
+from typing import Optional
+
+from dateutil import parser as date_parser
+
+_DURATION = re.compile(r"^(\d+(?:\.\d+)?)s$")
+
+ONE_HOUR_SECONDS = 3600
+ONE_DAY_SECONDS = 86400
+TWO_WEEKS_SECONDS = 1209600
+ONE_YEAR_SECONDS = 31536000
+
+# The API reports enforcement being OFF as the protobuf zero-value Timestamp
+# rather than as a null or an empty string.
+_ENFORCEMENT_OFF_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
+
+
+def parse_duration_seconds(value: Optional[str]) -> Optional[int]:
+    """Return the seconds in a protobuf duration string such as "1209600s".
+
+    Returns None when the value is missing or not a duration Prowler knows how
+    to read, so callers can tell "not configured" apart from a real length.
+    """
+    if not value or not isinstance(value, str):
+        return None
+    match = _DURATION.match(value.strip())
+    if not match:
+        return None
+    return int(float(match.group(1)))
+
+
+def format_duration(value: Optional[str]) -> str:
+    """Render a duration string in the largest whole unit, for a finding message."""
+    seconds = parse_duration_seconds(value)
+    if seconds is None:
+        return "not configured"
+    if seconds == 0:
+        return "none"
+    for unit_seconds, name in (
+        (ONE_DAY_SECONDS, "day"),
+        (ONE_HOUR_SECONDS, "hour"),
+    ):
+        units = seconds / unit_seconds
+        if units.is_integer():
+            return f"{int(units)} {name}(s)"
+    return f"{seconds} second(s)"
+
+
+def _parse_timestamp(value: Optional[str]) -> Optional[datetime]:
+    """Parse an API timestamp, tolerating any fractional-second precision.
+
+    protobuf emits up to nanosecond precision, which `datetime.fromisoformat`
+    rejects before Python 3.11, so the shared dateutil parser is used instead.
+    """
+    if not value or not isinstance(value, str):
+        return None
+    try:
+        parsed = date_parser.isoparse(value)
+    except (ValueError, OverflowError):
+        return None
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+def enforcement_issue(
+    enforced_from: Optional[str],
+    allow_scheduled: bool = False,
+    now: Optional[datetime] = None,
+) -> Optional[str]:
+    """Return why 2-Step Verification enforcement is not in effect, or None.
+
+    Google accepts a future start date, which means the policy is scheduled but
+    not yet applied to anyone. CIS 4.1.1.2 accepts "On from <date>" explicitly
+    while 4.1.1.1 and 4.1.1.3 ask for plain "On", hence `allow_scheduled`.
+    """
+    if not enforced_from:
+        return "enforcement is not configured and defaults to OFF"
+    parsed = _parse_timestamp(enforced_from)
+    if parsed is None:
+        return f"the enforcement start date '{enforced_from}' could not be read"
+    if parsed <= _ENFORCEMENT_OFF_EPOCH:
+        return "enforcement is set to OFF"
+    if not allow_scheduled and parsed > (now or datetime.now(timezone.utc)):
+        return f"enforcement does not start until {enforced_from}"
+    return None

prowler/providers/googleworkspace/services/security/lib/scope.py

@@ -0,0 +1,46 @@
+"""Whether the domain-wide policy values describe what every user actually gets."""
+
+from typing import FrozenSet, List, Optional
+
+
+def _listing(settings: List[str]) -> str:
+    return f"{', '.join(settings)} {'are' if len(settings) > 1 else 'is'}"
+
+
+def unevaluable_reason(policies, evaluated_settings: FrozenSet[str]) -> Optional[str]:
+    """Return why the domain-wide values cannot be judged at all, or None.
+
+    In both cases the values a check would read were never reported, so every
+    condition it evaluates would be built on Prowler's own defaults.
+    """
+    if policies.unresolved_scope:
+        return (
+            "the root organizational unit could not be resolved, so the "
+            "domain-wide policies could not be told apart from the ones scoped "
+            "to an organizational unit"
+        )
+    unobserved = sorted(set(policies.unobserved_settings) & evaluated_settings)
+    if unobserved:
+        return (
+            f"{_listing(unobserved)} only configured for a group or an "
+            f"organizational unit, so no domain-wide value was reported"
+        )
+    return None
+
+
+def failures_shadowed_by_overrides(policies, failing_settings: FrozenSet[str]) -> bool:
+    """Whether every failing setting is also overridden below the domain."""
+    return bool(failing_settings) and failing_settings <= set(
+        policies.overridden_settings
+    )
+
+
+def override_caveat(policies, evaluated_settings: FrozenSet[str]) -> str:
+    """Return what a group or an OU also overrides on top of the domain, or an empty string."""
+    overridden = sorted(set(policies.overridden_settings) & evaluated_settings)
+    if not overridden:
+        return ""
+    return (
+        f"{_listing(overridden)} also overridden for at least one group or "
+        f"organizational unit, so this does not describe every user"
+    )

prowler/providers/googleworkspace/services/security/security_2sv_enforced/security_2sv_enforced.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "high",
   "ResourceType": "NotDefined",
   "ResourceGroup": "IAM",
-  "Description": "The domain-level policy **enforces 2-Step Verification (Multi-Factor Authentication)** for all users. 2-Step Verification requires users to present a second form of authentication beyond their password, significantly reducing the risk of account compromise.",
+  "Description": "The domain-level policy **enforces 2-Step Verification** for all users, allows users to turn it on, keeps the new user enrollment period at two weeks or less, disables device trust and excludes verification codes via text or phone call from the accepted methods.",
   "Risk": "Without 2-Step Verification enforcement, users can access their accounts with **only a password**. If credentials are compromised through phishing, credential stuffing, or data breaches, attackers gain **immediate access** to the user's account and organizational data without any additional verification.",
   "RelatedUrl": "",
   "AdditionalURLs": [
@@ -35,5 +35,5 @@
   "RelatedTo": [
     "security_2sv_hardware_keys_admins"
   ],
-  "Notes": ""
+  "Notes": "CIS 4.1.1.1 audits the group holding every admin role, but the Cloud Identity Policy API returns domain-wide policies only. This check evaluates the customer-level policy, which applies to administrators as well, so it cannot confirm a separate admin-role group is configured."
 }

prowler/providers/googleworkspace/services/security/security_2sv_enforced/security_2sv_enforced.py

@@ -1,17 +1,54 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.security.lib.durations import (
+    TWO_WEEKS_SECONDS,
+    enforcement_issue,
+    format_duration,
+    parse_duration_seconds,
+)
+from prowler.providers.googleworkspace.services.security.lib.scope import (
+    failures_shadowed_by_overrides,
+    override_caveat,
+    unevaluable_reason,
+)
 from prowler.providers.googleworkspace.services.security.security_client import (
     security_client,
 )
 
+# "Methods: Any except verification codes via text, phone call". Listed as an
+# allow list so a value Prowler does not know cannot pass by not being "ALL".
+TELEPHONY_FREE_FACTOR_SETS = {
+    "NO_TELEPHONY",
+    "PASSKEY_ONLY",
+    "PASSKEY_PLUS_SECURITY_CODE",
+    "PASSKEY_PLUS_IP_BOUND_SECURITY_CODE",
+}
+
+# The settings this check reads, to tell whether an override reaches it.
+EVALUATED_SETTINGS = frozenset(
+    {
+        "security.two_step_verification_enrollment",
+        "security.two_step_verification_enforcement",
+        "security.two_step_verification_enforcement_factor",
+        "security.two_step_verification_device_trust",
+        "security.two_step_verification_grace_period",
+    }
+)
+
 
 class security_2sv_enforced(Check):
-    """Check that 2-Step Verification is enforced for all users.
+    """Check that 2-Step Verification is enforced following the CIS audit steps.
 
-    This check verifies that the domain-level policy enforces 2-Step
-    Verification (Multi-Factor Authentication) for all users, reducing
-    the risk of account compromise through stolen credentials.
+    CIS 4.1.1.1 and 4.1.1.3 ask for more than enforcement being on: users must
+    be allowed to turn 2-Step Verification on, the new user enrollment period
+    must not exceed two weeks, device trust must be off and verification codes
+    via text or phone call must not be an accepted method. Each of those is
+    evaluated here so the requirement cannot pass on enforcement alone.
+
+    Note: 4.1.1.1 audits the group holding every admin role, but the Cloud
+    Identity Policy API returns domain-wide policies only. This check evaluates
+    the customer-level policy, which applies to administrators too.
     """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
@@ -26,33 +63,125 @@ class security_2sv_enforced(Check):
                 customer_id=security_client.provider.identity.customer_id,
             )
 
-            enforced_from = security_client.policies.two_sv_enforced_from
-            # The API returns "1970-01-01T00:00:00Z" (protobuf zero-value
-            # Timestamp) when enforcement is OFF, not null or empty.
-            enforcement_off_epoch = "1970-01-01T00:00:00Z"
+            policies = security_client.policies
+            domain = security_client.provider.identity.domain
 
-            if enforced_from and enforced_from != enforcement_off_epoch:
-                report.status = "PASS"
+            unevaluable = unevaluable_reason(policies, EVALUATED_SETTINGS)
+            if unevaluable:
+                report.status = "MANUAL"
                 report.status_extended = (
-                    f"2-Step Verification enforcement is active "
-                    f"(enforced from {enforced_from}) "
-                    f"in domain {security_client.provider.identity.domain}."
+                    f"2-Step Verification could not be evaluated in domain "
+                    f"{domain}: {unevaluable}. Review it in the Admin console."
+                )
+                findings.append(report)
+                return findings
+
+            caveat = override_caveat(policies, EVALUATED_SETTINGS)
+            issues = []  # (setting, why it fails)
+
+            enforced_from = policies.two_sv_enforced_from
+            enforcement = enforcement_issue(enforced_from)
+            if enforcement:
+                issues.append(
+                    ("security.two_step_verification_enforcement", enforcement)
+                )
+
+            if policies.two_sv_allow_enrollment is False:
+                issues.append(
+                    (
+                        "security.two_step_verification_enrollment",
+                        "users are not allowed to turn on 2-Step Verification",
+                    )
+                )
+
+            # Google's default is no enrollment period, which is stricter than
+            # the two weeks the benchmark asks for, so only longer periods fail.
+            # A value Prowler cannot read fails closed rather than being skipped.
+            raw_grace_period = policies.two_sv_enrollment_grace_period
+            grace_period = parse_duration_seconds(raw_grace_period)
+            if raw_grace_period and grace_period is None:
+                issues.append(
+                    (
+                        "security.two_step_verification_grace_period",
+                        f"the new user enrollment period '{raw_grace_period}' "
+                        f"could not be read",
+                    )
+                )
+            elif grace_period is not None and grace_period > TWO_WEEKS_SECONDS:
+                issues.append(
+                    (
+                        "security.two_step_verification_grace_period",
+                        f"the new user enrollment period is "
+                        f"{format_duration(policies.two_sv_enrollment_grace_period)} "
+                        f"(should not exceed 2 weeks)",
+                    )
+                )
+
+            if policies.two_sv_allow_trusting_device is not False:
+                issues.append(
+                    (
+                        "security.two_step_verification_device_trust",
+                        (
+                            "users are allowed to trust their device"
+                            if policies.two_sv_allow_trusting_device
+                            else "device trust is not configured and defaults to allowed"
+                        ),
+                    )
+                )
+
+            factor_set = policies.two_sv_allowed_factor_set
+            if factor_set not in TELEPHONY_FREE_FACTOR_SETS:
+                issues.append(
+                    (
+                        "security.two_step_verification_enforcement_factor",
+                        (
+                            "the allowed methods are not configured and default to "
+                            "any method, including verification codes via text and "
+                            "phone call"
+                            if factor_set is None
+                            else f"the allowed methods are {factor_set}, which does "
+                            f"not exclude verification codes via text and phone call"
+                        ),
+                    )
+                )
+
+            failing_settings = frozenset(setting for setting, _ in issues)
+            reasons = "; ".join(text for _, text in issues)
+
+            if issues and failures_shadowed_by_overrides(policies, failing_settings):
+                # The audited scope (e.g. the admin group of 4.1.1.1) may get
+                # the overriding value, which the Policy API does not expose,
+                # so the domain-wide failure cannot be confirmed for it.
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"2-Step Verification is not enforced as required in the "
+                    f"domain-wide policy of {domain}: {reasons}. However, every "
+                    f"failing setting is also overridden for at least one group "
+                    f"or organizational unit, so the audited scope may be "
+                    f"configured correctly. Review those overrides in the Admin "
+                    f"console."
+                )
+            elif issues:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"2-Step Verification is not enforced as required in domain "
+                    f"{domain}: {reasons}." + (f" Note: {caveat}." if caveat else "")
+                )
+            elif caveat:
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"2-Step Verification meets the benchmark in the domain-wide "
+                    f"policy of {domain}, but {caveat}. Review those overrides "
+                    f"in the Admin console."
                 )
             else:
-                report.status = "FAIL"
-                if enforced_from is None:
-                    report.status_extended = (
-                        f"2-Step Verification enforcement is not configured "
-                        f"in domain {security_client.provider.identity.domain}. "
-                        f"The default is OFF. 2-Step Verification should be "
-                        f"enforced for all users."
-                    )
-                else:
-                    report.status_extended = (
-                        f"2-Step Verification enforcement is set to OFF "
-                        f"in domain {security_client.provider.identity.domain}. "
-                        f"2-Step Verification should be enforced for all users."
-                    )
+                report.status = "PASS"
+                report.status_extended = (
+                    f"2-Step Verification is enforced in domain {domain} "
+                    f"(enforced from {enforced_from}), device trust is disabled "
+                    f"and verification codes via text or phone call are not an "
+                    f"accepted method."
+                )
 
             findings.append(report)
 

prowler/providers/googleworkspace/services/security/security_2sv_hardware_keys_admins/security_2sv_hardware_keys_admins.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "high",
   "ResourceType": "NotDefined",
   "ResourceGroup": "IAM",
-  "Description": "The domain-level 2-Step Verification policy requires **hardware security keys only** as the allowed sign-in factor, providing the strongest phishing-resistant authentication. **Note**: the Policy API returns domain-wide policies only and cannot verify admin role-specific enforcement.",
+  "Description": "The domain-level 2-Step Verification policy requires **hardware security keys only** as the allowed sign-in factor, with enforcement on or scheduled and a policy suspension grace period of at most one day. **Note**: the Policy API returns domain-wide policies only and cannot verify admin role-specific enforcement.",
   "Risk": "When 2SV methods include **SMS, phone calls, or software-based authenticators**, users are vulnerable to **SIM swapping, SS7 attacks, and real-time phishing proxies** that can intercept one-time codes. Hardware security keys are resistant to all known remote phishing techniques.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/security/security_2sv_hardware_keys_admins/security_2sv_hardware_keys_admins.py

@@ -1,20 +1,50 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.security.lib.durations import (
+    ONE_DAY_SECONDS,
+    enforcement_issue,
+    format_duration,
+    parse_duration_seconds,
+)
+from prowler.providers.googleworkspace.services.security.lib.scope import (
+    failures_shadowed_by_overrides,
+    override_caveat,
+    unevaluable_reason,
+)
 from prowler.providers.googleworkspace.services.security.security_client import (
     security_client,
 )
 
+# "Methods: Only security key". The values that also accept security codes are
+# PASSKEY_PLUS_SECURITY_CODE and PASSKEY_PLUS_IP_BOUND_SECURITY_CODE, so
+# requiring this one covers the benchmark's "don't allow users to generate
+# security codes" step as well.
+SECURITY_KEYS_ONLY = "PASSKEY_ONLY"
+
+# The settings this check reads, to tell whether an override reaches it.
+EVALUATED_SETTINGS = frozenset(
+    {
+        "security.two_step_verification_enrollment",
+        "security.two_step_verification_enforcement",
+        "security.two_step_verification_enforcement_factor",
+        "security.two_step_verification_sign_in_code",
+    }
+)
+
 
 class security_2sv_hardware_keys_admins(Check):
     """Check that 2SV enforcement requires hardware security keys.
 
-    This check verifies that the domain-level 2-Step Verification enforcement
-    factor is set to security keys only, providing the strongest protection
-    against phishing attacks. Note: the Cloud Identity Policy API returns
-    domain-wide policies — it cannot verify enforcement for admin roles
-    specifically. This check evaluates the customer-level policy which
-    applies to all users including administrators.
+    CIS 4.1.1.2 asks for security keys to be the only accepted method, with
+    enrollment allowed, enforcement on or scheduled, and a policy suspension
+    grace period of at most one day, so the requirement cannot pass on the
+    accepted method alone.
+
+    Note: the Cloud Identity Policy API returns domain-wide policies, it cannot
+    verify enforcement for admin roles specifically. This check evaluates the
+    customer-level policy, which applies to all users including administrators,
+    so a passing domain-wide policy also covers the administrative accounts.
     """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
@@ -29,35 +59,119 @@ class security_2sv_hardware_keys_admins(Check):
                 customer_id=security_client.provider.identity.customer_id,
             )
 
-            factor_set = security_client.policies.two_sv_allowed_factor_set
+            policies = security_client.policies
+            domain = security_client.provider.identity.domain
 
-            if factor_set == "PASSKEY_ONLY":
-                report.status = "PASS"
+            unevaluable = unevaluable_reason(policies, EVALUATED_SETTINGS)
+            if unevaluable:
+                report.status = "MANUAL"
                 report.status_extended = (
-                    f"2-Step Verification enforcement requires security keys only "
-                    f"in domain {security_client.provider.identity.domain}."
+                    f"2-Step Verification could not be evaluated in domain "
+                    f"{domain}: {unevaluable}. Review it in the Admin console."
+                )
+                findings.append(report)
+                return findings
+
+            caveat = override_caveat(policies, EVALUATED_SETTINGS)
+            issues = []  # (setting, why it fails)
+
+            factor_set = policies.two_sv_allowed_factor_set
+            if factor_set != SECURITY_KEYS_ONLY:
+                issues.append(
+                    (
+                        "security.two_step_verification_enforcement_factor",
+                        (
+                            "the accepted method is not configured and defaults to "
+                            "any method, including SMS and phone call"
+                            if factor_set is None
+                            else f"the accepted method is {factor_set} "
+                            f"(should be {SECURITY_KEYS_ONLY})"
+                        ),
+                    )
+                )
+
+            # 4.1.1.2 accepts "On from <date>", unlike 4.1.1.1 and 4.1.1.3.
+            enforcement = enforcement_issue(
+                policies.two_sv_enforced_from, allow_scheduled=True
+            )
+            if enforcement:
+                issues.append(
+                    ("security.two_step_verification_enforcement", enforcement)
+                )
+
+            if policies.two_sv_allow_enrollment is False:
+                issues.append(
+                    (
+                        "security.two_step_verification_enrollment",
+                        "users are not allowed to turn on 2-Step Verification",
+                    )
+                )
+
+            # Google's default is no suspension grace period, which is stricter
+            # than the one day the benchmark asks for, so only longer periods
+            # fail. A value Prowler cannot read fails closed.
+            raw_grace_period = policies.two_sv_backup_code_exception_period
+            grace_period = parse_duration_seconds(raw_grace_period)
+            if raw_grace_period and grace_period is None:
+                issues.append(
+                    (
+                        "security.two_step_verification_sign_in_code",
+                        f"the 2-Step Verification policy suspension grace period "
+                        f"'{raw_grace_period}' could not be read",
+                    )
+                )
+            elif grace_period is not None and grace_period > ONE_DAY_SECONDS:
+                issues.append(
+                    (
+                        "security.two_step_verification_sign_in_code",
+                        f"the 2-Step Verification policy suspension grace period "
+                        f"is {format_duration(raw_grace_period)} "
+                        f"(should not exceed 1 day)",
+                    )
+                )
+
+            failing_settings = frozenset(setting for setting, _ in issues)
+            reasons = "; ".join(text for _, text in issues)
+
+            if issues and failures_shadowed_by_overrides(policies, failing_settings):
+                # The admin group of 4.1.1.2 may get the overriding value,
+                # which the Policy API does not expose, so the domain-wide
+                # failure cannot be confirmed for it.
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"2-Step Verification does not require security keys in the "
+                    f"domain-wide policy of {domain}: {reasons}. However, every "
+                    f"failing setting is also overridden for at least one group "
+                    f"or organizational unit, so the administrative accounts may "
+                    f"be configured correctly. Review those overrides in the "
+                    f"Admin console."
+                )
+            elif issues:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"2-Step Verification does not require security keys as "
+                    f"configured in domain {domain}: {reasons}. "
+                    + (f"Note: {caveat}. " if caveat else "")
+                    + "Note: this check evaluates the domain-wide policy, the "
+                    "Policy API does not expose role-specific 2SV enforcement."
+                )
+            elif caveat:
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"2-Step Verification meets the benchmark in the domain-wide "
+                    f"policy of {domain}, but {caveat}. Review those overrides "
+                    f"in the Admin console."
                 )
             else:
-                report.status = "FAIL"
-                if factor_set is None:
-                    report.status_extended = (
-                        f"2-Step Verification enforcement factor is not configured "
-                        f"in domain {security_client.provider.identity.domain}. "
-                        f"The default allows all methods including SMS and phone call. "
-                        f"Security keys should be required for administrative accounts. "
-                        f"Note: this check evaluates the domain-wide policy, the Policy "
-                        f"API does not expose role-specific 2SV enforcement."
-                    )
-                else:
-                    report.status_extended = (
-                        f"2-Step Verification enforcement factor is set to "
-                        f"{factor_set} "
-                        f"in domain {security_client.provider.identity.domain}. "
-                        f"Only security keys (PASSKEY_ONLY) should be allowed for "
-                        f"administrative accounts. "
-                        f"Note: this check evaluates the domain-wide policy, the Policy "
-                        f"API does not expose role-specific 2SV enforcement."
-                    )
+                report.status = "PASS"
+                report.status_extended = (
+                    f"2-Step Verification requires security keys only in domain "
+                    f"{domain}, enforcement is on or scheduled and the policy "
+                    f"suspension grace period is "
+                    f"{format_duration(policies.two_sv_backup_code_exception_period)}. "
+                    f"Note: this check evaluates the domain-wide policy, the "
+                    f"Policy API does not expose role-specific 2SV enforcement."
+                )
 
             findings.append(report)
 

prowler/providers/googleworkspace/services/security/security_login_challenges_configured/security_login_challenges_configured.metadata.json

@@ -9,7 +9,7 @@
   "Severity": "medium",
   "ResourceType": "NotDefined",
   "ResourceGroup": "IAM",
-  "Description": "The domain-level login challenges configuration has the **employee ID challenge disabled**. CIS 4.1.4.1 also requires Post-SSO verification to be enabled, but that setting is **not exposed by the Cloud Identity Policy API**. This check only covers the employee ID challenge portion of the control.",
+  "Description": "The domain-level login challenges configuration has the **employee ID challenge disabled**. This check is **not mapped to CIS 4.1.4.1** because that recommendation also requires Post-SSO verification, a setting **not exposed by the Cloud Identity Policy API**, so the requirement cannot be evaluated end to end.",
   "Risk": "When the employee ID login challenge is enabled without proper configuration, it may create a **false sense of security** or interfere with the login flow. The employee ID challenge is a supplementary verification method that should only be used when specifically required by the organization.",
   "RelatedUrl": "",
   "AdditionalURLs": [

prowler/providers/googleworkspace/services/security/security_login_challenges_configured/security_login_challenges_configured.py

@@ -9,11 +9,11 @@ from prowler.providers.googleworkspace.services.security.security_client import
 class security_login_challenges_configured(Check):
     """Check that login challenges are configured correctly.
 
-    This check verifies that the employee ID login challenge is disabled,
-    as recommended by CIS. Note: CIS 4.1.4.1 also requires Post-SSO
-    verification to be enabled, but that setting is not exposed by the
-    Cloud Identity Policy API. This check only covers the employee ID
-    challenge portion of the control.
+    This check verifies that the employee ID login challenge is disabled.
+
+    It is not mapped to CIS 4.1.4.1: that recommendation also requires Post-SSO
+    verification, a setting the Cloud Identity Policy API does not expose, so
+    the requirement cannot be evaluated end to end and is reported as manual.
     """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:

prowler/providers/googleworkspace/services/security/security_password_policy_strong/security_password_policy_strong.py

@@ -1,6 +1,11 @@
 from typing import List
 
 from prowler.lib.check.models import Check, CheckReportGoogleWorkspace
+from prowler.providers.googleworkspace.services.security.lib.durations import (
+    ONE_YEAR_SECONDS,
+    format_duration,
+    parse_duration_seconds,
+)
 from prowler.providers.googleworkspace.services.security.security_client import (
     security_client,
 )
@@ -11,8 +16,9 @@ class security_password_policy_strong(Check):
 
     This check verifies that the domain-level password policy meets CIS
     requirements: minimum length of 14 characters, strong passwords enforced,
-    password reuse disallowed, enforcement at next sign-in, and password
-    expiration configured.
+    password reuse disallowed, enforcement at next sign-in, and a password
+    reset frequency of 365 days or less. Shorter periods are more restrictive
+    than the benchmark asks for, so only longer ones fail.
     """
 
     def execute(self) -> List[CheckReportGoogleWorkspace]:
@@ -39,12 +45,11 @@ class security_password_policy_strong(Check):
                     else f"minimum length is {min_length} (requires 14+)"
                 )
 
-            if policies.password_allowed_strength != "STRONG":
-                issues.append(
-                    "password strength is not configured (requires STRONG)"
-                    if policies.password_allowed_strength is None
-                    else f"password strength is {policies.password_allowed_strength} (requires STRONG)"
-                )
+            # Google enforces strong passwords by default, so an unset value is
+            # the secure default rather than a missing configuration.
+            strength = policies.password_allowed_strength
+            if strength is not None and strength != "STRONG":
+                issues.append(f"password strength is {strength} (requires STRONG)")
 
             if policies.password_allow_reuse is True:
                 issues.append("password reuse is allowed")
@@ -52,9 +57,22 @@ class security_password_policy_strong(Check):
             if policies.password_enforce_at_login is not True:
                 issues.append("password policy is not enforced at next sign-in")
 
-            expiration = policies.password_expiration_duration
-            if expiration is None or expiration == "0s":
+            raw_expiration = policies.password_expiration_duration
+            expiration = parse_duration_seconds(raw_expiration)
+            if raw_expiration and expiration is None:
+                issues.append(
+                    f"password expiration '{raw_expiration}' could not be read"
+                )
+            elif expiration is None:
                 issues.append("password expiration is not configured")
+            elif expiration == 0:
+                issues.append("passwords are set to never expire")
+            elif expiration > ONE_YEAR_SECONDS:
+                issues.append(
+                    f"password expiration is "
+                    f"{format_duration(policies.password_expiration_duration)} "
+                    f"(requires 365 days or less)"
+                )
 
             if not issues:
                 report.status = "PASS"
@@ -62,7 +80,8 @@ class security_password_policy_strong(Check):
                     f"Password policy meets CIS requirements "
                     f"in domain {domain}: minimum length {min_length}, "
                     f"strong passwords enforced, reuse disallowed, "
-                    f"enforced at next sign-in, expiration configured."
+                    f"enforced at next sign-in, expiration "
+                    f"{format_duration(policies.password_expiration_duration)}."
                 )
             else:
                 report.status = "FAIL"

prowler/providers/googleworkspace/services/security/security_service.py

@@ -1,9 +1,13 @@
-from typing import Optional
+from typing import List, Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from prowler.lib.logger import logger
-from prowler.providers.googleworkspace.lib.service.service import GoogleWorkspaceService
+from prowler.providers.googleworkspace.lib.service.service import (
+    CUSTOMER_SCOPE,
+    UNKNOWN_SCOPE,
+    GoogleWorkspaceService,
+)
 
 
 class Security(GoogleWorkspaceService):
@@ -18,6 +22,8 @@ class Security(GoogleWorkspaceService):
         super().__init__(provider)
         self.policies = SecurityPolicies()
         self.policies_fetched = False
+        self._overridden = set()
+        self._observed = set()
         self._fetch_security_policies()
 
     def _fetch_security_policies(self):
@@ -48,6 +54,10 @@ class Security(GoogleWorkspaceService):
                 service, 'setting.type.matches("rule.dlp")', fetch_succeeded
             )
 
+            self.policies.overridden_settings = sorted(self._overridden)
+            self.policies.unobserved_settings = sorted(
+                self._overridden - self._observed
+            )
             self.policies_fetched = fetch_succeeded
 
             if fetch_succeeded:
@@ -79,11 +89,18 @@ class Security(GoogleWorkspaceService):
                     response = request.execute()
 
                     for policy in response.get("policies", []):
-                        if not self._is_customer_level_policy(policy):
-                            continue
-
                         setting = policy.get("setting", {})
                         setting_type = setting.get("type", "").removeprefix("settings/")
+
+                        scope = self._policy_scope(policy)
+                        if scope != CUSTOMER_SCOPE:
+                            if scope == UNKNOWN_SCOPE:
+                                self.policies.unresolved_scope = True
+                            elif setting_type:
+                                self._overridden.add(setting_type)
+                            continue
+
+                        self._observed.add(setting_type)
                         value = setting.get("value", {})
 
                         self._process_setting(setting_type, value)
@@ -239,6 +256,17 @@ class Security(GoogleWorkspaceService):
 class SecurityPolicies(BaseModel):
     """Model for domain-level Security policy settings."""
 
+    # Setting types that a group or a sub-OU overrides. Sorted lists rather than
+    # sets: a set reaches the OCSF output as its Python repr, in a different
+    # order on every scan.
+    overridden_settings: List[str] = Field(default_factory=list)
+    # Overridden settings with no domain-wide policy of their own, so the values
+    # below are Prowler's defaults and not something the domain reported.
+    unobserved_settings: List[str] = Field(default_factory=list)
+    # True when a policy's scope could not be determined because the root
+    # organizational unit id is unknown, which blanks the values below.
+    unresolved_scope: bool = False
+
     # security.two_step_verification_enrollment
     two_sv_allow_enrollment: Optional[bool] = None
     # security.two_step_verification_enforcement

tests/providers/aws/aws_provider_test.py

@@ -1783,6 +1783,29 @@ aws:
         assert sts_session._endpoint._endpoint_prefix == "sts"
         assert sts_session._endpoint.host == f"https://sts.{aws_region}.amazonaws.eu"
 
+    @mock_aws
+    def test_create_sts_session_empty_endpoint_url(self):
+        current_session = session.Session()
+        aws_region = AWS_REGION_US_EAST_1
+        with mock.patch.dict(os.environ, {"AWS_ENDPOINT_URL": ""}):
+            sts_session = AwsProvider.create_sts_session(current_session, aws_region)
+
+        assert sts_session._service_model.service_name == "sts"
+        assert sts_session._client_config.region_name == aws_region
+        assert sts_session._endpoint._endpoint_prefix == "sts"
+        assert sts_session._endpoint.host == f"https://sts.{aws_region}.amazonaws.com"
+
+    @mock_aws
+    def test_create_sts_session_iso(self):
+        current_session = session.Session()
+        aws_region = "us-iso-east-1"
+        sts_session = AwsProvider.create_sts_session(current_session, aws_region)
+
+        assert sts_session._service_model.service_name == "sts"
+        assert sts_session._client_config.region_name == aws_region
+        assert sts_session._endpoint._endpoint_prefix == "sts"
+        assert sts_session._endpoint.host == f"https://sts.{aws_region}.c2s.ic.gov"
+
     @mock_aws
     @patch(
         "prowler.lib.check.utils.recover_checks_from_provider",
@@ -2219,6 +2242,239 @@ aws:
             == AWS_REGION_US_EAST_1
         )
 
+    def test_get_aws_region_for_sts_env_partition_gov_cloud(self):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+            clear=False,
+        ):
+            assert get_aws_region_for_sts(None, None) == AWS_REGION_GOV_CLOUD_US_EAST_1
+
+    def test_get_aws_region_for_sts_env_partition_china(self):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_CHINA_PARTITION},
+            clear=False,
+        ):
+            assert get_aws_region_for_sts(None, None) == AWS_REGION_CN_NORTH_1
+
+    def test_get_aws_region_for_sts_env_partition_eusc(self):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_EUSC_PARTITION},
+            clear=False,
+        ):
+            assert get_aws_region_for_sts(None, None) == AWS_REGION_EUSC_DE_EAST_1
+
+    def test_get_aws_region_for_sts_env_partition_iso(self):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_ISO_PARTITION},
+            clear=False,
+        ):
+            assert get_aws_region_for_sts(None, None) == "us-iso-east-1"
+
+    def test_get_aws_region_for_sts_env_partition_overrides_session_region(self):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+            clear=False,
+        ):
+            assert (
+                get_aws_region_for_sts(AWS_REGION_EU_WEST_1, None)
+                == AWS_REGION_GOV_CLOUD_US_EAST_1
+            )
+
+    def test_get_aws_region_for_sts_input_regions_take_precedence_over_env_partition(
+        self,
+    ):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+            clear=False,
+        ):
+            assert (
+                get_aws_region_for_sts(None, {AWS_REGION_EU_WEST_1})
+                == AWS_REGION_EU_WEST_1
+            )
+
+    def test_get_aws_region_for_sts_env_partition_invalid_raises(self):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": "aws-invalid"},
+            clear=False,
+        ):
+            with pytest.raises(AWSInvalidPartitionError):
+                get_aws_region_for_sts(None, None)
+
+    @mock_aws
+    def test_test_connection_uses_env_partition_sts_region(self):
+        with (
+            mock.patch.dict(
+                os.environ,
+                {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+                clear=False,
+            ),
+            mock.patch.object(
+                AwsProvider,
+                "validate_credentials",
+                return_value=AWSCallerIdentity(
+                    user_id="test-user-id",
+                    account=AWS_ACCOUNT_NUMBER,
+                    arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN),
+                    region=AWS_REGION_GOV_CLOUD_US_EAST_1,
+                ),
+            ) as mock_validate_credentials,
+        ):
+            connection = AwsProvider.test_connection(
+                aws_access_key_id="test-access-key",
+                aws_secret_access_key="test-secret-key",
+                raise_on_exception=False,
+            )
+
+            assert connection.is_connected
+            assert (
+                mock_validate_credentials.call_args.args[1]
+                == AWS_REGION_GOV_CLOUD_US_EAST_1
+            )
+
+    @mock_aws
+    def test_test_connection_role_uses_env_partition_sts_region(self):
+        with (
+            mock.patch.dict(
+                os.environ,
+                {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+                clear=False,
+            ),
+            mock.patch.object(
+                AwsProvider,
+                "assume_role",
+                return_value=AWSCredentials(
+                    aws_access_key_id="assumed-access-key",
+                    aws_secret_access_key="assumed-secret-key",
+                    aws_session_token="assumed-session-token",
+                    expiration=datetime.now(),
+                ),
+            ) as mock_assume_role,
+            mock.patch.object(
+                AwsProvider,
+                "validate_credentials",
+                return_value=AWSCallerIdentity(
+                    user_id="test-user-id",
+                    account=AWS_ACCOUNT_NUMBER,
+                    arn=ARN(AWS_GOV_CLOUD_ACCOUNT_ARN),
+                    region=AWS_REGION_GOV_CLOUD_US_EAST_1,
+                ),
+            ),
+        ):
+            connection = AwsProvider.test_connection(
+                role_arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:role/test-role",
+                aws_access_key_id="test-access-key",
+                aws_secret_access_key="test-secret-key",
+                raise_on_exception=False,
+            )
+
+            assert connection.is_connected
+            assumed_role_info = mock_assume_role.call_args.args[1]
+            assert assumed_role_info.sts_region == AWS_REGION_GOV_CLOUD_US_EAST_1
+
+    def test_get_aws_region_for_sts_env_partition_commercial(self):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_COMMERCIAL_PARTITION},
+            clear=False,
+        ):
+            assert get_aws_region_for_sts(None, None) == AWS_REGION_US_EAST_1
+
+    def test_get_aws_region_for_sts_env_partition_excluded_region_stays_in_partition(
+        self,
+    ):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+            clear=False,
+        ):
+            assert (
+                get_aws_region_for_sts(None, None, {AWS_REGION_GOV_CLOUD_US_EAST_1})
+                == "us-gov-west-1"
+            )
+
+    def test_get_aws_region_for_sts_env_partition_all_regions_excluded_stays_in_partition(
+        self,
+    ):
+        with mock.patch.dict(
+            os.environ,
+            {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+            clear=False,
+        ):
+            assert (
+                get_aws_region_for_sts(
+                    None, None, {AWS_REGION_GOV_CLOUD_US_EAST_1, "us-gov-west-1"}
+                )
+                == AWS_REGION_GOV_CLOUD_US_EAST_1
+            )
+
+    @mock_aws
+    def test_setup_session_mfa_uses_env_partition_sts_region(self):
+        with (
+            mock.patch.dict(
+                os.environ,
+                {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+                clear=False,
+            ),
+            mock.patch.object(
+                AwsProvider,
+                "input_role_mfa_token_and_code",
+                return_value=AWSMFAInfo(
+                    arn=f"arn:{AWS_GOV_CLOUD_PARTITION}:iam::{AWS_ACCOUNT_NUMBER}:mfa/test",
+                    totp="123456",
+                ),
+            ),
+            mock.patch.object(
+                AwsProvider,
+                "create_sts_session",
+                side_effect=AwsProvider.create_sts_session,
+            ) as mock_create_sts_session,
+        ):
+            AwsProvider.setup_session(
+                mfa=True,
+                aws_access_key_id="test-access-key",
+                aws_secret_access_key="test-secret-key",
+            )
+
+            assert (
+                mock_create_sts_session.call_args.args[1]
+                == AWS_REGION_GOV_CLOUD_US_EAST_1
+            )
+
+    @mock_aws
+    def test_test_connection_env_partition_mismatch(self):
+        with (
+            mock.patch.dict(
+                os.environ,
+                {"PROWLER_AWS_PARTITION": AWS_GOV_CLOUD_PARTITION},
+                clear=False,
+            ),
+            mock.patch.object(
+                AwsProvider,
+                "validate_credentials",
+                return_value=AWSCallerIdentity(
+                    user_id="test-user-id",
+                    account=AWS_ACCOUNT_NUMBER,
+                    arn=ARN(AWS_ACCOUNT_ARN),
+                    region=AWS_REGION_US_EAST_1,
+                ),
+            ),
+        ):
+            connection = AwsProvider.test_connection(
+                aws_access_key_id="test-access-key",
+                aws_secret_access_key="test-secret-key",
+                raise_on_exception=False,
+            )
+
+            assert not connection.is_connected
+            assert isinstance(connection.error, AWSInvalidPartitionError)
+
     def test_get_profile_region_avoids_excluded_session_region(self):
         mocked_session = mock.Mock(region_name=AWS_REGION_EU_WEST_1)
 

tests/providers/cloudflare/cloudflare_provider_test.py

@@ -9,7 +9,6 @@ from prowler.providers.cloudflare.exceptions.exceptions import (
     CloudflareInvalidAPIKeyError,
     CloudflareInvalidAPITokenError,
     CloudflareNoAccountsError,
-    CloudflareUserTokenRequiredError,
 )
 from prowler.providers.cloudflare.models import (
     CloudflareAccount,
@@ -308,8 +307,8 @@ class TestCloudflareValidateCredentials:
         CloudflareProvider.validate_credentials(session)
         mock_client.user.get.assert_called_once()
 
-    def test_validate_credentials_user_token_required(self):
-        """Test that user token required error is raised for Account tokens."""
+    def test_validate_credentials_account_token_without_accounts(self):
+        """Test that account tokens fall back to account discovery."""
         mock_client = MagicMock()
         # Simulate error code 9109 - user-level authentication required
         from cloudflare._exceptions import PermissionDeniedError
@@ -319,6 +318,7 @@ class TestCloudflareValidateCredentials:
             response=MagicMock(status_code=403),
             body=None,
         )
+        mock_client.accounts.list.return_value = iter([])
 
         session = CloudflareSession(
             client=mock_client,
@@ -327,8 +327,10 @@ class TestCloudflareValidateCredentials:
             api_email=None,
         )
 
-        with pytest.raises(CloudflareUserTokenRequiredError):
+        with pytest.raises(CloudflareNoAccountsError):
             CloudflareProvider.validate_credentials(session)
+        mock_client.user.get.assert_called_once()
+        mock_client.accounts.list.assert_called_once()
 
     def test_validate_credentials_invalid_api_token(self):
         """Test that invalid API token error is raised."""
@@ -487,8 +489,8 @@ class TestCloudflareTestConnection:
             assert connection.is_connected is False
             assert isinstance(connection.error, CloudflareInvalidAPITokenError)
 
-    def test_test_connection_user_token_required(self):
-        """Test that user token required error is properly returned."""
+    def test_test_connection_account_token_without_accounts(self):
+        """Test that account tokens without accessible accounts return an error."""
         mock_client = MagicMock()
         from cloudflare._exceptions import PermissionDeniedError
 
@@ -497,6 +499,7 @@ class TestCloudflareTestConnection:
             response=MagicMock(status_code=403),
             body=None,
         )
+        mock_client.accounts.list.return_value = iter([])
 
         with patch(
             "prowler.providers.cloudflare.cloudflare_provider.CloudflareProvider.setup_session",
@@ -512,9 +515,10 @@ class TestCloudflareTestConnection:
             )
 
             assert connection.is_connected is False
-            assert isinstance(connection.error, CloudflareUserTokenRequiredError)
-            # Verify the error message is user-friendly
-            assert "User-level API token required" in str(connection.error)
+            assert isinstance(connection.error, CloudflareNoAccountsError)
+            assert "No Cloudflare accounts found" in str(connection.error)
+            mock_client.user.get.assert_called_once()
+            mock_client.accounts.list.assert_called_once()
 
     def test_test_connection_invalid_api_key(self):
         """Test that invalid API key error is properly returned."""

tests/providers/cloudflare/services/zone/zone_challenge_passage_configured/zone_challenge_passage_configured_test.py

@@ -240,3 +240,7 @@ class Test_zone_challenge_passage_configured:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Challenge Passage is not configured for zone {ZONE_NAME}."
+            )

tests/providers/cloudflare/services/zone/zone_development_mode_disabled/zone_development_mode_disabled_test.py

@@ -102,8 +102,10 @@ class Test_zone_development_mode_disabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert "Development mode is enabled" in result[0].status_extended
-            assert "bypasses" in result[0].status_extended
+            assert (
+                result[0].status_extended
+                == f"Development mode is enabled for zone {ZONE_NAME}."
+            )
 
     def test_zone_development_mode_none(self):
         zone_client = mock.MagicMock

tests/providers/googleworkspace/googleworkspace_fixtures.py

@@ -1,5 +1,6 @@
 """Test fixtures for Google Workspace provider tests"""
 
+from typing import Optional
 from unittest.mock import MagicMock
 
 from prowler.providers.googleworkspace.models import (
@@ -81,17 +82,19 @@ ROLE_GROUPS_ADMIN = {
 
 
 def set_mocked_googleworkspace_provider(
-    identity: GoogleWorkspaceIdentityInfo = GoogleWorkspaceIdentityInfo(
+    identity: Optional[GoogleWorkspaceIdentityInfo] = None,
+):
+    provider = MagicMock()
+    provider.type = "googleworkspace"
+    # Built per call: as a default argument every test would share one instance,
+    # and a test mutating it would leak into the rest of the session.
+    provider.identity = identity or GoogleWorkspaceIdentityInfo(
         domain=DOMAIN,
         customer_id=CUSTOMER_ID,
         delegated_user=DELEGATED_USER,
         root_org_unit_id=ROOT_ORG_UNIT_ID,
         profile="default",
-    ),
-):
-    provider = MagicMock()
-    provider.type = "googleworkspace"
-    provider.identity = identity
+    )
     provider.domain_resource = build_googleworkspace_domain_resource()
     return provider
 

tests/providers/googleworkspace/lib/service/googleworkspace_service_test.py

@@ -1,6 +1,13 @@
 from unittest.mock import MagicMock
 
-from prowler.providers.googleworkspace.lib.service.service import GoogleWorkspaceService
+import pytest
+
+from prowler.providers.googleworkspace.lib.service.service import (
+    CUSTOMER_SCOPE,
+    OVERRIDE_SCOPE,
+    UNKNOWN_SCOPE,
+    GoogleWorkspaceService,
+)
 
 ROOT_OU_ID = "03ph8a2z1234"
 
@@ -80,3 +87,41 @@ class TestIsCustomerLevelPolicy:
             )
             is False
         )
+
+
+class TestPolicyScope:
+    @pytest.mark.parametrize(
+        "policy, expected",
+        [
+            ({}, CUSTOMER_SCOPE),
+            ({"policyQuery": {}}, CUSTOMER_SCOPE),
+            ({"policyQuery": None}, CUSTOMER_SCOPE),
+            ({"policyQuery": {"orgUnit": ""}}, CUSTOMER_SCOPE),
+            ({"policyQuery": {"orgUnit": f"orgUnits/{ROOT_OU_ID}"}}, CUSTOMER_SCOPE),
+            ({"policyQuery": {"orgUnit": "orgUnits/sub_ou"}}, OVERRIDE_SCOPE),
+            ({"policyQuery": {"group": "groups/xyz"}}, OVERRIDE_SCOPE),
+            (
+                {"policyQuery": {"group": "groups/xyz", "orgUnit": "orgUnits/sub_ou"}},
+                OVERRIDE_SCOPE,
+            ),
+        ],
+    )
+    def test_scope_with_a_known_root_org_unit(self, policy, expected):
+        assert _make_service()._policy_scope(policy) == expected
+
+    @pytest.mark.parametrize("org_unit", [f"orgUnits/{ROOT_OU_ID}", "orgUnits/sub_ou"])
+    def test_without_the_root_id_an_org_unit_scope_is_unknown(self, org_unit):
+        """The root OU and a sub-OU are indistinguishable, so neither may be assumed"""
+        svc = _make_service(root_org_unit_id=None)
+
+        assert (
+            svc._policy_scope({"policyQuery": {"orgUnit": org_unit}}) == UNKNOWN_SCOPE
+        )
+
+    def test_a_group_is_an_override_even_without_the_root_id(self):
+        svc = _make_service(root_org_unit_id=None)
+
+        assert (
+            svc._policy_scope({"policyQuery": {"group": "groups/xyz"}})
+            == OVERRIDE_SCOPE
+        )

tests/providers/googleworkspace/services/gmail/gmail_domain_spoofing_protection_enabled/gmail_domain_spoofing_protection_enabled_test.py

@@ -40,6 +40,38 @@ class TestGmailDomainSpoofingProtectionEnabled:
             assert findings[0].resource_name == "Gmail Policies"
             assert findings[0].customer_id == CUSTOMER_ID
 
+    def test_pass_enable_flag_not_returned(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.gmail.gmail_domain_spoofing_protection_enabled.gmail_domain_spoofing_protection_enabled.gmail_client"
+            ) as mock_client,
+        ):
+            from prowler.providers.googleworkspace.services.gmail.gmail_domain_spoofing_protection_enabled.gmail_domain_spoofing_protection_enabled import (
+                gmail_domain_spoofing_protection_enabled,
+            )
+
+            mock_client.provider = mock_provider
+            mock_client.policies_fetched = True
+            mock_client.policies = GmailPolicies(
+                domain_spoofing_consequence="SPAM_FOLDER",
+            )
+
+            check = gmail_domain_spoofing_protection_enabled()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "PASS"
+            assert "uses Google's default (enabled)" in findings[0].status_extended
+            assert "is enabled with action" not in findings[0].status_extended
+            assert findings[0].resource_name == "Gmail Policies"
+            assert findings[0].customer_id == CUSTOMER_ID
+
     def test_fail_no_action(self):
         mock_provider = set_mocked_googleworkspace_provider()
 
@@ -70,6 +102,36 @@ class TestGmailDomainSpoofingProtectionEnabled:
             assert findings[0].status == "FAIL"
             assert "no action" in findings[0].status_extended
 
+    def test_fail_warning_action(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.gmail.gmail_domain_spoofing_protection_enabled.gmail_domain_spoofing_protection_enabled.gmail_client"
+            ) as mock_client,
+        ):
+            from prowler.providers.googleworkspace.services.gmail.gmail_domain_spoofing_protection_enabled.gmail_domain_spoofing_protection_enabled import (
+                gmail_domain_spoofing_protection_enabled,
+            )
+
+            mock_client.provider = mock_provider
+            mock_client.policies_fetched = True
+            mock_client.policies = GmailPolicies(
+                detect_domain_name_spoofing=True,
+                domain_spoofing_consequence="WARNING",
+            )
+
+            check = gmail_domain_spoofing_protection_enabled()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "show a warning" in findings[0].status_extended
+
     def test_fail_protection_disabled(self):
         mock_provider = set_mocked_googleworkspace_provider()
 
@@ -100,7 +162,7 @@ class TestGmailDomainSpoofingProtectionEnabled:
             assert findings[0].status == "FAIL"
             assert "disabled" in findings[0].status_extended
 
-    def test_pass_using_default(self):
+    def test_fail_using_default(self):
         mock_provider = set_mocked_googleworkspace_provider()
 
         with (
@@ -124,8 +186,8 @@ class TestGmailDomainSpoofingProtectionEnabled:
             findings = check.execute()
 
             assert len(findings) == 1
-            assert findings[0].status == "PASS"
-            assert "secure default" in findings[0].status_extended
+            assert findings[0].status == "FAIL"
+            assert "default action" in findings[0].status_extended
 
     def test_no_findings_when_fetch_failed(self):
         mock_provider = set_mocked_googleworkspace_provider()

tests/providers/googleworkspace/services/gmail/gmail_employee_name_spoofing_protection_enabled/gmail_employee_name_spoofing_protection_enabled_test.py

@@ -70,6 +70,36 @@ class TestGmailEmployeeNameSpoofingProtectionEnabled:
             assert findings[0].status == "FAIL"
             assert "no action" in findings[0].status_extended
 
+    def test_fail_warning_action(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.gmail.gmail_employee_name_spoofing_protection_enabled.gmail_employee_name_spoofing_protection_enabled.gmail_client"
+            ) as mock_client,
+        ):
+            from prowler.providers.googleworkspace.services.gmail.gmail_employee_name_spoofing_protection_enabled.gmail_employee_name_spoofing_protection_enabled import (
+                gmail_employee_name_spoofing_protection_enabled,
+            )
+
+            mock_client.provider = mock_provider
+            mock_client.policies_fetched = True
+            mock_client.policies = GmailPolicies(
+                detect_employee_name_spoofing=True,
+                employee_name_spoofing_consequence="WARNING",
+            )
+
+            check = gmail_employee_name_spoofing_protection_enabled()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "show a warning" in findings[0].status_extended
+
     def test_fail_protection_disabled(self):
         mock_provider = set_mocked_googleworkspace_provider()
 
@@ -100,7 +130,7 @@ class TestGmailEmployeeNameSpoofingProtectionEnabled:
             assert findings[0].status == "FAIL"
             assert "disabled" in findings[0].status_extended
 
-    def test_pass_using_default(self):
+    def test_fail_using_default(self):
         mock_provider = set_mocked_googleworkspace_provider()
 
         with (
@@ -124,8 +154,8 @@ class TestGmailEmployeeNameSpoofingProtectionEnabled:
             findings = check.execute()
 
             assert len(findings) == 1
-            assert findings[0].status == "PASS"
-            assert "secure default" in findings[0].status_extended
+            assert findings[0].status == "FAIL"
+            assert "default action" in findings[0].status_extended
 
     def test_no_findings_when_fetch_failed(self):
         mock_provider = set_mocked_googleworkspace_provider()

tests/providers/googleworkspace/services/gmail/gmail_groups_spoofing_protection_enabled/gmail_groups_spoofing_protection_enabled_test.py

@@ -103,6 +103,36 @@ class TestGmailGroupsSpoofingProtectionEnabled:
             assert findings[0].status == "FAIL"
             assert "no action" in findings[0].status_extended
 
+    def test_fail_warning_action(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.gmail.gmail_groups_spoofing_protection_enabled.gmail_groups_spoofing_protection_enabled.gmail_client"
+            ) as mock_client,
+        ):
+            from prowler.providers.googleworkspace.services.gmail.gmail_groups_spoofing_protection_enabled.gmail_groups_spoofing_protection_enabled import (
+                gmail_groups_spoofing_protection_enabled,
+            )
+
+            mock_client.provider = mock_provider
+            mock_client.policies_fetched = True
+            mock_client.policies = GmailPolicies(
+                detect_groups_spoofing=True,
+                groups_spoofing_consequence="WARNING",
+            )
+
+            check = gmail_groups_spoofing_protection_enabled()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "show a warning" in findings[0].status_extended
+
     def test_fail_protection_disabled(self):
         mock_provider = set_mocked_googleworkspace_provider()
 

tests/providers/googleworkspace/services/gmail/gmail_inbound_domain_spoofing_protection_enabled/gmail_inbound_domain_spoofing_protection_enabled_test.py

@@ -70,6 +70,36 @@ class TestGmailInboundDomainSpoofingProtectionEnabled:
             assert findings[0].status == "FAIL"
             assert "no action" in findings[0].status_extended
 
+    def test_fail_warning_action(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.gmail.gmail_inbound_domain_spoofing_protection_enabled.gmail_inbound_domain_spoofing_protection_enabled.gmail_client"
+            ) as mock_client,
+        ):
+            from prowler.providers.googleworkspace.services.gmail.gmail_inbound_domain_spoofing_protection_enabled.gmail_inbound_domain_spoofing_protection_enabled import (
+                gmail_inbound_domain_spoofing_protection_enabled,
+            )
+
+            mock_client.provider = mock_provider
+            mock_client.policies_fetched = True
+            mock_client.policies = GmailPolicies(
+                detect_inbound_domain_spoofing=True,
+                inbound_domain_spoofing_consequence="WARNING",
+            )
+
+            check = gmail_inbound_domain_spoofing_protection_enabled()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "show a warning" in findings[0].status_extended
+
     def test_fail_protection_disabled(self):
         mock_provider = set_mocked_googleworkspace_provider()
 
@@ -100,7 +130,7 @@ class TestGmailInboundDomainSpoofingProtectionEnabled:
             assert findings[0].status == "FAIL"
             assert "disabled" in findings[0].status_extended
 
-    def test_pass_using_default(self):
+    def test_fail_using_default(self):
         mock_provider = set_mocked_googleworkspace_provider()
 
         with (
@@ -124,8 +154,8 @@ class TestGmailInboundDomainSpoofingProtectionEnabled:
             findings = check.execute()
 
             assert len(findings) == 1
-            assert findings[0].status == "PASS"
-            assert "secure default" in findings[0].status_extended
+            assert findings[0].status == "FAIL"
+            assert "default action" in findings[0].status_extended
 
     def test_no_findings_when_fetch_failed(self):
         mock_provider = set_mocked_googleworkspace_provider()

tests/providers/googleworkspace/services/gmail/lib/spoofing_test.py

@@ -0,0 +1,34 @@
+import pytest
+
+from prowler.providers.googleworkspace.services.gmail.lib.spoofing import (
+    PROTECTIVE_CONSEQUENCES,
+    describe_consequence,
+    is_protective,
+)
+
+
+class TestIsProtective:
+    @pytest.mark.parametrize("consequence", sorted(PROTECTIVE_CONSEQUENCES))
+    def test_actions_that_move_the_message_out_of_the_inbox(self, consequence):
+        assert is_protective(consequence) is True
+
+    @pytest.mark.parametrize(
+        "consequence", ["WARNING", "NO_ACTION", None, "", "spam_folder", "UNKNOWN"]
+    )
+    def test_everything_else_is_not_protective(self, consequence):
+        """Anything the benchmark does not accept, including unknown values"""
+        assert is_protective(consequence) is False
+
+
+class TestDescribeConsequence:
+    @pytest.mark.parametrize(
+        "consequence, expected",
+        [
+            (None, "uses Google's default action"),
+            ("NO_ACTION", "is set to take no action"),
+            ("WARNING", "show a warning"),
+            ("SOMETHING_NEW", "is set to 'SOMETHING_NEW'"),
+        ],
+    )
+    def test_renders_for_finding_messages(self, consequence, expected):
+        assert expected in describe_consequence(consequence)

tests/providers/googleworkspace/services/rules/lib/alerts_test.py

@@ -0,0 +1,154 @@
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from prowler.lib.check.models import CheckMetadata
+from prowler.providers.googleworkspace.services.rules import rules_service
+from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+    evaluate_system_defined_alert,
+)
+from prowler.providers.googleworkspace.services.rules.rules_service import (
+    SystemDefinedAlert,
+)
+from tests.providers.googleworkspace.googleworkspace_fixtures import (
+    set_mocked_googleworkspace_provider,
+)
+
+RULE_NAME = "Leaked password"
+OTHER_RULE = "Suspicious login"
+
+
+def make_client(alerts, policies_fetched=True):
+    client = MagicMock()
+    client.provider = set_mocked_googleworkspace_provider()
+    client.policies_fetched = policies_fetched
+    client.system_defined_alerts = alerts
+    return client
+
+
+def configured(**overrides):
+    values = dict(
+        display_name=RULE_NAME,
+        state="ACTIVE",
+        severity="MEDIUM",
+        email_notifications_enabled=True,
+        all_super_admins=True,
+    )
+    values.update(overrides)
+    return SystemDefinedAlert(**values)
+
+
+# Real metadata: CheckReportGoogleWorkspace validates it, a mock will not do.
+# Loaded from the file rather than from the check class, whose module import
+# builds the service client and needs a live provider.
+METADATA_FILE = (
+    Path(rules_service.__file__).parent
+    / "rules_leaked_password_alert_configured"
+    / "rules_leaked_password_alert_configured.metadata.json"
+)
+METADATA = CheckMetadata.parse_file(METADATA_FILE).json()
+
+
+def run(alerts, minimum_severity="MEDIUM", policies_fetched=True):
+    return evaluate_system_defined_alert(
+        make_client(alerts, policies_fetched), METADATA, RULE_NAME, minimum_severity
+    )
+
+
+class TestEvaluateSystemDefinedAlert:
+    def test_evaluates_only_the_requested_rule(self):
+        findings = run(
+            [
+                configured(display_name=OTHER_RULE, state="INACTIVE", severity=None),
+                configured(),
+                configured(display_name="Government-backed attacks", severity="HIGH"),
+            ]
+        )
+
+        assert len(findings) == 1
+        assert findings[0].resource_name == RULE_NAME
+        assert findings[0].status == "PASS"
+
+    def test_no_finding_when_the_rule_is_absent(self):
+        assert run([configured(display_name=OTHER_RULE)]) == []
+
+    def test_no_finding_when_fetch_failed(self):
+        assert run([configured()], policies_fetched=False) == []
+
+    @pytest.mark.parametrize("severity", ["MEDIUM", "HIGH"])
+    def test_a_severity_above_the_minimum_is_stricter_not_weaker(self, severity):
+        findings = run([configured(severity=severity)], "MEDIUM")
+
+        assert findings[0].status == "PASS"
+
+    @pytest.mark.parametrize("severity", ["CRITICAL", "high", "SEVERITY_UNSPECIFIED"])
+    def test_an_unrankable_severity_is_not_claimed_to_be_below_the_minimum(
+        self, severity
+    ):
+        """Saying CRITICAL falls short of MEDIUM would be a lie, not a finding"""
+        findings = run([configured(severity=severity)], "MEDIUM")
+
+        assert findings[0].status == "FAIL"
+        assert f"severity is {severity}, which is not one of" in (
+            findings[0].status_extended
+        )
+        assert f"should be at least {severity}" not in findings[0].status_extended
+
+    def test_reports_the_minimum_severity_on_failure(self):
+        findings = run([configured(severity="LOW")], "MEDIUM")
+
+        assert findings[0].status == "FAIL"
+        assert "severity is LOW (should be at least MEDIUM)" in (
+            findings[0].status_extended
+        )
+
+    def test_an_unobserved_rule_left_on_an_active_default_is_manual(self):
+        """Google documents no default severity, so it cannot be verified"""
+        findings = run([configured(severity=None, from_default=True)])
+
+        assert findings[0].status == "MANUAL"
+        assert "was not returned by the API" in findings[0].status_extended
+
+    def test_an_unobserved_rule_that_defaults_to_off_still_fails(self):
+        """The OFF default is documented, so it fails whatever the severity is"""
+        findings = run([configured(state="INACTIVE", severity=None, from_default=True)])
+
+        assert findings[0].status == "FAIL"
+        assert "Google's default for it is OFF" in findings[0].status_extended
+
+    def test_fail_when_the_alert_is_not_sent_to_the_alert_center(self):
+        """An active rule whose alert center delivery is DISABLED is not compliant"""
+        findings = run([configured(alert_center_status="DISABLED")], "MEDIUM")
+
+        assert findings[0].status == "FAIL"
+        assert "not sent to the alert center" in findings[0].status_extended
+
+    def test_pass_when_the_alert_center_status_is_not_reported(self):
+        """The API never returns this field, so its absence cannot fail a rule"""
+        findings = run([configured(alert_center_status=None)], "MEDIUM")
+
+        assert findings[0].status == "PASS"
+
+    def test_a_failing_condition_wins_over_an_unreported_delivery(self):
+        findings = run([configured(severity="LOW", alert_center_status=None)], "MEDIUM")
+
+        assert findings[0].status == "FAIL"
+        assert "severity is LOW" in findings[0].status_extended
+
+    def test_reports_every_failing_condition(self):
+        findings = run(
+            [
+                configured(
+                    state="INACTIVE",
+                    severity="LOW",
+                    email_notifications_enabled=False,
+                )
+            ]
+        )
+
+        extended = findings[0].status_extended
+        assert findings[0].status == "FAIL"
+        assert "alert is OFF" in extended
+        assert "email notifications are disabled" in extended
+        assert "severity is LOW" in extended

tests/providers/googleworkspace/services/rules/rules_admin_privilege_granted_alert_configured/rules_admin_privilege_granted_alert_configured_test.py

@@ -46,7 +46,79 @@ class TestRulesAdminPrivilegeGrantedAlertConfigured:
 
             assert len(findings) == 1
             assert findings[0].status == "PASS"
-            assert "properly configured" in findings[0].status_extended
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_below_the_minimum(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_admin_privilege_granted_alert_configured.rules_admin_privilege_granted_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_admin_privilege_granted_alert_configured.rules_admin_privilege_granted_alert_configured import (
+                rules_admin_privilege_granted_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    severity="LOW",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_admin_privilege_granted_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is LOW" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_not_configured(self):
+        """Test FAIL when the alert is on but no severity is configured."""
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_admin_privilege_granted_alert_configured.rules_admin_privilege_granted_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_admin_privilege_granted_alert_configured.rules_admin_privilege_granted_alert_configured import (
+                rules_admin_privilege_granted_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_admin_privilege_granted_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is not configured" in findings[0].status_extended
             assert findings[0].customer_id == CUSTOMER_ID
 
     def test_fail_alert_off(self):

tests/providers/googleworkspace/services/rules/rules_gmail_employee_spoofing_alert_configured/rules_gmail_employee_spoofing_alert_configured_test.py

@@ -46,7 +46,79 @@ class TestRulesGmailEmployeeSpoofingAlertConfigured:
 
             assert len(findings) == 1
             assert findings[0].status == "PASS"
-            assert "properly configured" in findings[0].status_extended
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_below_the_minimum(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_gmail_employee_spoofing_alert_configured.rules_gmail_employee_spoofing_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_gmail_employee_spoofing_alert_configured.rules_gmail_employee_spoofing_alert_configured import (
+                rules_gmail_employee_spoofing_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    severity="LOW",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_gmail_employee_spoofing_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is LOW" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_not_configured(self):
+        """Test FAIL when the alert is on but no severity is configured."""
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_gmail_employee_spoofing_alert_configured.rules_gmail_employee_spoofing_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_gmail_employee_spoofing_alert_configured.rules_gmail_employee_spoofing_alert_configured import (
+                rules_gmail_employee_spoofing_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_gmail_employee_spoofing_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is not configured" in findings[0].status_extended
             assert findings[0].customer_id == CUSTOMER_ID
 
     def test_fail_alert_off(self):

tests/providers/googleworkspace/services/rules/rules_government_backed_attacks_alert_configured/rules_government_backed_attacks_alert_configured_test.py

@@ -35,7 +35,7 @@ class TestRulesGovernmentBackedAttacksAlertConfigured:
                 SystemDefinedAlert(
                     display_name=RULE_NAME,
                     state="ACTIVE",
-                    severity="MEDIUM",
+                    severity="HIGH",
                     email_notifications_enabled=True,
                     all_super_admins=True,
                 )
@@ -46,7 +46,79 @@ class TestRulesGovernmentBackedAttacksAlertConfigured:
 
             assert len(findings) == 1
             assert findings[0].status == "PASS"
-            assert "properly configured" in findings[0].status_extended
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_below_the_minimum(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_government_backed_attacks_alert_configured.rules_government_backed_attacks_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_government_backed_attacks_alert_configured.rules_government_backed_attacks_alert_configured import (
+                rules_government_backed_attacks_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    severity="MEDIUM",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_government_backed_attacks_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is MEDIUM" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_not_configured(self):
+        """Test FAIL when the alert is on but no severity is configured."""
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_government_backed_attacks_alert_configured.rules_government_backed_attacks_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_government_backed_attacks_alert_configured.rules_government_backed_attacks_alert_configured import (
+                rules_government_backed_attacks_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_government_backed_attacks_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is not configured" in findings[0].status_extended
             assert findings[0].customer_id == CUSTOMER_ID
 
     def test_fail_alert_off(self):
@@ -105,7 +177,7 @@ class TestRulesGovernmentBackedAttacksAlertConfigured:
                 SystemDefinedAlert(
                     display_name=RULE_NAME,
                     state="ACTIVE",
-                    severity="MEDIUM",
+                    severity="HIGH",
                     email_notifications_enabled=False,
                     all_super_admins=False,
                 )
@@ -141,7 +213,7 @@ class TestRulesGovernmentBackedAttacksAlertConfigured:
                 SystemDefinedAlert(
                     display_name=RULE_NAME,
                     state="ACTIVE",
-                    severity="MEDIUM",
+                    severity="HIGH",
                     email_notifications_enabled=True,
                     all_super_admins=False,
                 )

tests/providers/googleworkspace/services/rules/rules_leaked_password_alert_configured/rules_leaked_password_alert_configured_test.py

@@ -46,7 +46,116 @@ class TestRulesLeakedPasswordAlertConfigured:
 
             assert len(findings) == 1
             assert findings[0].status == "PASS"
-            assert "properly configured" in findings[0].status_extended
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_pass_high_severity(self):
+        """Test PASS with High severity: CIS 6.7 sets High in the remediation and Medium in the audit."""
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_leaked_password_alert_configured.rules_leaked_password_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_leaked_password_alert_configured.rules_leaked_password_alert_configured import (
+                rules_leaked_password_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    severity="HIGH",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_leaked_password_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "PASS"
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_below_the_minimum(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_leaked_password_alert_configured.rules_leaked_password_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_leaked_password_alert_configured.rules_leaked_password_alert_configured import (
+                rules_leaked_password_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    severity="LOW",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_leaked_password_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is LOW" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_not_configured(self):
+        """Test FAIL when the alert is on but no severity is configured."""
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_leaked_password_alert_configured.rules_leaked_password_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_leaked_password_alert_configured.rules_leaked_password_alert_configured import (
+                rules_leaked_password_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_leaked_password_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is not configured" in findings[0].status_extended
             assert findings[0].customer_id == CUSTOMER_ID
 
     def test_fail_alert_off(self):

tests/providers/googleworkspace/services/rules/rules_password_changed_alert_configured/rules_password_changed_alert_configured_test.py

@@ -46,7 +46,79 @@ class TestRulesPasswordChangedAlertConfigured:
 
             assert len(findings) == 1
             assert findings[0].status == "PASS"
-            assert "properly configured" in findings[0].status_extended
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_below_the_minimum(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_password_changed_alert_configured.rules_password_changed_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_password_changed_alert_configured.rules_password_changed_alert_configured import (
+                rules_password_changed_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    severity="LOW",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_password_changed_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is LOW" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_not_configured(self):
+        """Test FAIL when the alert is on but no severity is configured."""
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_password_changed_alert_configured.rules_password_changed_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_password_changed_alert_configured.rules_password_changed_alert_configured import (
+                rules_password_changed_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_password_changed_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is not configured" in findings[0].status_extended
             assert findings[0].customer_id == CUSTOMER_ID
 
     def test_fail_alert_off(self):

tests/providers/googleworkspace/services/rules/rules_service_test.py

@@ -1,9 +1,19 @@
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+from prowler.lib.check.models import CheckMetadata
+from prowler.providers.googleworkspace.services.rules import rules_service
 from tests.providers.googleworkspace.googleworkspace_fixtures import (
     set_mocked_googleworkspace_provider,
 )
 
+METADATA_FILE = (
+    Path(rules_service.__file__).parent
+    / "rules_government_backed_attacks_alert_configured"
+    / "rules_government_backed_attacks_alert_configured.metadata.json"
+)
+METADATA = CheckMetadata.parse_file(METADATA_FILE).json()
+
 
 class TestRulesService:
     def test_fetch_fully_configured_rule(self):
@@ -28,7 +38,10 @@ class TestRulesService:
                             "action": {
                                 "alertCenterAction": {
                                     "recipients": [{"allSuperAdmins": True}],
-                                    "alertCenterConfig": {"severity": "LOW"},
+                                    "alertCenterConfig": {
+                                        "severity": "LOW",
+                                        "status": "ENABLED",
+                                    },
                                 }
                             },
                             "state": "ACTIVE",
@@ -68,6 +81,7 @@ class TestRulesService:
             assert suspicious_login.email_notifications_enabled is True
             assert suspicious_login.all_super_admins is True
             assert suspicious_login.severity == "LOW"
+            assert suspicious_login.alert_center_status == "ENABLED"
 
     def test_fetch_rule_without_email_notifications(self):
         """Test a rule that is ACTIVE but has no email recipients configured."""
@@ -204,6 +218,114 @@ class TestRulesService:
             assert gov_attacks.email_notifications_enabled is True
             assert gov_attacks.all_super_admins is True
 
+    def test_ou_and_group_scoped_policies_are_skipped(self):
+        """Only the customer-level policy describes the whole domain"""
+        mock_provider = set_mocked_googleworkspace_provider()
+        mock_provider.audit_config = {}
+        mock_provider.fixer_config = {}
+        mock_session = MagicMock()
+        mock_session.credentials = MagicMock()
+        mock_provider.session = mock_session
+
+        def alert_policy(display_name, state, policy_query=None):
+            policy = {
+                "setting": {
+                    "type": "settings/rule.system_defined_alerts",
+                    "value": {"displayName": display_name, "state": state},
+                }
+            }
+            if policy_query:
+                policy["policyQuery"] = policy_query
+            return policy
+
+        mock_service = MagicMock()
+        mock_policies_list = MagicMock()
+        mock_policies_list.execute.return_value = {
+            "policies": [
+                alert_policy("Suspicious login", "ACTIVE"),
+                alert_policy(
+                    "Suspicious login", "INACTIVE", {"orgUnit": "orgUnits/sales_team"}
+                ),
+                alert_policy(
+                    "Leaked password", "INACTIVE", {"group": "groups/contractors"}
+                ),
+            ]
+        }
+        mock_service.policies().list.return_value = mock_policies_list
+        mock_service.policies().list_next.return_value = None
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_service.GoogleWorkspaceService._build_service",
+                return_value=mock_service,
+            ),
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_service import (
+                Rules,
+            )
+
+            rules = Rules(mock_provider)
+
+            by_name = {a.display_name: a for a in rules.system_defined_alerts}
+            assert by_name["Suspicious login"].state == "ACTIVE"
+            assert by_name["Suspicious login"].from_default is False
+            # Only seen in a group-scoped policy, so it falls back to the default.
+            assert by_name["Leaked password"].from_default is True
+
+    def test_empty_response_marks_alerts_as_inferred(self):
+        """A rule the API never returned must not be reported as tenant configuration."""
+        mock_provider = set_mocked_googleworkspace_provider()
+        mock_provider.audit_config = {}
+        mock_provider.fixer_config = {}
+        mock_session = MagicMock()
+        mock_session.credentials = MagicMock()
+        mock_provider.session = mock_session
+
+        mock_service = MagicMock()
+        mock_policies_list = MagicMock()
+        mock_policies_list.execute.return_value = {"policies": []}
+        mock_service.policies().list.return_value = mock_policies_list
+        mock_service.policies().list_next.return_value = None
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_service.GoogleWorkspaceService._build_service",
+                return_value=mock_service,
+            ),
+        ):
+            from prowler.providers.googleworkspace.services.rules.lib.alerts import (
+                evaluate_system_defined_alert,
+            )
+            from prowler.providers.googleworkspace.services.rules.rules_service import (
+                Rules,
+            )
+
+            rules = Rules(mock_provider)
+
+            assert all(alert.from_default for alert in rules.system_defined_alerts)
+
+            # End to end: nothing was observed for a rule that defaults to ON,
+            # so it has to be reviewed by hand instead of blamed on the tenant.
+            client = MagicMock()
+            client.provider = mock_provider
+            client.policies_fetched = True
+            client.system_defined_alerts = rules.system_defined_alerts
+            findings = evaluate_system_defined_alert(
+                client, METADATA, "Government-backed attacks", "HIGH"
+            )
+
+            assert len(findings) == 1
+            assert findings[0].status == "MANUAL"
+            assert "was not returned by the API" in findings[0].status_extended
+
     def test_api_error_sets_policies_fetched_false(self):
         """Test that API errors result in policies_fetched being False."""
         mock_provider = set_mocked_googleworkspace_provider()

tests/providers/googleworkspace/services/rules/rules_suspicious_activity_suspension_alert_configured/rules_suspicious_activity_suspension_alert_configured_test.py

@@ -35,7 +35,7 @@ class TestRulesSuspiciousActivitySuspensionAlertConfigured:
                 SystemDefinedAlert(
                     display_name=RULE_NAME,
                     state="ACTIVE",
-                    severity="MEDIUM",
+                    severity="HIGH",
                     email_notifications_enabled=True,
                     all_super_admins=True,
                 )
@@ -46,7 +46,79 @@ class TestRulesSuspiciousActivitySuspensionAlertConfigured:
 
             assert len(findings) == 1
             assert findings[0].status == "PASS"
-            assert "properly configured" in findings[0].status_extended
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_below_the_minimum(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_suspicious_activity_suspension_alert_configured.rules_suspicious_activity_suspension_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_suspicious_activity_suspension_alert_configured.rules_suspicious_activity_suspension_alert_configured import (
+                rules_suspicious_activity_suspension_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    severity="LOW",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_suspicious_activity_suspension_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is LOW" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_not_configured(self):
+        """Test FAIL when the alert is on but no severity is configured."""
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_suspicious_activity_suspension_alert_configured.rules_suspicious_activity_suspension_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_suspicious_activity_suspension_alert_configured.rules_suspicious_activity_suspension_alert_configured import (
+                rules_suspicious_activity_suspension_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_suspicious_activity_suspension_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is not configured" in findings[0].status_extended
             assert findings[0].customer_id == CUSTOMER_ID
 
     def test_fail_alert_off(self):
@@ -105,7 +177,7 @@ class TestRulesSuspiciousActivitySuspensionAlertConfigured:
                 SystemDefinedAlert(
                     display_name=RULE_NAME,
                     state="ACTIVE",
-                    severity="MEDIUM",
+                    severity="HIGH",
                     email_notifications_enabled=False,
                     all_super_admins=False,
                 )
@@ -141,7 +213,7 @@ class TestRulesSuspiciousActivitySuspensionAlertConfigured:
                 SystemDefinedAlert(
                     display_name=RULE_NAME,
                     state="ACTIVE",
-                    severity="MEDIUM",
+                    severity="HIGH",
                     email_notifications_enabled=True,
                     all_super_admins=False,
                 )

tests/providers/googleworkspace/services/rules/rules_suspicious_login_alert_configured/rules_suspicious_login_alert_configured_test.py

@@ -16,6 +16,42 @@ class TestRulesSuspiciousLoginAlertConfigured:
         """Test PASS when alert is ON, email notifications ON, recipients = all super admins."""
         mock_provider = set_mocked_googleworkspace_provider()
 
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_suspicious_login_alert_configured.rules_suspicious_login_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_suspicious_login_alert_configured.rules_suspicious_login_alert_configured import (
+                rules_suspicious_login_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    severity="LOW",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_suspicious_login_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "PASS"
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_pass_severity_above_the_minimum(self):
+        mock_provider = set_mocked_googleworkspace_provider()
+
         with (
             patch(
                 "prowler.providers.common.provider.Provider.get_global_provider",
@@ -46,7 +82,43 @@ class TestRulesSuspiciousLoginAlertConfigured:
 
             assert len(findings) == 1
             assert findings[0].status == "PASS"
-            assert "properly configured" in findings[0].status_extended
+            assert "is properly configured" in findings[0].status_extended
+            assert findings[0].customer_id == CUSTOMER_ID
+
+    def test_fail_severity_not_configured(self):
+        """Test FAIL when the alert is on but no severity is configured."""
+        mock_provider = set_mocked_googleworkspace_provider()
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=mock_provider,
+            ),
+            patch(
+                "prowler.providers.googleworkspace.services.rules.rules_suspicious_login_alert_configured.rules_suspicious_login_alert_configured.rules_client"
+            ) as mock_rules_client,
+        ):
+            from prowler.providers.googleworkspace.services.rules.rules_suspicious_login_alert_configured.rules_suspicious_login_alert_configured import (
+                rules_suspicious_login_alert_configured,
+            )
+
+            mock_rules_client.provider = mock_provider
+            mock_rules_client.policies_fetched = True
+            mock_rules_client.system_defined_alerts = [
+                SystemDefinedAlert(
+                    display_name=RULE_NAME,
+                    state="ACTIVE",
+                    email_notifications_enabled=True,
+                    all_super_admins=True,
+                )
+            ]
+
+            check = rules_suspicious_login_alert_configured()
+            findings = check.execute()
+
+            assert len(findings) == 1
+            assert findings[0].status == "FAIL"
+            assert "severity is not configured" in findings[0].status_extended
             assert findings[0].customer_id == CUSTOMER_ID
 
     def test_fail_alert_off(self):
@@ -105,7 +177,7 @@ class TestRulesSuspiciousLoginAlertConfigured:
                 SystemDefinedAlert(
                     display_name=RULE_NAME,
                     state="ACTIVE",
-                    severity="MEDIUM",
+                    severity="LOW",
                     email_notifications_enabled=False,
                     all_super_admins=False,
                 )
@@ -141,7 +213,7 @@ class TestRulesSuspiciousLoginAlertConfigured:
                 SystemDefinedAlert(
                     display_name=RULE_NAME,
                     state="ACTIVE",
-                    severity="MEDIUM",
+                    severity="LOW",
                     email_notifications_enabled=True,
                     all_super_admins=False,
                 )