fix(ui): build onboarding replay URL with the right query separator

A flow route may already carry a query string (e.g. /scans?tab=active), so
appending ?onboarding=<id> produced a malformed double-? URL. Pick & or ?
based on whether the route already has a query, in both the navbar replay
and the gate, and update the stale test expectation.

.github/workflows/api-codeql.yml

@@ -61,12 +61,12 @@ jobs:
           persist-credentials: false
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@7211b7c8077ea37d8641b6271f6a365a22a5fbfa # v4.36.0
+        uses: github/codeql-action/init@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v4.35.4
         with:
           languages: ${{ matrix.language }}
           config-file: ./.github/codeql/api-codeql-config.yml
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@7211b7c8077ea37d8641b6271f6a365a22a5fbfa # v4.36.0
+        uses: github/codeql-action/analyze@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v4.35.4
         with:
           category: '/language:${{ matrix.language }}'

.github/workflows/sdk-codeql.yml

@@ -66,12 +66,12 @@ jobs:
           persist-credentials: false
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@7211b7c8077ea37d8641b6271f6a365a22a5fbfa # v4.36.0
+        uses: github/codeql-action/init@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v4.35.4
         with:
           languages: ${{ matrix.language }}
           config-file: ./.github/codeql/sdk-codeql-config.yml
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@7211b7c8077ea37d8641b6271f6a365a22a5fbfa # v4.36.0
+        uses: github/codeql-action/analyze@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v4.35.4
         with:
           category: '/language:${{ matrix.language }}'

.github/workflows/ui-codeql.yml

@@ -62,12 +62,12 @@ jobs:
           persist-credentials: false
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@7211b7c8077ea37d8641b6271f6a365a22a5fbfa # v4.36.0
+        uses: github/codeql-action/init@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v4.35.4
         with:
           languages: ${{ matrix.language }}
           config-file: ./.github/codeql/ui-codeql-config.yml
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@7211b7c8077ea37d8641b6271f6a365a22a5fbfa # v4.36.0
+        uses: github/codeql-action/analyze@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v4.35.4
         with:
           category: '/language:${{ matrix.language }}'

.github/workflows/ui-tests.yml

@@ -131,6 +131,10 @@ jobs:
         if: steps.check-changes.outputs.any_changed == 'true'
         run: pnpm run healthcheck
 
+      - name: Check product-tour alignment
+        if: steps.check-changes.outputs.any_changed == 'true'
+        run: pnpm run tour:check
+
       - name: Run pnpm audit
         if: steps.check-changes.outputs.any_changed == 'true'
         run: pnpm run audit

AGENTS.md

@@ -51,6 +51,7 @@ Use these skills for detailed patterns on-demand:
 | `django-migration-psql` | Django migration best practices for PostgreSQL | [SKILL.md](skills/django-migration-psql/SKILL.md) |
 | `postgresql-indexing` | PostgreSQL indexing, EXPLAIN, monitoring, maintenance | [SKILL.md](skills/postgresql-indexing/SKILL.md) |
 | `prowler-attack-paths-query` | Create Attack Paths openCypher queries | [SKILL.md](skills/prowler-attack-paths-query/SKILL.md) |
+| `prowler-tour` | Keep product-tour definitions aligned with the UI | [SKILL.md](skills/prowler-tour/SKILL.md) |
 | `gh-aw` | GitHub Agentic Workflows (gh-aw) | [SKILL.md](skills/gh-aw/SKILL.md) |
 | `skill-creator` | Create new AI agent skills | [SKILL.md](skills/skill-creator/SKILL.md) |
 
@@ -67,10 +68,12 @@ When performing these actions, ALWAYS invoke the corresponding skill FIRST:
 | Adding new providers | `prowler-provider` |
 | Adding privilege escalation detection queries | `prowler-attack-paths-query` |
 | Adding services to existing providers | `prowler-provider` |
+| Adding, updating, or removing a tour definition (*.tour.ts) | `prowler-tour` |
 | After creating/modifying a skill | `skill-sync` |
 | App Router / Server Actions | `nextjs-16` |
 | Auditing check-to-requirement mappings as a cloud auditor | `prowler-compliance` |
 | Building AI chat features | `ai-sdk-5` |
+| Changing button labels or section headings on a tour-covered page | `prowler-tour` |
 | Committing changes | `prowler-commit` |
 | Configuring MCP servers in agentic workflows | `gh-aw` |
 | Create PR that requires changelog entry | `prowler-changelog` |
@@ -89,6 +92,7 @@ When performing these actions, ALWAYS invoke the corresponding skill FIRST:
 | Creating/updating compliance frameworks | `prowler-compliance` |
 | Debug why a GitHub Actions job is failing | `prowler-ci` |
 | Debugging gh-aw compilation errors | `gh-aw` |
+| Editing a UI file containing data-tour-id attributes | `prowler-tour` |
 | Fill .github/pull_request_template.md (Context/Description/Steps to review/Checklist) | `prowler-pr` |
 | Fixing bug | `tdd` |
 | Fixing compliance JSON bugs (duplicate IDs, empty Section, stale refs) | `prowler-compliance` |
@@ -105,6 +109,8 @@ When performing these actions, ALWAYS invoke the corresponding skill FIRST:
 | Modifying gh-aw workflow frontmatter or safe-outputs | `gh-aw` |
 | Refactoring code | `tdd` |
 | Regenerate AGENTS.md Auto-invoke tables (sync.sh) | `skill-sync` |
+| Renaming or removing a data-tour-id attribute value | `prowler-tour` |
+| Restructuring routes or layouts covered by a tour | `prowler-tour` |
 | Review PR requirements: template, title conventions, changelog gate | `prowler-pr` |
 | Review changelog format and conventions | `prowler-changelog` |
 | Reviewing JSON:API compliance | `jsonapi` |

action.yml

@@ -167,7 +167,7 @@ runs:
 
     - name: Upload SARIF to GitHub Code Scanning
       if: always() && inputs.upload-sarif == 'true' && steps.find-sarif.outputs.sarif_path != ''
-      uses: github/codeql-action/upload-sarif@7211b7c8077ea37d8641b6271f6a365a22a5fbfa # v4.36.0
+      uses: github/codeql-action/upload-sarif@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v4.35.4
       with:
         sarif_file: ${{ steps.find-sarif.outputs.sarif_path }}
         category: ${{ inputs.sarif-category }}

api/CHANGELOG.md

@@ -9,6 +9,7 @@ All notable changes to the **Prowler API** are documented in this file.
 - Automatic recovery of allowlisted idempotent background tasks whose worker died during a deploy or crash: stuck scan and summary tasks are detected and re-run instead of staying pending forever, with a `reconcile_orphan_tasks` management command for on-demand recovery [(#11416)](https://github.com/prowler-cloud/prowler/pull/11416)
 - Jira integration no longer creates duplicate issues on a retried send; findings already ticketed are skipped [(#11416)](https://github.com/prowler-cloud/prowler/pull/11416)
 - DORA compliance framework support [(#11131)](https://github.com/prowler-cloud/prowler/pull/11131)
+- Label Postgres connections with `application_name="<component>:<alias>"` (component injected per process via `DJANGO_APP_COMPONENT`) so connections are attributable by component in `pg_stat_activity` [(#11494)](https://github.com/prowler-cloud/prowler/pull/11494)
 
 ### 🔄 Changed
 
@@ -21,6 +22,14 @@ All notable changes to the **Prowler API** are documented in this file.
 
 ---
 
+## [1.30.3] (Prowler v5.29.3)
+
+### 🐞 Fixed
+
+- API startup no longer crashes when Neo4j is unreachable, as the Neo4j driver now connects lazily on first use rather than during app initialization [(#11491)](https://github.com/prowler-cloud/prowler/pull/11491)
+
+---
+
 ## [1.30.1] (Prowler v5.29.1)
 
 ### 🐞 Fixed

api/docker-entrypoint.sh

@@ -68,6 +68,15 @@ manage_db_partitions() {
   fi
 }
 
+# Identify this process to Postgres (application_name=<component>:<alias>) so
+# connections are attributable by component in pg_stat_activity. Web tiers
+# report "api"; everything else uses the launch subcommand.
+case "$1" in
+  prod|dev) DJANGO_APP_COMPONENT="api" ;;
+  *)        DJANGO_APP_COMPONENT="$1" ;;
+esac
+export DJANGO_APP_COMPONENT
+
 case "$1" in
   dev)
     apply_migrations

api/src/backend/api/apps.py

@@ -1,12 +1,14 @@
 import logging
 import os
 import sys
+
 from pathlib import Path
 
+from django.apps import AppConfig
+from django.conf import settings
+
 from config.custom_logging import BackendLogger
 from config.env import env
-from django.apps import AppConfig
-from django.conf import settings
 
 logger = logging.getLogger(BackendLogger.API)
 
@@ -30,7 +32,6 @@ class ApiConfig(AppConfig):
     def ready(self):
         from api import schema_extensions  # noqa: F401
         from api import signals  # noqa: F401
-        from api.attack_paths import database as graph_database
 
         # Generate required cryptographic keys if not present, but only if:
         #   `"manage.py" not in sys.argv[0]`: If an external server (e.g., Gunicorn) is running the app
@@ -41,37 +42,8 @@ class ApiConfig(AppConfig):
         ):
             self._ensure_crypto_keys()
 
-        # Commands that don't need Neo4j
-        SKIP_NEO4J_DJANGO_COMMANDS = [
-            "makemigrations",
-            "migrate",
-            "pgpartition",
-            "check",
-            "help",
-            "showmigrations",
-            "check_and_fix_socialaccount_sites_migration",
-        ]
-
-        # Skip eager Neo4j init for tests, some Django commands, and Celery (prefork pool: driver must stay lazy, no post_fork hook)
-        if getattr(settings, "TESTING", False) or (
-            len(sys.argv) > 1
-            and (
-                (
-                    "manage.py" in sys.argv[0]
-                    and sys.argv[1] in SKIP_NEO4J_DJANGO_COMMANDS
-                )
-                or "celery" in sys.argv[0]
-            )
-        ):
-            logger.info(
-                "Skipping eager Neo4j init: tests, some Django commands, or Celery prefork pool (driver stays lazy)"
-            )
-
-        else:
-            graph_database.init_driver()
-
-        # Neo4j driver is initialized at API startup (see api.attack_paths.database)
-        # It remains lazy for Celery workers and selected Django commands
+        # Neo4j driver is created lazily on first use (see api.attack_paths.database).
+        # App init never contacts Neo4j, so a Neo4j outage cannot block API startup.
 
     def _ensure_crypto_keys(self):
         """

api/src/backend/api/attack_paths/database.py

@@ -1,22 +1,24 @@
 import atexit
 import logging
 import threading
+
 from contextlib import contextmanager
 from typing import Any, Iterator
 from uuid import UUID
 
 import neo4j
 import neo4j.exceptions
+
 from config.env import env
 from django.conf import settings
+
+from api.attack_paths.retryable_session import RetryableSession
 from tasks.jobs.attack_paths.config import (
     BATCH_SIZE,
     PROVIDER_RESOURCE_LABEL,
     get_provider_label,
 )
 
-from api.attack_paths.retryable_session import RetryableSession
-
 # Without this Celery goes crazy with Neo4j logging
 logging.getLogger("neo4j").setLevel(logging.ERROR)
 logging.getLogger("neo4j").propagate = False
@@ -28,6 +30,9 @@ READ_QUERY_TIMEOUT_SECONDS = env.int(
     "ATTACK_PATHS_READ_QUERY_TIMEOUT_SECONDS", default=30
 )
 MAX_CUSTOM_QUERY_NODES = env.int("ATTACK_PATHS_MAX_CUSTOM_QUERY_NODES", default=250)
+# Shorter than CONN_ACQUISITION_TIMEOUT — the driver requires acquisition to be
+# the longer of the two (it may include opening a new connection).
+CONNECTION_TIMEOUT = env.int("NEO4J_CONNECTION_TIMEOUT", default=5)
 CONN_ACQUISITION_TIMEOUT = env.int("NEO4J_CONN_ACQUISITION_TIMEOUT", default=15)
 READ_EXCEPTION_CODES = [
     "Neo.ClientError.Statement.AccessMode",
@@ -58,15 +63,24 @@ def init_driver() -> neo4j.Driver:
             uri = get_uri()
             config = settings.DATABASES["neo4j"]
 
-            _driver = neo4j.GraphDatabase.driver(
+            driver = neo4j.GraphDatabase.driver(
                 uri,
                 auth=(config["USER"], config["PASSWORD"]),
                 keep_alive=True,
                 max_connection_lifetime=7200,
+                connection_timeout=CONNECTION_TIMEOUT,
                 connection_acquisition_timeout=CONN_ACQUISITION_TIMEOUT,
                 max_connection_pool_size=50,
             )
-            _driver.verify_connectivity()
+            # Publish the singleton only after connectivity is verified so a
+            # failed probe does not leave an unverified driver behind. Close the
+            # driver on failure so a repeatedly-probed outage cannot leak pools.
+            try:
+                driver.verify_connectivity()
+            except Exception:
+                driver.close()
+                raise
+            _driver = driver
 
             # Register cleanup handler (only runs once since we're inside the _driver is None block)
             atexit.register(close_driver)

api/src/backend/api/tests/test_apps.py

@@ -182,23 +182,19 @@ def _make_app():
     return ApiConfig("api", api)
 
 
-def test_ready_initializes_driver_for_api_process(monkeypatch):
+@pytest.mark.parametrize(
+    "argv",
+    [
+        ["gunicorn"],
+        ["celery", "-A", "api"],
+        ["manage.py", "migrate"],
+    ],
+    ids=["api", "celery", "manage_py"],
+)
+def test_ready_never_eagerly_initializes_neo4j_driver(monkeypatch, argv):
+    """ready() must never contact Neo4j; the driver is created lazily on first use."""
     config = _make_app()
-    _set_argv(monkeypatch, ["gunicorn"])
-    _set_testing(monkeypatch, False)
-
-    with (
-        patch.object(ApiConfig, "_ensure_crypto_keys", return_value=None),
-        patch("api.attack_paths.database.init_driver") as init_driver,
-    ):
-        config.ready()
-
-    init_driver.assert_called_once()
-
-
-def test_ready_skips_driver_for_celery(monkeypatch):
-    config = _make_app()
-    _set_argv(monkeypatch, ["celery", "-A", "api"])
+    _set_argv(monkeypatch, argv)
     _set_testing(monkeypatch, False)
 
     with (
@@ -208,31 +204,3 @@ def test_ready_skips_driver_for_celery(monkeypatch):
         config.ready()
 
     init_driver.assert_not_called()
-
-
-def test_ready_skips_driver_for_manage_py_skip_command(monkeypatch):
-    config = _make_app()
-    _set_argv(monkeypatch, ["manage.py", "migrate"])
-    _set_testing(monkeypatch, False)
-
-    with (
-        patch.object(ApiConfig, "_ensure_crypto_keys", return_value=None),
-        patch("api.attack_paths.database.init_driver") as init_driver,
-    ):
-        config.ready()
-
-    init_driver.assert_not_called()
-
-
-def test_ready_skips_driver_when_testing(monkeypatch):
-    config = _make_app()
-    _set_argv(monkeypatch, ["gunicorn"])
-    _set_testing(monkeypatch, True)
-
-    with (
-        patch.object(ApiConfig, "_ensure_crypto_keys", return_value=None),
-        patch("api.attack_paths.database.init_driver") as init_driver,
-    ):
-        config.ready()
-
-    init_driver.assert_not_called()

api/src/backend/api/tests/test_attack_paths_database.py

@@ -1,15 +1,16 @@
 """
 Tests for Neo4j database lazy initialization.
 
-The Neo4j driver connects on first use by default. API processes may
-eagerly initialize the driver during app startup, while Celery workers
-remain lazy. These tests validate the database module behavior itself.
+The Neo4j driver is created on first use for every process type; app startup
+never contacts Neo4j. These tests validate the database module behavior itself.
 """
 
 import threading
+
 from unittest.mock import MagicMock, patch
 
 import neo4j
+import neo4j.exceptions
 import pytest
 
 import api.attack_paths.database as db_module
@@ -59,6 +60,32 @@ class TestLazyInitialization:
         assert result is mock_driver
         assert db_module._driver is mock_driver
 
+    @patch("api.attack_paths.database.settings")
+    @patch("api.attack_paths.database.neo4j.GraphDatabase.driver")
+    def test_init_driver_leaves_driver_none_when_verify_fails(
+        self, mock_driver_factory, mock_settings
+    ):
+        """A failed verify_connectivity() must not publish or leak the driver."""
+        mock_driver = MagicMock()
+        mock_driver.verify_connectivity.side_effect = (
+            neo4j.exceptions.ServiceUnavailable("down")
+        )
+        mock_driver_factory.return_value = mock_driver
+        mock_settings.DATABASES = {
+            "neo4j": {
+                "HOST": "localhost",
+                "PORT": 7687,
+                "USER": "neo4j",
+                "PASSWORD": "password",
+            }
+        }
+
+        with pytest.raises(neo4j.exceptions.ServiceUnavailable):
+            db_module.init_driver()
+
+        assert db_module._driver is None
+        mock_driver.close.assert_called_once()
+
     @patch("api.attack_paths.database.settings")
     @patch("api.attack_paths.database.neo4j.GraphDatabase.driver")
     def test_init_driver_returns_cached_driver_on_subsequent_calls(
@@ -116,21 +143,23 @@ class TestConnectionAcquisitionTimeout:
     @pytest.fixture(autouse=True)
     def reset_module_state(self):
         original_driver = db_module._driver
-        original_timeout = db_module.CONN_ACQUISITION_TIMEOUT
+        original_acq_timeout = db_module.CONN_ACQUISITION_TIMEOUT
+        original_conn_timeout = db_module.CONNECTION_TIMEOUT
 
         db_module._driver = None
 
         yield
 
         db_module._driver = original_driver
-        db_module.CONN_ACQUISITION_TIMEOUT = original_timeout
+        db_module.CONN_ACQUISITION_TIMEOUT = original_acq_timeout
+        db_module.CONNECTION_TIMEOUT = original_conn_timeout
 
     @patch("api.attack_paths.database.settings")
     @patch("api.attack_paths.database.neo4j.GraphDatabase.driver")
     def test_driver_receives_configured_timeout(
         self, mock_driver_factory, mock_settings
     ):
-        """init_driver() should pass CONN_ACQUISITION_TIMEOUT to the neo4j driver."""
+        """init_driver() should pass the configured timeouts to the neo4j driver."""
         mock_driver_factory.return_value = MagicMock()
         mock_settings.DATABASES = {
             "neo4j": {
@@ -141,11 +170,13 @@ class TestConnectionAcquisitionTimeout:
             }
         }
         db_module.CONN_ACQUISITION_TIMEOUT = 42
+        db_module.CONNECTION_TIMEOUT = 7
 
         db_module.init_driver()
 
         _, kwargs = mock_driver_factory.call_args
         assert kwargs["connection_acquisition_timeout"] == 42
+        assert kwargs["connection_timeout"] == 7
 
 
 class TestAtexitRegistration:

api/src/backend/api/tests/test_db_connection_labels.py

@@ -0,0 +1,55 @@
+from config.django.base import label_postgres_connections
+
+
+class TestLabelPostgresConnections:
+    def test_labels_postgres_and_skips_neo4j(self, monkeypatch):
+        monkeypatch.setenv("DJANGO_APP_COMPONENT", "scan")
+        databases = {
+            "default": {"ENGINE": "psqlextra.backend"},
+            "neo4j": {"HOST": "neo4j", "PORT": "7687"},
+        }
+
+        label_postgres_connections(databases)
+
+        assert databases["default"]["OPTIONS"]["application_name"] == "scan:default"
+        assert "OPTIONS" not in databases["neo4j"]
+
+    def test_labels_plain_postgresql_backend(self, monkeypatch):
+        monkeypatch.setenv("DJANGO_APP_COMPONENT", "api")
+        databases = {"saas": {"ENGINE": "django.db.backends.postgresql"}}
+
+        label_postgres_connections(databases)
+
+        assert databases["saas"]["OPTIONS"]["application_name"] == "api:saas"
+
+    def test_defaults_component_to_api_when_unset(self, monkeypatch):
+        monkeypatch.delenv("DJANGO_APP_COMPONENT", raising=False)
+        databases = {"default": {"ENGINE": "psqlextra.backend"}}
+
+        label_postgres_connections(databases)
+
+        assert databases["default"]["OPTIONS"]["application_name"] == "api:default"
+
+    def test_preserves_existing_options(self, monkeypatch):
+        monkeypatch.setenv("DJANGO_APP_COMPONENT", "worker")
+        databases = {
+            "replica": {
+                "ENGINE": "psqlextra.backend",
+                "OPTIONS": {"sslmode": "require"},
+            }
+        }
+
+        label_postgres_connections(databases)
+
+        assert databases["replica"]["OPTIONS"] == {
+            "sslmode": "require",
+            "application_name": "worker:replica",
+        }
+
+    def test_truncates_application_name_to_63_bytes(self, monkeypatch):
+        monkeypatch.setenv("DJANGO_APP_COMPONENT", "c" * 80)
+        databases = {"default": {"ENGINE": "psqlextra.backend"}}
+
+        label_postgres_connections(databases)
+
+        assert len(databases["default"]["OPTIONS"]["application_name"]) == 63

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

@@ -306,3 +306,20 @@ SESSION_COOKIE_SECURE = True
 ATTACK_PATHS_SCAN_STALE_THRESHOLD_MINUTES = env.int(
     "ATTACK_PATHS_SCAN_STALE_THRESHOLD_MINUTES", 2880
 )  # 48h
+
+
+def label_postgres_connections(databases):
+    """Tag each Postgres connection with ``application_name="<component>:<alias>"``
+    so connections are attributable by component in ``pg_stat_activity`` (and any
+    tooling that surfaces ``application_name``). The component (api / worker /
+    scan / ...) is injected per process by the container entrypoint via
+    ``DJANGO_APP_COMPONENT``; the alias distinguishes which pool inside the
+    process owns the connection. The neo4j entry is skipped (not a Postgres
+    backend). Postgres truncates ``application_name`` at 63 bytes.
+    """
+    component = env.str("DJANGO_APP_COMPONENT", default="api")
+    for alias, config in databases.items():
+        engine = config.get("ENGINE", "")
+        if engine.startswith("psqlextra") or "postgresql" in engine:
+            name = f"{component}:{alias}"[:63]
+            config.setdefault("OPTIONS", {})["application_name"] = name

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

@@ -54,6 +54,8 @@ DATABASES = {
 
 DATABASES["default"] = DATABASES["prowler_user"]
 
+label_postgres_connections(DATABASES)  # noqa: F405
+
 REST_FRAMEWORK["DEFAULT_RENDERER_CLASSES"] = tuple(  # noqa: F405
     render_class
     for render_class in REST_FRAMEWORK["DEFAULT_RENDERER_CLASSES"]  # noqa: F405

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

@@ -58,3 +58,5 @@ DATABASES = {
 }
 
 DATABASES["default"] = DATABASES["prowler_user"]
+
+label_postgres_connections(DATABASES)  # noqa: F405

dashboard/compliance/aws_ai_security_framework_aws.py

@@ -0,0 +1,27 @@
+import warnings
+
+from dashboard.common_methods import get_section_containers_3_levels
+
+warnings.filterwarnings("ignore")
+
+
+def get_table(data):
+    aux = data[
+        [
+            "REQUIREMENTS_ATTRIBUTES_SECTION",
+            "REQUIREMENTS_ATTRIBUTES_SUBSECTION",
+            "NAME",
+            "CHECKID",
+            "STATUS",
+            "REGION",
+            "ACCOUNTID",
+            "RESOURCEID",
+        ]
+    ]
+
+    return get_section_containers_3_levels(
+        aux,
+        "REQUIREMENTS_ATTRIBUTES_SECTION",
+        "REQUIREMENTS_ATTRIBUTES_SUBSECTION",
+        "NAME",
+    )

docs/user-guide/providers/okta/authentication.mdx

@@ -35,14 +35,18 @@ The bundled checks require the following read-only scopes:
 - `okta.policies.read`
 - `okta.brands.read`
 - `okta.apps.read`
+- `okta.logStreams.read`
+- `okta.idps.read`
 
 Additional scopes will be needed as more services and checks are added. These are the current ones needed:
 
 | Scope | Used by |
 |---|---|
-| `okta.policies.read` | Sign-on, password, and authentication policies |
+| `okta.policies.read` | Sign-on, password, authentication, and `USER_LIFECYCLE` (Workflow > Automations) policies |
 | `okta.brands.read` | Sign-in page customizations (DOD Notice and Consent Banner check) |
 | `okta.apps.read` | First-party app settings (Okta Admin Console session), integrated app inventory, and the Authentication Policies bound to Okta applications |
+| `okta.logStreams.read` | Log Stream configuration (`/api/v1/logStreams`) |
+| `okta.idps.read` | Identity Providers, including Smart Card (X509) IdPs (`/api/v1/idps`) |
 
 ### Required Admin Role
 
@@ -68,7 +72,9 @@ Okta filters the first-party apps (`saasure`, `okta_enduser`) out of `/api/v1/ap
 
 A fifth check — `application_admin_console_session_idle_timeout_15min` (STIG V-273187) — also requires Super Administrator: it calls `GET /api/v1/first-party-app-settings/admin-console`, which returns `403 E0000006` for every role below Super Administrator.
 
-When the service app runs with Read-Only Administrator, the five checks listed in this section return **MANUAL** instead of PASS/FAIL — the rest of the scan keeps running.
+`user_inactivity_automation_35d_enabled` (STIG V-273188) reads `USER_LIFECYCLE` policies (`list_policies(type='USER_LIFECYCLE')`) using the `okta.policies.read` scope. The Read-Only Administrator role is enough to list them; no Super Administrator requirement.
+
+When the service app runs with Read-Only Administrator, the checks listed in this section return **MANUAL** instead of PASS/FAIL — the rest of the scan keeps running.
 
 <Note>
 Read-Only Administrator stays the recommended default for the least-privilege framing that aligns with DISA STIG. Assign Super Administrator on a separate run when full coverage of the first-party app checks is needed.
@@ -158,8 +164,8 @@ export OKTA_PRIVATE_KEY_FILE="/secure/path/to/prowler-okta.pem"
 # or
 export OKTA_PRIVATE_KEY="$(cat /secure/path/to/prowler-okta.pem)"
 
-# Optional — defaults to "okta.policies.read,okta.brands.read,okta.apps.read"
-export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read"
+# Optional — defaults to "okta.policies.read,okta.brands.read,okta.apps.read,okta.logStreams.read,okta.idps.read"
+export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read,okta.logStreams.read,okta.idps.read"
 
 uv run python prowler-cli.py okta
 ```

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

@@ -85,8 +85,8 @@ Follow the [Okta Authentication](/user-guide/providers/okta/authentication) guid
 export OKTA_ORG_DOMAIN="acme.okta.com"
 export OKTA_CLIENT_ID="0oa1234567890abcdef"
 export OKTA_PRIVATE_KEY_FILE="/secure/path/to/prowler-okta.pem"
-# Optional — defaults to "okta.policies.read,okta.brands.read,okta.apps.read"
-export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read"
+# Optional — defaults to "okta.policies.read,okta.brands.read,okta.apps.read,okta.logStreams.read,okta.idps.read"
+export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read,okta.logStreams.read,okta.idps.read"
 ```
 
 The private key file may contain either a PEM-encoded RSA key or a JWK JSON document.
@@ -143,10 +143,13 @@ prowler okta --config-file /path/to/config.yaml
 
 Prowler for Okta includes security checks across the following services:
 
-| Service         | Description                                                                                                                            |
-| --------------- | -------------------------------------------------------------------------------------------------------------------------------------- |
-| **Sign-On**     | Global session policy controls (idle timeout, lifetime, rule priority and ordering)                                                    |
-| **Application** | Okta Admin Console sign-on settings plus Authentication Policy controls for Okta applications (session idle, MFA, phishing resistance, network zones) |
+| Service               | Description                                                                                                                                           |
+| --------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
+| **Sign-On**           | Global session policy controls (idle timeout, lifetime, rule priority and ordering)                                                                   |
+| **Application**       | Okta Admin Console sign-on settings plus Authentication Policy controls for Okta applications (session idle, MFA, phishing resistance, network zones) |
+| **User**              | User lifecycle automations (inactivity-based deprovisioning)                                                                                          |
+| **System Log**        | Log Stream configuration that off-loads audit records to a central SIEM                                                                               |
+| **Identity Provider** | Identity Providers, including Smart Card (X509) IdP status and certificate-chain visibility                                                           |
 
 ## Troubleshooting
 
@@ -158,11 +161,13 @@ This is stricter than simply finding the same timeout value somewhere else in th
 
 ### Default Scopes
 
-Prowler requests a fixed set of OAuth scopes on every token exchange. The defaults cover every bundled check across the Sign-On and Application services:
+Prowler requests a fixed set of OAuth scopes on every token exchange. The defaults cover every bundled check across the Sign-On, Application, User, System Log, and Identity Provider services:
 
 - `okta.policies.read`
 - `okta.brands.read`
 - `okta.apps.read`
+- `okta.logStreams.read`
+- `okta.idps.read`
 
 The service app must have these scopes granted in the **Okta API Scopes** tab. When the granted set is narrower than the requested set, the token request fails with an `invalid_scope` error and the scan stops at provider initialization.
 
@@ -170,10 +175,10 @@ When additional checks are enabled — or when running against a service app tha
 
 ```bash
 # Environment variable — comma-separated
-export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read,okta.users.read"
+export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read,okta.logStreams.read,okta.idps.read,okta.users.read"
 
 # CLI flag — space-separated
-prowler okta --okta-scopes okta.policies.read okta.brands.read okta.apps.read okta.users.read
+prowler okta --okta-scopes okta.policies.read okta.brands.read okta.apps.read okta.logStreams.read okta.idps.read okta.users.read
 ```
 
 For the full catalog of OAuth scopes exposed by the Okta Management API, refer to the [Okta OAuth 2.0 scopes documentation](https://developer.okta.com/docs/api/oauth2/).

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

@@ -47,7 +47,11 @@ Follow these steps to remove a user of your account:
 1. Navigate to **Users** from the side menu.
 2. Click the delete button of your current user.
 
-> **Note: Each user will be able to delete himself and not others, regardless of his permissions.**
+> **Note: Each user can only delete their own account, regardless of their permissions. For this reason, the delete button is only shown on your own row and not on other users' rows.**
+
+Deleting a user removes the **entire user account** from Prowler, not just its membership in your organization. Because a single account can belong to more than one tenant, allowing one administrator to delete it outright could affect organizations they don't manage and irreversibly remove another person's identity. To keep this destructive action under the control of the account owner, the API only permits a user to delete themselves (it rejects any other target with a `400` response), and the UI mirrors this by showing the delete button exclusively on your own row.
+
+To remove **another** user from your organization, use the [_Expel from organization_](/user-guide/tutorials/prowler-app-multi-tenant#expelling-a-user-from-an-organization) action instead. Expelling removes the user's membership, role grants, and active sessions for your tenant only, and deletes the underlying account just for that user if your organization was their last remaining membership. This action is reserved for tenant **owners**.
 
 <img src="/images/prowler-app/rbac/user_remove.png" alt="Remove User" width="700" />
 

prowler/CHANGELOG.md

@@ -8,6 +8,19 @@ All notable changes to the **Prowler SDK** are documented in this file.
 
 - `sagemaker_models_monitor_enabled` check for AWS provider, verifying that each SageMaker monitoring schedule is in the `Scheduled` state so data and model drift is actively detected [(#11278)](https://github.com/prowler-cloud/prowler/pull/11278)
 - DORA (Digital Operational Resilience Act, Regulation (EU) 2022/2554) universal compliance framework with AWS provider coverage across the five DORA pillars [(#11131)](https://github.com/prowler-cloud/prowler/pull/11131)
+- `elbv2_alb_drop_invalid_header_fields_enabled` check for AWS provider, verifying Application Load Balancers have `routing.http.drop_invalid_header_fields.enabled` set to `true` to mitigate HTTP desync attacks (AWS FSBP ELB.4) [(#11471)](https://github.com/prowler-cloud/prowler/pull/11471)
+- `user`, `systemlog` and `idp` service for Okta provider with `user_inactivity_automation_35d_enabled`, `systemlog_streaming_enabled` and `idp_smart_card_dod_approved_ca` checks [(#11496)](https://github.com/prowler-cloud/prowler/pull/11496)
+
+---
+
+## [5.29.3] (Prowler UNRELEASED)
+
+### 🐞 Fixed
+
+- GCP `logging_sink_created` now recognizes organization-level aggregated sinks with `includeChildren=True`, avoiding false failures for covered projects [(#11355)](https://github.com/prowler-cloud/prowler/pull/11355)
+- Jira integration no longer fails with `400 INVALID_INPUT` when a finding has empty fields [(#11474)](https://github.com/prowler-cloud/prowler/pull/11474)
+- GCP `iam_service_account_unused` now passes disabled service accounts instead of failing them, since a disabled account cannot authenticate or be used [(#11467)](https://github.com/prowler-cloud/prowler/pull/11467)
+- AWS AI Security Framework now renders in the dashboard instead of showing "No data found for this compliance", by adding the missing compliance view module [(#11470)](https://github.com/prowler-cloud/prowler/pull/11470)
 
 ---
 

prowler/compliance/aws/aws_foundational_security_best_practices_aws.json

@@ -1863,7 +1863,9 @@
       "Id": "ELB.4",
       "Name": "Application load balancers should be configured to drop HTTP headers",
       "Description": "This control evaluates AWS Application Load Balancers (ALB) to ensure they are configured to drop invalid HTTP headers. The control fails if the value of routing.http.drop_invalid_header_fields.enabled is set to false. By default, ALBs are not configured to drop invalid HTTP header values. Removing these header values prevents HTTP desync attacks.",
-      "Checks": [],
+      "Checks": [
+        "elbv2_alb_drop_invalid_header_fields_enabled"
+      ],
       "Attributes": [
         {
           "ItemId": "ELB.4",

prowler/lib/outputs/jira/jira.py

@@ -229,7 +229,9 @@ class MarkdownToADFConverter:
         return node
 
     def _paragraph_with_text(self, text: str) -> Dict:
-        return {"type": "paragraph", "content": [self._create_text_node(text, None)]}
+        # ADF forbids empty text nodes; emit an empty paragraph instead.
+        content = [self._create_text_node(text, None)] if text else []
+        return {"type": "paragraph", "content": content}
 
     @staticmethod
     def _pop_mark(marks_stack: List[Dict], mark_type: str) -> None:
@@ -1118,6 +1120,18 @@ class Jira:
         tenant_info: str = "",
     ) -> dict:
 
+        # ADF forbids empty text nodes, so Jira rejects them with 400 INVALID_INPUT.
+        def _safe(value: str) -> str:
+            return value if (value and value.strip()) else "-"
+
+        check_id = _safe(check_id)
+        check_title = _safe(check_title)
+        status_extended = _safe(status_extended)
+        provider = _safe(provider)
+        region = _safe(region)
+        resource_uid = _safe(resource_uid)
+        resource_name = _safe(resource_name)
+
         table_rows = [
             {
                 "type": "tableRow",

prowler/providers/aws/services/elbv2/elbv2_alb_drop_invalid_header_fields_enabled/elbv2_alb_drop_invalid_header_fields_enabled.metadata.json

@@ -0,0 +1,40 @@
+{
+  "Provider": "aws",
+  "CheckID": "elbv2_alb_drop_invalid_header_fields_enabled",
+  "CheckTitle": "Application Load Balancer should be configured to drop invalid HTTP header fields",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices/Network Reachability",
+    "Software and Configuration Checks/Industry and Regulatory Standards/AWS Foundational Security Best Practices",
+    "TTPs/Initial Access",
+    "Effects/Data Exposure"
+  ],
+  "ServiceName": "elbv2",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "AwsElbv2LoadBalancer",
+  "ResourceGroup": "network",
+  "Description": "Ensure that Application Load Balancers (ALB) are configured to drop invalid HTTP header fields. The check fails when `routing.http.drop_invalid_header_fields.enabled` is not set to `true`. By default, ALBs do not remove HTTP headers that do not conform to RFC 7230.",
+  "Risk": "Forwarding non-RFC-compliant HTTP headers to backend targets enables HTTP desync (request smuggling):\n- **Confidentiality**: session/token theft, data exfiltration\n- **Integrity**: cache poisoning, request routing bypass, unauthorized actions\n- **Availability**: backend exhaustion.\nDropping invalid header fields removes a primary smuggling vector.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/elasticloadbalancing/latest/application/application-load-balancers.html#drop-invalid-header-fields",
+    "https://docs.aws.amazon.com/securityhub/latest/userguide/elb-controls.html#elb-4"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "aws elbv2 modify-load-balancer-attributes --load-balancer-arn <ALB_ARN> --attributes Key=routing.http.drop_invalid_header_fields.enabled,Value=true",
+      "NativeIaC": "```yaml\n# CloudFormation: enable drop invalid header fields on an ALB\nResources:\n  <example_resource_name>:\n    Type: AWS::ElasticLoadBalancingV2::LoadBalancer\n    Properties:\n      Type: application\n      Subnets:\n        - <example_subnet_id1>\n        - <example_subnet_id2>\n      LoadBalancerAttributes:\n        - Key: routing.http.drop_invalid_header_fields.enabled  # Critical: drop non-RFC-compliant headers\n          Value: true\n```",
+      "Other": "1. Open the Amazon EC2 console and choose Load Balancers.\n2. Select the Application Load Balancer.\n3. On the Attributes tab, choose Edit.\n4. Set 'Drop invalid header fields' to Enabled.\n5. Save changes.",
+      "Terraform": "```hcl\n# Terraform: enable drop invalid header fields on an ALB\nresource \"aws_lb\" \"<example_resource_name>\" {\n  name                       = \"<example_resource_name>\"\n  load_balancer_type         = \"application\"\n  subnets                    = [\"<example_subnet_id1>\", \"<example_subnet_id2>\"]\n  drop_invalid_header_fields = true # Critical: drop non-RFC-compliant headers\n}\n```"
+    },
+    "Recommendation": {
+      "Text": "Enable 'drop invalid header fields' on Application Load Balancers so non-RFC-compliant HTTP headers are removed before requests reach backend targets, reducing exposure to HTTP desync and request smuggling. Apply defense in depth and validate requests at the application layer as well.",
+      "Url": "https://hub.prowler.com/check/elbv2_alb_drop_invalid_header_fields_enabled"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/elbv2/elbv2_alb_drop_invalid_header_fields_enabled/elbv2_alb_drop_invalid_header_fields_enabled.py

@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
+
+
+class elbv2_alb_drop_invalid_header_fields_enabled(Check):
+    def execute(self):
+        findings = []
+        for lb in elbv2_client.loadbalancersv2.values():
+            if lb.type == "application":
+                report = Check_Report_AWS(
+                    metadata=self.metadata(),
+                    resource=lb,
+                )
+                report.status = "PASS"
+                report.status_extended = (
+                    f"ELBv2 ALB {lb.name} is configured to drop invalid "
+                    "header fields."
+                )
+                if lb.drop_invalid_header_fields != "true":
+                    report.status = "FAIL"
+                    report.status_extended = (
+                        f"ELBv2 ALB {lb.name} is not configured to drop "
+                        "invalid header fields."
+                    )
+                findings.append(report)
+
+        return findings

prowler/providers/gcp/services/iam/iam_service.py

@@ -37,6 +37,7 @@ class IAM(GCPService):
                                 display_name=account.get("displayName", ""),
                                 project_id=project_id,
                                 uniqueId=account.get("uniqueId", ""),
+                                disabled=account.get("disabled", False),
                             )
                         )
 
@@ -102,6 +103,7 @@ class ServiceAccount(BaseModel):
     keys: list[Key] = []
     project_id: str
     uniqueId: str
+    disabled: bool = False
 
 
 class AccessApproval(GCPService):

prowler/providers/gcp/services/iam/iam_service_account_unused/iam_service_account_unused.py

@@ -19,7 +19,12 @@ class iam_service_account_unused(Check):
                 resource_id=account.email,
                 location=iam_client.region,
             )
-            if account.uniqueId in sa_ids_used:
+            if account.disabled:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"Service Account {account.email} is disabled and cannot be used."
+                )
+            elif account.uniqueId in sa_ids_used:
                 report.status = "PASS"
                 report.status_extended = f"Service Account {account.email} was used over the last {max_unused_days} days."
             else:

prowler/providers/gcp/services/logging/logging_service.py

@@ -12,6 +12,7 @@ class Logging(GCPService):
         self.sinks = []
         self.metrics = []
         self._get_sinks()
+        self._get_org_sinks()
         self._get_metrics()
 
     def _get_sinks(self):
@@ -39,6 +40,38 @@ class Logging(GCPService):
                     f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                 )
 
+    def _get_org_sinks(self):
+        """Fetch org-level sinks with includeChildren so child projects are not falsely failed."""
+        org_ids = set()
+        for project in self.projects.values():
+            if project.organization:
+                org_ids.add(project.organization.id)
+
+        for org_id in org_ids:
+            try:
+                request = self.client.sinks().list(parent=f"organizations/{org_id}")
+                while request is not None:
+                    response = request.execute(num_retries=DEFAULT_RETRY_ATTEMPTS)
+
+                    for sink in response.get("sinks", []):
+                        self.sinks.append(
+                            Sink(
+                                name=sink["name"],
+                                destination=sink["destination"],
+                                filter=sink.get("filter", "all"),
+                                project_id=f"organizations/{org_id}",
+                                include_children=sink.get("includeChildren", False),
+                            )
+                        )
+
+                    request = self.client.sinks().list_next(
+                        previous_request=request, previous_response=response
+                    )
+            except Exception as error:
+                logger.error(
+                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+
     def _get_metrics(self):
         for project_id in self.project_ids:
             try:
@@ -76,6 +109,7 @@ class Sink(BaseModel):
     destination: str
     filter: str
     project_id: str
+    include_children: bool = False
 
 
 class Metric(BaseModel):

prowler/providers/gcp/services/logging/logging_sink_created/logging_sink_created.py

@@ -5,26 +5,30 @@ from prowler.providers.gcp.services.logging.logging_client import logging_client
 class logging_sink_created(Check):
     def execute(self) -> Check_Report_GCP:
         findings = []
+
+        # Map project_id -> sink for direct project-level sinks
         projects_with_logging_sink = {}
         for sink in logging_client.sinks:
-            if sink.filter == "all":
+            if sink.filter == "all" and not sink.include_children:
                 projects_with_logging_sink[sink.project_id] = sink
 
+        # Collect org resource names that have a covering sink (includeChildren=True)
+        covering_org_sinks = {}
+        for sink in logging_client.sinks:
+            if sink.filter == "all" and sink.include_children:
+                covering_org_sinks[sink.project_id] = sink
+
         for project in logging_client.project_ids:
-            if project not in projects_with_logging_sink.keys():
-                project_obj = logging_client.projects.get(project)
-                report = Check_Report_GCP(
-                    metadata=self.metadata(),
-                    resource=project_obj,
-                    resource_id=project,
-                    project_id=project,
-                    location=logging_client.region,
-                    resource_name=(getattr(project_obj, "name", None) or "GCP Project"),
-                )
-                report.status = "FAIL"
-                report.status_extended = f"There are no logging sinks to export copies of all the log entries in project {project}."
-                findings.append(report)
-            else:
+            project_obj = logging_client.projects.get(project)
+
+            # Determine whether this project is covered by an org-level sink
+            org = getattr(project_obj, "organization", None) if project_obj else None
+            org_resource = f"organizations/{org.id}" if org else None
+            covering_sink = (
+                covering_org_sinks.get(org_resource) if org_resource else None
+            )
+
+            if project in projects_with_logging_sink:
                 sink = projects_with_logging_sink[project]
                 sink_name = getattr(sink, "name", None) or "unknown"
                 report = Check_Report_GCP(
@@ -40,4 +44,31 @@ class logging_sink_created(Check):
                 report.status = "PASS"
                 report.status_extended = f"Sink {sink_name} is enabled exporting copies of all the log entries in project {project}."
                 findings.append(report)
+            elif covering_sink:
+                sink_name = getattr(covering_sink, "name", None) or "unknown"
+                report = Check_Report_GCP(
+                    metadata=self.metadata(),
+                    resource=covering_sink,
+                    resource_id=sink_name,
+                    project_id=project,
+                    location=logging_client.region,
+                    resource_name=(
+                        sink_name if sink_name != "unknown" else "Logging Sink"
+                    ),
+                )
+                report.status = "PASS"
+                report.status_extended = f"Sink {sink_name} at organization level is exporting copies of all the log entries in project {project}."
+                findings.append(report)
+            else:
+                report = Check_Report_GCP(
+                    metadata=self.metadata(),
+                    resource=project_obj,
+                    resource_id=project,
+                    project_id=project,
+                    location=logging_client.region,
+                    resource_name=(getattr(project_obj, "name", None) or "GCP Project"),
+                )
+                report.status = "FAIL"
+                report.status_extended = f"There are no logging sinks to export copies of all the log entries in project {project}."
+                findings.append(report)
         return findings

prowler/providers/okta/lib/arguments/arguments.py

@@ -35,7 +35,8 @@ def init_parser(self):
         nargs="+",
         help=(
             "OAuth scopes to request, space-separated "
-            "(e.g. okta.policies.read okta.brands.read okta.apps.read). "
+            "(e.g. okta.policies.read okta.brands.read okta.apps.read "
+            "okta.logStreams.read okta.idps.read). "
             "Defaults to the read scopes required by the bundled checks."
         ),
         default=None,

prowler/providers/okta/lib/service/pagination.py

@@ -0,0 +1,69 @@
+"""Shared pagination helpers for Okta SDK list calls.
+
+The Okta SDK exposes paginated list endpoints (`list_applications`,
+`list_policies`, `list_log_streams`, `list_identity_providers`, …) that
+return a tuple `(items, response, error)`. The next page is signalled
+through an RFC 5988 `Link: <…>; rel="next"` header carrying an opaque
+`after` cursor.
+
+These helpers are used by every Okta service that needs to drain a
+paginated endpoint. They live here so we don't keep copy-pasting them
+into each service module.
+"""
+
+from typing import Optional
+from urllib.parse import parse_qs, urlparse
+
+
+def next_after_cursor(resp) -> Optional[str]:
+    """Extract the `after` cursor from a `Link: ...; rel="next"` header.
+
+    Returns None when there is no next page. Header format follows RFC
+    5988 and Okta's pagination guide.
+    """
+    if resp is None:
+        return None
+    headers = getattr(resp, "headers", None) or {}
+    link = headers.get("link") or headers.get("Link") or ""
+    if not link:
+        return None
+    for part in link.split(","):
+        if 'rel="next"' not in part:
+            continue
+        url_segment = part.split(";", 1)[0].strip().lstrip("<").rstrip(">")
+        cursor = parse_qs(urlparse(url_segment).query).get("after", [None])[0]
+        if cursor:
+            return cursor
+    return None
+
+
+async def paginate(fetch):
+    """Drain all pages of an SDK list call.
+
+    `fetch` is a callable that accepts the `after` cursor (or None for
+    the first page) and returns the SDK's standard `(items, resp, err)`
+    tuple — or the 2-tuple early-error shape `(items, err)`. Follows the
+    `Link: rel="next"` header until exhausted. The returned tuple is
+    `(all_items, error)` — error is non-None only when a page fails
+    to fetch.
+    """
+    all_items = []
+    result = await fetch(None)
+    err = result[-1]
+    if err is not None:
+        return [], err
+    items = result[0]
+    resp = result[1] if len(result) >= 3 else None
+    all_items.extend(items or [])
+    while True:
+        cursor = next_after_cursor(resp)
+        if not cursor:
+            break
+        result = await fetch(cursor)
+        err = result[-1]
+        if err is not None:
+            return all_items, err
+        items = result[0]
+        resp = result[1] if len(result) >= 3 else None
+        all_items.extend(items or [])
+    return all_items, None

prowler/providers/okta/lib/service/raw_fetch.py

@@ -0,0 +1,141 @@
+"""Raw-JSON HTTP fetch via the Okta SDK's request executor.
+
+Some Okta Management API endpoints are not yet exposed as typed methods
+on the SDK client (e.g. `/api/v1/automations`), or the typed path's
+pydantic deserialization rejects values the API actually returns (e.g.
+the `KnowledgeConstraint.types` lowercase issue we hit on
+`list_policy_rules`). In both cases we go around the typed layer:
+construct the request via `client._request_executor.create_request`,
+execute without a response type, and parse the body ourselves.
+
+`get_json` returns the parsed JSON payload (typically a list or dict)
+or raises with a descriptive log line on any of the failure modes —
+request build, transport, decode, parse. `get_json_paginated` drains
+list endpoints by following the `Link: rel="next"` cursor — without it,
+the raw fallback would silently truncate at the per-request `limit`.
+Callers are expected to project the JSON onto their own pydantic snapshot.
+"""
+
+import json
+from typing import Any, Optional
+from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
+
+from prowler.lib.logger import logger
+from prowler.providers.okta.lib.service.pagination import next_after_cursor
+
+
+async def get_json(
+    client,
+    path: str,
+    *,
+    accept: str = "application/json",
+    context: Optional[str] = None,
+) -> Optional[Any]:
+    """GET `path` via the SDK's request executor and return parsed JSON.
+
+    Returns the decoded JSON payload on success, or None when the
+    request, transport, or decode steps fail. Each failure path emits a
+    `logger.error` line tagged with `context` so the caller can grep
+    for it.
+    """
+    label = context or path
+    request, error = await client._request_executor.create_request(
+        method="GET",
+        url=path,
+        body=None,
+        headers={"Accept": accept},
+    )
+    if error is not None:
+        logger.error(f"Raw fetch (create_request) failed for {label}: {error}")
+        return None
+
+    _response, response_body, error = await client._request_executor.execute(request)
+    if error is not None:
+        logger.error(f"Raw fetch (execute) failed for {label}: {error}")
+        return None
+
+    if isinstance(response_body, (bytes, bytearray)):
+        try:
+            response_body = response_body.decode("utf-8")
+        except UnicodeDecodeError as decode_err:
+            logger.error(f"Could not decode response for {label}: {decode_err}")
+            return None
+    try:
+        return json.loads(response_body) if response_body else None
+    except json.JSONDecodeError as decode_err:
+        logger.error(f"Could not parse JSON for {label}: {decode_err}")
+        return None
+
+
+async def get_json_paginated(
+    client,
+    path: str,
+    *,
+    page_size: int = 200,
+    accept: str = "application/json",
+    context: Optional[str] = None,
+) -> Optional[list]:
+    """Drain all pages of a raw-JSON list endpoint.
+
+    Mirrors the typed `pagination.paginate` shape but operates on the
+    SDK's request executor directly. Follows the `Link: rel="next"`
+    header until exhausted, accumulating items across pages. Returns
+    the concatenated list, or None if any page fails to fetch or the
+    response is not a JSON array.
+
+    `page_size` is appended as `limit=N` to the first request; subsequent
+    requests use the URL Okta returns via the cursor.
+    """
+    label = context or path
+    all_items: list = []
+    current_path = _set_query(path, {"limit": str(page_size)})
+    while True:
+        request, error = await client._request_executor.create_request(
+            method="GET",
+            url=current_path,
+            body=None,
+            headers={"Accept": accept},
+        )
+        if error is not None:
+            logger.error(f"Raw fetch (create_request) failed for {label}: {error}")
+            return None
+
+        response, response_body, error = await client._request_executor.execute(request)
+        if error is not None:
+            logger.error(f"Raw fetch (execute) failed for {label}: {error}")
+            return None
+
+        if isinstance(response_body, (bytes, bytearray)):
+            try:
+                response_body = response_body.decode("utf-8")
+            except UnicodeDecodeError as decode_err:
+                logger.error(f"Could not decode response for {label}: {decode_err}")
+                return None
+        if not response_body:
+            break
+        try:
+            page = json.loads(response_body)
+        except json.JSONDecodeError as decode_err:
+            logger.error(f"Could not parse JSON for {label}: {decode_err}")
+            return None
+        if not isinstance(page, list):
+            logger.error(
+                f"Unexpected raw payload shape for {label}: "
+                f"{type(page).__name__}; expected list"
+            )
+            return None
+        all_items.extend(page)
+
+        cursor = next_after_cursor(response)
+        if not cursor:
+            break
+        current_path = _set_query(path, {"limit": str(page_size), "after": cursor})
+    return all_items
+
+
+def _set_query(path: str, params: dict) -> str:
+    """Return `path` with the given query params merged in (overriding existing)."""
+    parsed = urlparse(path)
+    qs = dict(parse_qsl(parsed.query))
+    qs.update({k: v for k, v in params.items() if v is not None})
+    return urlunparse(parsed._replace(query=urlencode(qs)))

prowler/providers/okta/okta_provider.py

@@ -32,7 +32,13 @@ from prowler.providers.okta.exceptions.exceptions import (
 from prowler.providers.okta.lib.mutelist.mutelist import OktaMutelist
 from prowler.providers.okta.models import OktaIdentityInfo, OktaSession
 
-DEFAULT_SCOPES = ["okta.policies.read", "okta.brands.read", "okta.apps.read"]
+DEFAULT_SCOPES = [
+    "okta.policies.read",
+    "okta.brands.read",
+    "okta.apps.read",
+    "okta.logStreams.read",
+    "okta.idps.read",
+]
 # Accept only Okta-managed domains. Custom (vanity) domains are rejected on
 # purpose — they're a recurring source of typos and silent misconfig and
 # Prowler's audience overwhelmingly uses Okta-managed hosts. The TLDs below

prowler/providers/okta/services/application/application_service.py

@@ -1,10 +1,13 @@
-import json
 from typing import Optional
-from urllib.parse import parse_qs, urlparse
+from urllib.parse import urlparse
 
 from pydantic import BaseModel, ValidationError
 
 from prowler.lib.logger import logger
+from prowler.providers.okta.lib.service.pagination import paginate as _paginate_shared
+from prowler.providers.okta.lib.service.raw_fetch import (
+    get_json_paginated as _raw_get_json_paginated,
+)
 from prowler.providers.okta.lib.service.service import OktaService
 
 # These three keys are Okta-platform constants, not tenant-configurable:
@@ -28,29 +31,6 @@ DASHBOARD_APP_NAME = "okta_enduser"
 ADMIN_CONSOLE_FIRST_PARTY_APP_KEY = "admin-console"
 
 
-def _next_after_cursor(resp) -> Optional[str]:
-    """Extract the `after` cursor from a `Link: ...; rel="next"` header.
-
-    Returns None when there is no next page. Header format follows RFC 5988
-    and Okta's pagination guide. Mirrors the helper in `signon_service` —
-    duplicated rather than shared until a third Okta service appears.
-    """
-    if resp is None:
-        return None
-    headers = getattr(resp, "headers", None) or {}
-    link = headers.get("link") or headers.get("Link") or ""
-    if not link:
-        return None
-    for part in link.split(","):
-        if 'rel="next"' not in part:
-            continue
-        url_segment = part.split(";", 1)[0].strip().lstrip("<").rstrip(">")
-        cursor = parse_qs(urlparse(url_segment).query).get("after", [None])[0]
-        if cursor:
-            return cursor
-    return None
-
-
 REQUIRED_SCOPES: dict[str, str] = {
     "admin_console_app_settings": "okta.apps.read",
     "built_in_apps": "okta.apps.read",
@@ -321,69 +301,24 @@ class Application(OktaService):
         """Raw-JSON fallback for `list_policy_rules`.
 
         Bypasses the Okta SDK's typed deserialization by calling the
-        request executor directly without a response type. The response
-        body is then `json.loads`-ed and projected onto our own pydantic
-        snapshot, which only validates the fields the STIG checks
-        actually read. This keeps the checks evaluable on tenants where
-        the Management API returns values the SDK validators reject.
+        request executor directly via the shared `get_json_paginated`
+        helper, which follows `Link: rel=next` so policies with more
+        rules than `rule_fetch_limit` are not silently truncated.
+        Projects the response onto our own pydantic snapshot which only
+        validates the fields the STIG checks actually read. This keeps
+        the checks evaluable on tenants where the Management API returns
+        values the SDK validators reject.
         """
-        request, error = await self.client._request_executor.create_request(
-            method="GET",
-            url=f"/api/v1/policies/{policy_id}/rules?limit={rule_fetch_limit}",
-            body=None,
-            headers={"Accept": "application/json"},
+        rules_data = await _raw_get_json_paginated(
+            self.client,
+            f"/api/v1/policies/{policy_id}/rules",
+            page_size=rule_fetch_limit,
+            context=f"access policy {policy_id} rules",
         )
-        if error is not None:
-            logger.error(
-                f"Raw rules fetch (create_request) failed for {policy_id}: {error}"
-            )
+        if rules_data is None:
             return AuthenticationPolicy(
                 id=policy_id, name="", status="", is_default=False, rules=[]
             )
-
-        _response, response_body, error = await self.client._request_executor.execute(
-            request
-        )
-        if error is not None:
-            logger.error(f"Raw rules fetch (execute) failed for {policy_id}: {error}")
-            return AuthenticationPolicy(
-                id=policy_id, name="", status="", is_default=False, rules=[]
-            )
-
-        if isinstance(response_body, (bytes, bytearray)):
-            try:
-                response_body = response_body.decode("utf-8")
-            except UnicodeDecodeError as decode_err:
-                logger.error(
-                    f"Could not decode rules response for {policy_id}: {decode_err}"
-                )
-                return AuthenticationPolicy(
-                    id=policy_id, name="", status="", is_default=False, rules=[]
-                )
-        try:
-            rules_data = json.loads(response_body) if response_body else []
-        except json.JSONDecodeError as decode_err:
-            logger.error(f"Could not parse rules JSON for {policy_id}: {decode_err}")
-            return AuthenticationPolicy(
-                id=policy_id, name="", status="", is_default=False, rules=[]
-            )
-
-        if not isinstance(rules_data, list):
-            logger.error(
-                f"Unexpected raw rules payload shape for {policy_id}: "
-                f"got {type(rules_data).__name__}, expected list"
-            )
-            return AuthenticationPolicy(
-                id=policy_id, name="", status="", is_default=False, rules=[]
-            )
-
-        if len(rules_data) >= rule_fetch_limit:
-            logger.warning(
-                f"Access policy {policy_id} returned {len(rules_data)} rules "
-                f"via raw-JSON fallback — the per-policy fetch limit "
-                f"({rule_fetch_limit}) was hit; any rules beyond this limit "
-                "are not evaluated by Prowler."
-            )
         rules_out = [_raw_rule_to_model(rule) for rule in rules_data]
         return AuthenticationPolicy(
             id=policy_id, name="", status="", is_default=False, rules=rules_out
@@ -391,33 +326,7 @@ class Application(OktaService):
 
     @staticmethod
     async def _paginate(fetch):
-        """Drain all pages of an SDK list call.
-
-        `fetch` is a callable taking the `after` cursor (or None) and
-        returning the SDK's `(items, resp, err)` tuple. Follows the
-        `Link: rel="next"` header until exhausted. Mirrors the helper in
-        `signon_service`.
-        """
-        all_items = []
-        result = await fetch(None)
-        err = result[-1]
-        if err is not None:
-            return [], err
-        items = result[0]
-        resp = result[1] if len(result) >= 3 else None
-        all_items.extend(items or [])
-        while True:
-            cursor = _next_after_cursor(resp)
-            if not cursor:
-                break
-            result = await fetch(cursor)
-            err = result[-1]
-            if err is not None:
-                return all_items, err
-            items = result[0]
-            resp = result[1] if len(result) >= 3 else None
-            all_items.extend(items or [])
-        return all_items, None
+        return await _paginate_shared(fetch)
 
 
 def _policy_id_from_href(href: Optional[str]) -> Optional[str]:

prowler/providers/okta/services/idp/idp_client.py

@@ -0,0 +1,4 @@
+from prowler.providers.common.provider import Provider
+from prowler.providers.okta.services.idp.idp_service import Idp
+
+idp_client = Idp(Provider.get_global_provider())

prowler/providers/okta/services/idp/idp_service.py

@@ -0,0 +1,118 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+from prowler.lib.logger import logger
+from prowler.providers.okta.lib.service.pagination import paginate
+from prowler.providers.okta.lib.service.service import OktaService
+
+# Okta's API value for the "Smart Card" IdP shown in the Admin Console.
+# The UI label is "Smart Card IdP" but the `type` field on the API response
+# is `X509` (Mutual TLS) — that is the value we filter on.
+SMART_CARD_IDP_TYPE = "X509"
+
+REQUIRED_SCOPES: dict[str, str] = {
+    "identity_providers": "okta.idps.read",
+}
+
+
+class Idp(OktaService):
+    """Fetches Okta Identity Providers.
+
+    Populates `self.identity_providers` keyed by IdP id. Each entry
+    captures the minimum fields the bundled checks read: identity
+    (`id`, `name`), `type`, `status`, and — for `X509` Smart Card IdPs
+    — the certificate-chain `issuer` and `kid` exposed by Okta's
+    `protocol.credentials.trust` structure. Reading the issuer DN lets
+    the check surface it for out-of-band verification against the
+    DOD-approved CA list.
+
+    Required OAuth scopes (`REQUIRED_SCOPES`) are compared against the
+    access token's granted scopes (`provider.identity.granted_scopes`).
+    Missing scopes are recorded in `self.missing_scope` so the check
+    can emit an explicit MANUAL finding.
+    """
+
+    def __init__(self, provider):
+        super().__init__(__class__.__name__, provider)
+        granted = set(getattr(provider.identity, "granted_scopes", None) or [])
+        self.missing_scope: dict[str, Optional[str]] = {
+            resource: (scope if granted and scope not in granted else None)
+            for resource, scope in REQUIRED_SCOPES.items()
+        }
+
+        self.identity_providers: dict[str, OktaIdentityProvider] = (
+            {}
+            if self.missing_scope["identity_providers"]
+            else self._list_identity_providers()
+        )
+
+    def _list_identity_providers(self) -> dict:
+        logger.info("Idp - Listing Okta Identity Providers...")
+        try:
+            return self._run(self._fetch_identity_providers())
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return {}
+
+    async def _fetch_identity_providers(self) -> dict:
+        result: dict[str, OktaIdentityProvider] = {}
+        all_idps, err = await paginate(
+            lambda after: self.client.list_identity_providers(after=after)
+        )
+        if err is not None:
+            logger.error(f"Error listing identity providers: {err}")
+            return result
+
+        for idp in all_idps:
+            idp_id = getattr(idp, "id", "") or ""
+            if not idp_id:
+                continue
+            issuer, kid = _trust_fields(idp)
+            result[idp_id] = OktaIdentityProvider(
+                id=idp_id,
+                name=getattr(idp, "name", "") or "",
+                type=_stringify_enum(getattr(idp, "type", None)) or "",
+                status=_stringify_enum(getattr(idp, "status", None)) or "",
+                trust_issuer=issuer,
+                trust_kid=kid,
+            )
+        return result
+
+
+def _trust_fields(idp) -> tuple[Optional[str], Optional[str]]:
+    """Extract `issuer` and `kid` from an `X509` IdP's protocol.credentials.trust.
+
+    The SDK exposes `IdentityProvider.protocol` as `IdentityProviderProtocol`,
+    a Pydantic v2 oneOf wrapper that holds the concrete protocol (ProtocolMtls
+    for X509 IdPs) on `actual_instance`. `credentials` is not proxied on the
+    wrapper, so reading it directly returns None — we have to unwrap first.
+    """
+    protocol = getattr(idp, "protocol", None)
+    if protocol is None:
+        return None, None
+    actual_protocol = getattr(protocol, "actual_instance", None) or protocol
+    credentials = getattr(actual_protocol, "credentials", None)
+    if credentials is None:
+        return None, None
+    trust = getattr(credentials, "trust", None)
+    if trust is None:
+        return None, None
+    return getattr(trust, "issuer", None), getattr(trust, "kid", None)
+
+
+def _stringify_enum(value) -> Optional[str]:
+    if value is None:
+        return None
+    return getattr(value, "value", None) or str(value)
+
+
+class OktaIdentityProvider(BaseModel):
+    id: str
+    name: str = ""
+    type: str = ""
+    status: str = ""
+    trust_issuer: Optional[str] = None
+    trust_kid: Optional[str] = None

prowler/providers/okta/services/idp/idp_smart_card_dod_approved_ca/idp_smart_card_dod_approved_ca.metadata.json

@@ -0,0 +1,37 @@
+{
+  "Provider": "okta",
+  "CheckID": "idp_smart_card_dod_approved_ca",
+  "CheckTitle": "Okta Smart Card (X509) Identity Provider uses a DOD-approved certificate authority",
+  "CheckType": [],
+  "ServiceName": "idp",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "NotDefined",
+  "ResourceGroup": "IAM",
+  "Description": "Every Okta Smart Card (X509) Identity Provider must be `ACTIVE` and its certificate chain must be issued by a DOD-approved CA. The check ships default issuer-DN patterns covering DOD PKI and ECA, and matches them against the chain's `issuer`. Override or extend via `okta_dod_approved_ca_issuer_patterns` in the audit config to recognise tenant-specific DOD CAs.",
+  "Risk": "An Okta Smart Card IdP whose certificate chain is not issued by a DOD-approved CA can be used to authenticate non-vetted identities.\n\n- **Trust on an unverified CA** allows impersonation of CAC/PIV holders\n- **Bypass of the federal PKI** required for DOD-grade identity assurance\n- **Acceptance of certificates** from a private or unaccredited issuer",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://help.okta.com/en-us/content/topics/security/idp-enable-smart-card.htm",
+    "https://developer.okta.com/docs/api/openapi/okta-management/management/tag/IdentityProvider/"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "1. Sign in to the **Okta Admin Console** as a *Super Admin*.\n2. Navigate to **Security** > **Identity Providers**.\n3. For each IdP whose **Type** is **Smart Card**, click **Actions** > **Configure**.\n4. Under **Certificate chain**, verify the certificate is from a DOD-approved Certificate Authority (DOD PKI, ECA, JITC, or equivalent).\n5. If the IdP is not **Active**, activate it once the chain is validated.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Verify each Okta Smart Card (X509) Identity Provider is ACTIVE and its certificate chain is issued by a DOD-approved Certificate Authority. Document the issuer for audit evidence.",
+      "Url": "https://hub.prowler.com/check/idp_smart_card_dod_approved_ca"
+    }
+  },
+  "Categories": [
+    "identity-access"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Aligns with DISA STIG V-273207 / OKTA-APP-001920."
+}

prowler/providers/okta/services/idp/idp_smart_card_dod_approved_ca/idp_smart_card_dod_approved_ca.py

@@ -0,0 +1,148 @@
+import re
+
+from prowler.lib.check.models import Check, CheckReportOkta
+from prowler.providers.okta.services.idp.idp_client import idp_client
+from prowler.providers.okta.services.idp.idp_service import (
+    SMART_CARD_IDP_TYPE,
+    OktaIdentityProvider,
+)
+from prowler.providers.okta.services.idp.lib.idp_helpers import (
+    missing_idps_scope_finding,
+)
+
+# Default issuer-DN substring patterns recognised as DOD-approved Certificate
+# Authorities. The DOD PKI publishes canonical DN forms that include
+# `O=U.S. Government, OU=DoD` (for DoD Root, DoD ID, DoD EMAIL, DoD SW, DoD
+# JITC CAs) and `O=U.S. Government, OU=ECA` for the External Certificate
+# Authorities. Customers running an internal CA outside these patterns can
+# extend the list via the `okta_dod_approved_ca_issuer_patterns` audit-config
+# entry — see the per-check Notes in metadata.json.
+DEFAULT_DOD_CA_ISSUER_PATTERNS = (
+    # `OU=DoD` is the distinctive DISA DN component for every CA in the DoD
+    # PKI (Root, ID, EMAIL, SW, JITC). `OU=ECA` is the equivalent for the
+    # External Certificate Authorities. The trailing `\b` prevents accidental
+    # matches against superstrings like `OU=DoDExtra`.
+    r"\bOU=DoD\b",
+    r"\bOU=ECA\b",
+)
+
+
+class idp_smart_card_dod_approved_ca(Check):
+    """Verifies that Okta Smart Card (X509) IdPs are configured and use a DOD-approved CA.
+
+    PASS when the IdP is `ACTIVE` and its certificate chain's `issuer`
+    DN matches one of the configured DOD-approved CA patterns. MANUAL
+    when active but the issuer doesn't match (operator can verify
+    out-of-band or extend the pattern list). FAIL when no Smart Card
+    IdP is configured or when the configured IdP is inactive.
+    """
+
+    def execute(self) -> list[CheckReportOkta]:
+        findings: list[CheckReportOkta] = []
+        org_domain = idp_client.provider.identity.org_domain
+        audit_config = idp_client.audit_config or {}
+        configured_patterns = audit_config.get("okta_dod_approved_ca_issuer_patterns")
+        patterns = (
+            tuple(configured_patterns)
+            if configured_patterns
+            else DEFAULT_DOD_CA_ISSUER_PATTERNS
+        )
+
+        missing_scope = idp_client.missing_scope.get("identity_providers")
+        if missing_scope:
+            findings.append(
+                missing_idps_scope_finding(self.metadata(), org_domain, missing_scope)
+            )
+            return findings
+
+        smart_card_idps = [
+            idp
+            for idp in idp_client.identity_providers.values()
+            if (idp.type or "").upper() == SMART_CARD_IDP_TYPE
+        ]
+
+        if not smart_card_idps:
+            placeholder = OktaIdentityProvider(
+                id="okta-smart-card-idp-missing",
+                name="(no Smart Card IdP configured)",
+                type=SMART_CARD_IDP_TYPE,
+                status="MISSING",
+            )
+            report = CheckReportOkta(
+                metadata=self.metadata(), resource=placeholder, org_domain=org_domain
+            )
+            report.status = "FAIL"
+            report.status_extended = (
+                "No Smart Card (X509) Identity Providers are configured. "
+                "Configure a Smart Card IdP in the Admin Console "
+                "(Security > Identity Providers) with a certificate chain "
+                "issued by a DOD-approved CA. If CAC/PIV authentication is "
+                "not required for this tenant, mutelist this check with "
+                "that documented exception."
+            )
+            findings.append(report)
+            return findings
+
+        for idp in smart_card_idps:
+            report = CheckReportOkta(
+                metadata=self.metadata(), resource=idp, org_domain=org_domain
+            )
+            label = f"Okta Smart Card IdP '{idp.name}' (id={idp.id}, type={idp.type})"
+            chain_detail = _format_chain_detail(idp)
+
+            if (idp.status or "").upper() != "ACTIVE":
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"{label} is not ACTIVE (status={idp.status or 'unset'}). "
+                    "Activate the IdP from Security > Identity Providers, then "
+                    f"verify the certificate chain. {chain_detail}"
+                )
+                findings.append(report)
+                continue
+
+            matched_pattern = _matched_issuer_pattern(idp.trust_issuer, patterns)
+            if matched_pattern is not None:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"{label} is ACTIVE and its chain issuer matches a "
+                    f"DOD-approved CA pattern (`{matched_pattern}`). "
+                    f"{chain_detail}"
+                )
+            else:
+                report.status = "MANUAL"
+                report.status_extended = (
+                    f"{label} is ACTIVE but its chain issuer does not match any "
+                    "configured DOD-approved CA pattern. Verify out-of-band "
+                    "that the certificate chain belongs to a DOD-approved "
+                    "Certificate Authority, or extend "
+                    "`okta_dod_approved_ca_issuer_patterns` in the audit "
+                    f"config. {chain_detail}"
+                )
+            findings.append(report)
+        return findings
+
+
+def _matched_issuer_pattern(issuer, patterns):
+    if not issuer:
+        return None
+    for pattern in patterns:
+        try:
+            if re.search(pattern, issuer):
+                return pattern
+        except re.error:
+            # Skip malformed operator-supplied patterns rather than crashing
+            # the whole check.
+            continue
+    return None
+
+
+def _format_chain_detail(idp: OktaIdentityProvider) -> str:
+    if idp.trust_issuer or idp.trust_kid:
+        return (
+            f"Chain issuer: {idp.trust_issuer or 'unset'}; "
+            f"kid: {idp.trust_kid or 'unset'}."
+        )
+    return (
+        "Chain issuer and kid were not exposed by the API; inspect the IdP in "
+        "the Admin Console under Security > Identity Providers > Configure."
+    )

prowler/providers/okta/services/idp/lib/idp_helpers.py

@@ -0,0 +1,26 @@
+"""Shared helpers for the OKTA idp STIG checks."""
+
+from prowler.lib.check.models import CheckReportOkta
+from prowler.providers.okta.services.idp.idp_service import OktaIdentityProvider
+
+
+def missing_idps_scope_finding(
+    metadata, org_domain: str, scope: str
+) -> CheckReportOkta:
+    """Build the MANUAL finding when the IdPs scope is not granted."""
+    placeholder = OktaIdentityProvider(
+        id="okta-idps-scope-missing",
+        name="(scope not granted)",
+        status="MISSING",
+    )
+    report = CheckReportOkta(
+        metadata=metadata, resource=placeholder, org_domain=org_domain
+    )
+    report.status = "MANUAL"
+    report.status_extended = (
+        "Could not retrieve Okta Identity Providers: the Okta service app is "
+        f"missing the required `{scope}` API scope. Grant it on the service "
+        "app's Okta API Scopes tab in the Okta Admin Console, then re-run the "
+        "check."
+    )
+    return report

prowler/providers/okta/services/signon/signon_service.py

@@ -1,34 +1,11 @@
 from typing import Optional
-from urllib.parse import parse_qs, urlparse
 
 from pydantic import BaseModel
 
 from prowler.lib.logger import logger
+from prowler.providers.okta.lib.service.pagination import paginate as _paginate_shared
 from prowler.providers.okta.lib.service.service import OktaService
 
-
-def _next_after_cursor(resp) -> Optional[str]:
-    """Extract the `after` cursor from a `Link: ...; rel="next"` header.
-
-    Returns None when there is no next page. Header format follows RFC 5988
-    and Okta's pagination guide.
-    """
-    if resp is None:
-        return None
-    headers = getattr(resp, "headers", None) or {}
-    link = headers.get("link") or headers.get("Link") or ""
-    if not link:
-        return None
-    for part in link.split(","):
-        if 'rel="next"' not in part:
-            continue
-        url_segment = part.split(";", 1)[0].strip().lstrip("<").rstrip(">")
-        cursor = parse_qs(urlparse(url_segment).query).get("after", [None])[0]
-        if cursor:
-            return cursor
-    return None
-
-
 REQUIRED_SCOPES: dict[str, str] = {
     "global_session_policies": "okta.policies.read",
     "sign_in_pages": "okta.brands.read",
@@ -228,33 +205,7 @@ class Signon(OktaService):
 
     @staticmethod
     async def _paginate(fetch):
-        """Drain all pages of an SDK list call.
-
-        `fetch` is a callable that takes the `after` cursor (or None for
-        the first page) and returns the SDK's standard `(items, resp, err)`
-        tuple. We follow `Link: rel="next"` headers until exhausted.
-        """
-        all_items = []
-        result = await fetch(None)
-        # Defensive against the SDK's 2-tuple early-error path: error is last.
-        err = result[-1]
-        if err is not None:
-            return [], err
-        items = result[0]
-        resp = result[1] if len(result) >= 3 else None
-        all_items.extend(items or [])
-        while True:
-            cursor = _next_after_cursor(resp)
-            if not cursor:
-                break
-            result = await fetch(cursor)
-            err = result[-1]
-            if err is not None:
-                return all_items, err
-            items = result[0]
-            resp = result[1] if len(result) >= 3 else None
-            all_items.extend(items or [])
-        return all_items, None
+        return await _paginate_shared(fetch)
 
 
 class GlobalSessionPolicyRule(BaseModel):

prowler/providers/okta/services/systemlog/lib/systemlog_helpers.py

@@ -0,0 +1,26 @@
+"""Shared helpers for the OKTA systemlog STIG checks."""
+
+from prowler.lib.check.models import CheckReportOkta
+from prowler.providers.okta.services.systemlog.systemlog_service import LogStream
+
+
+def missing_log_streams_scope_finding(
+    metadata, org_domain: str, scope: str
+) -> CheckReportOkta:
+    """Build the MANUAL finding when the log-streams scope is not granted."""
+    placeholder = LogStream(
+        id="okta-log-streams-scope-missing",
+        name="(scope not granted)",
+        status="MISSING",
+        type="",
+    )
+    report = CheckReportOkta(
+        metadata=metadata, resource=placeholder, org_domain=org_domain
+    )
+    report.status = "MANUAL"
+    report.status_extended = (
+        "Could not retrieve Okta Log Streams: the Okta service app is missing "
+        f"the required `{scope}` API scope. Grant it on the service app's "
+        "Okta API Scopes tab in the Okta Admin Console, then re-run the check."
+    )
+    return report

prowler/providers/okta/services/systemlog/systemlog_client.py

@@ -0,0 +1,4 @@
+from prowler.providers.common.provider import Provider
+from prowler.providers.okta.services.systemlog.systemlog_service import SystemLog
+
+systemlog_client = SystemLog(Provider.get_global_provider())

prowler/providers/okta/services/systemlog/systemlog_service.py

@@ -0,0 +1,136 @@
+from typing import Optional
+
+from pydantic import BaseModel, ValidationError
+
+from prowler.lib.logger import logger
+from prowler.providers.okta.lib.service.pagination import paginate
+from prowler.providers.okta.lib.service.raw_fetch import (
+    get_json_paginated as raw_get_json_paginated,
+)
+from prowler.providers.okta.lib.service.service import OktaService
+
+REQUIRED_SCOPES: dict[str, str] = {
+    "log_streams": "okta.logStreams.read",
+}
+
+
+class SystemLog(OktaService):
+    """Fetches Okta Log Stream configurations.
+
+    Populates `self.log_streams` keyed by Log Stream id. Each entry
+    carries `name`, `status`, `type` — enough for the streaming-enabled
+    check to evaluate whether the tenant has off-loaded audit records
+    to an external SIEM/event bus.
+
+    Required OAuth scopes (`REQUIRED_SCOPES`) are compared against the
+    access token's granted scopes (`provider.identity.granted_scopes`).
+    Missing scopes are recorded in `self.missing_scope` so the check
+    can emit an explicit MANUAL finding instead of a misleading
+    "no resources returned".
+    """
+
+    def __init__(self, provider):
+        super().__init__(__class__.__name__, provider)
+        granted = set(getattr(provider.identity, "granted_scopes", None) or [])
+        self.missing_scope: dict[str, Optional[str]] = {
+            resource: (scope if granted and scope not in granted else None)
+            for resource, scope in REQUIRED_SCOPES.items()
+        }
+
+        self.log_streams: dict[str, LogStream] = (
+            {} if self.missing_scope["log_streams"] else self._list_log_streams()
+        )
+
+    def _list_log_streams(self) -> dict:
+        logger.info("SystemLog - Listing Okta Log Streams...")
+        try:
+            return self._run(self._fetch_log_streams())
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return {}
+
+    async def _fetch_log_streams(self) -> dict:
+        result: dict[str, LogStream] = {}
+        try:
+            all_streams, err = await paginate(
+                lambda after: self.client.list_log_streams(after=after)
+            )
+        except ValidationError as ve:
+            # Upstream okta-sdk-python bug: e.g. `LogStreamSettingsAws`'s
+            # `eventSourceName` validator regex is `^[a-zA-Z0-9.\-_]$` —
+            # missing the `+` quantifier, so it rejects every
+            # multi-character name. Fall back to raw JSON so the check
+            # can still evaluate the tenant's actual log-stream state.
+            # Remove this workaround once okta-sdk-python fixes the
+            # validator (issue to be filed upstream).
+            logger.warning(
+                f"Okta SDK raised ValidationError parsing log streams "
+                f"({ve.error_count()} error(s)) — falling back to raw-JSON "
+                "parse. This is an okta-sdk-python deserialization bug."
+            )
+            return await self._fetch_log_streams_raw()
+
+        if err is not None:
+            logger.error(f"Error listing log streams: {err}")
+            return result
+
+        for stream in all_streams:
+            stream_id = getattr(stream, "id", "") or ""
+            if not stream_id:
+                continue
+            result[stream_id] = LogStream(
+                id=stream_id,
+                name=getattr(stream, "name", "") or "",
+                status=getattr(stream, "status", "") or "",
+                type=_stringify_enum(getattr(stream, "type", None)) or "",
+            )
+        return result
+
+    async def _fetch_log_streams_raw(self) -> dict:
+        """Raw-JSON fallback for `list_log_streams`.
+
+        Bypasses the SDK's typed deserialization via the shared
+        `get_json_paginated` helper (which follows the `Link: rel=next`
+        cursor so tenants with >200 streams are not silently truncated),
+        and projects the response onto our own pydantic snapshot which
+        only validates the four fields the check reads. Keeps the check
+        evaluable on tenants whose Log Stream settings happen to trip
+        an SDK enum/regex validator.
+        """
+        result: dict[str, LogStream] = {}
+        data = await raw_get_json_paginated(
+            self.client,
+            "/api/v1/logStreams",
+            page_size=200,
+            context="log streams",
+        )
+        if data is None:
+            return result
+        for item in data:
+            if not isinstance(item, dict):
+                continue
+            stream_id = item.get("id")
+            if not stream_id:
+                continue
+            result[stream_id] = LogStream(
+                id=stream_id,
+                name=item.get("name") or "",
+                status=(item.get("status") or "").upper(),
+                type=item.get("type") or "",
+            )
+        return result
+
+
+def _stringify_enum(value) -> Optional[str]:
+    if value is None:
+        return None
+    return getattr(value, "value", None) or str(value)
+
+
+class LogStream(BaseModel):
+    id: str
+    name: str = ""
+    status: str = ""
+    type: str = ""

prowler/providers/okta/services/systemlog/systemlog_streaming_enabled/systemlog_streaming_enabled.metadata.json

@@ -0,0 +1,37 @@
+{
+  "Provider": "okta",
+  "CheckID": "systemlog_streaming_enabled",
+  "CheckTitle": "Okta off-loads audit records to a central log server via Log Streaming",
+  "CheckType": [],
+  "ServiceName": "systemlog",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "NotDefined",
+  "ResourceGroup": "monitoring",
+  "Description": "Okta must off-load audit records to a central log server. At least one **Log Stream** (AWS EventBridge, Splunk Cloud, etc.) must be configured and `ACTIVE` in the tenant. Alternatively, an external SIEM pulling the System Log API can satisfy the requirement, but that pull-based path is verified manually.",
+  "Risk": "Audit records stored only inside the Okta tenant are exposed to accidental or incidental deletion or alteration.\n\n- **No central retention** of authentication events for incident investigations\n- **Single point of failure** for the audit trail\n- **No correlation** with other identity, network, and endpoint telemetry in the SIEM",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://help.okta.com/en-us/content/topics/reports/log-streaming/about-log-streams.htm",
+    "https://developer.okta.com/docs/api/openapi/okta-management/management/tag/LogStream/"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "1. Sign in to the **Okta Admin Console** as a *Super Admin*.\n2. Navigate to **Reports** > **Log Streaming**.\n3. Click **Add Log Stream** and select **AWS EventBridge**, **Splunk Cloud**, or another supported destination.\n4. Complete the connection fields and save.\n5. Activate the stream and verify the destination receives events.\n6. If the destination SIEM is not natively supported, document the pull-based ingestion that uses the System Log API.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Configure at least one ACTIVE Okta Log Stream that off-loads audit records to a central SIEM (AWS EventBridge, Splunk Cloud, or another supported destination). Document any alternative pull-based ingestion via the System Log API.",
+      "Url": "https://hub.prowler.com/check/systemlog_streaming_enabled"
+    }
+  },
+  "Categories": [
+    "logging"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Aligns with DISA STIG V-273202 / OKTA-APP-001430."
+}

prowler/providers/okta/services/systemlog/systemlog_streaming_enabled/systemlog_streaming_enabled.py

@@ -0,0 +1,88 @@
+from prowler.lib.check.models import Check, CheckReportOkta
+from prowler.providers.okta.services.systemlog.lib.systemlog_helpers import (
+    missing_log_streams_scope_finding,
+)
+from prowler.providers.okta.services.systemlog.systemlog_client import systemlog_client
+from prowler.providers.okta.services.systemlog.systemlog_service import LogStream
+
+
+class systemlog_streaming_enabled(Check):
+    """Verifies that at least one Okta Log Stream is configured and active.
+
+    Off-loading audit records to a central SIEM (AWS EventBridge, Splunk
+    Cloud, etc.) is the standard mechanism for centralised retention.
+    An alternative path — pulling the System Log API into an external
+    SIEM — is allowed by the requirement, but cannot be verified
+    automatically; this check emits a MANUAL note in that case.
+    """
+
+    def execute(self) -> list[CheckReportOkta]:
+        findings: list[CheckReportOkta] = []
+        org_domain = systemlog_client.provider.identity.org_domain
+
+        missing_scope = systemlog_client.missing_scope.get("log_streams")
+        if missing_scope:
+            findings.append(
+                missing_log_streams_scope_finding(
+                    self.metadata(), org_domain, missing_scope
+                )
+            )
+            return findings
+
+        active_streams = [
+            stream
+            for stream in systemlog_client.log_streams.values()
+            if not stream.status or stream.status.upper() == "ACTIVE"
+        ]
+
+        if not systemlog_client.log_streams:
+            placeholder = LogStream(
+                id="okta-log-streams-missing",
+                name="(no Log Streams configured)",
+                status="MISSING",
+                type="",
+            )
+            report = CheckReportOkta(
+                metadata=self.metadata(), resource=placeholder, org_domain=org_domain
+            )
+            report.status = "FAIL"
+            report.status_extended = (
+                "No Okta Log Streams are configured. Configure a Log Stream "
+                "(Reports > Log Streaming) to off-load audit records to a "
+                "central SIEM. If an external SIEM is already pulling logs "
+                "via the System Log API, mutelist this check with that "
+                "evidence."
+            )
+            findings.append(report)
+            return findings
+
+        if not active_streams:
+            placeholder = LogStream(
+                id="okta-log-streams-inactive",
+                name="(no active Log Streams)",
+                status="INACTIVE",
+                type="",
+            )
+            report = CheckReportOkta(
+                metadata=self.metadata(), resource=placeholder, org_domain=org_domain
+            )
+            report.status = "FAIL"
+            report.status_extended = (
+                f"{len(systemlog_client.log_streams)} Okta Log Stream(s) are "
+                "configured but none are ACTIVE. Activate a Log Stream to "
+                "off-load audit records to a central SIEM."
+            )
+            findings.append(report)
+            return findings
+
+        for stream in active_streams:
+            report = CheckReportOkta(
+                metadata=self.metadata(), resource=stream, org_domain=org_domain
+            )
+            report.status = "PASS"
+            report.status_extended = (
+                f"Okta Log Stream '{stream.name}' (type={stream.type or 'unset'}) "
+                "is ACTIVE and off-loads audit records to a central SIEM."
+            )
+            findings.append(report)
+        return findings

prowler/providers/okta/services/user/lib/user_helpers.py

@@ -0,0 +1,26 @@
+"""Shared helpers for the OKTA user STIG checks."""
+
+from prowler.lib.check.models import CheckReportOkta
+from prowler.providers.okta.services.user.user_service import UserAutomation
+
+
+def missing_user_scope_finding(
+    metadata, org_domain: str, scope: str
+) -> CheckReportOkta:
+    """Build the MANUAL finding when an OAuth scope is not granted."""
+    placeholder = UserAutomation(
+        id="okta-user-scope-missing",
+        name="(scope not granted)",
+        status="MISSING",
+    )
+    report = CheckReportOkta(
+        metadata=metadata, resource=placeholder, org_domain=org_domain
+    )
+    report.status = "MANUAL"
+    report.status_extended = (
+        f"Could not retrieve Okta user lifecycle automations: the Okta service "
+        f"app is missing the required `{scope}` API scope. Grant it on the "
+        "service app's Okta API Scopes tab in the Okta Admin Console, then "
+        "re-run the check."
+    )
+    return report

prowler/providers/okta/services/user/user_client.py

@@ -0,0 +1,4 @@
+from prowler.providers.common.provider import Provider
+from prowler.providers.okta.services.user.user_service import User
+
+user_client = User(Provider.get_global_provider())

prowler/providers/okta/services/user/user_inactivity_automation_35d_enabled/user_inactivity_automation_35d_enabled.metadata.json

@@ -0,0 +1,36 @@
+{
+  "Provider": "okta",
+  "CheckID": "user_inactivity_automation_35d_enabled",
+  "CheckTitle": "Okta automation suspends or deactivates users after 35 days of inactivity",
+  "CheckType": [],
+  "ServiceName": "user",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "medium",
+  "ResourceType": "NotDefined",
+  "ResourceGroup": "IAM",
+  "Description": "An Okta **Workflows Automation** must disable inactive user accounts. The automation must be `ACTIVE`, on an `ACTIVE` schedule, evaluate `User Inactivity = 35 days` (or less), apply to a group covering every user, and trigger `Suspended` / `Deactivated` / `Deprovisioned`. Threshold override: `okta_user_inactivity_max_days`. N/A when user sourcing is delegated to Active Directory or LDAP.",
+  "Risk": "Inactive Okta accounts retained indefinitely give an attacker who exploits one undetected access to downstream applications.\n\n- **Account takeover via dormant identities** that no one is monitoring\n- **Lateral movement** through SSO sessions of forgotten users\n- **Stale entitlements** that survive role and policy reorganisations",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://help.okta.com/en-us/content/topics/automation-hooks/automations-main.htm"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "1. Sign in to the **Okta Admin Console** as a *Super Admin*.\n2. Navigate to **Workflow** > **Automations** and click **Add Automation**.\n3. Name the automation (e.g., `User Inactivity`).\n4. Add a condition: select **User Inactivity in Okta** and enter `35` days.\n5. Configure the schedule to run daily and activate it.\n6. Apply the automation to a group that covers every user — typically `Everyone`.\n7. Add an action: **Change User lifecycle state in Okta** and choose `Suspended` (or `Deactivated`/`Deprovisioned`).\n8. Activate the automation.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Create an active Okta Workflows automation that runs daily, evaluates `User Inactivity in Okta = 35 days`, applies to a group covering every user, and changes the user lifecycle state to Suspended/Deactivated. If user sourcing is delegated to Active Directory or LDAP, document that the connected directory enforces this requirement instead.",
+      "Url": "https://hub.prowler.com/check/user_inactivity_automation_35d_enabled"
+    }
+  },
+  "Categories": [
+    "identity-access"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "Aligns with DISA STIG V-273188 / OKTA-APP-000090."
+}

prowler/providers/okta/services/user/user_inactivity_automation_35d_enabled/user_inactivity_automation_35d_enabled.py

@@ -0,0 +1,204 @@
+from prowler.lib.check.models import Check, CheckReportOkta
+from prowler.providers.okta.services.user.lib.user_helpers import (
+    missing_user_scope_finding,
+)
+from prowler.providers.okta.services.user.user_client import user_client
+from prowler.providers.okta.services.user.user_service import UserAutomation
+
+DEFAULT_INACTIVITY_DAYS = 35
+SUSPENSION_LIFECYCLE_ACTIONS = {"SUSPENDED", "DEACTIVATED", "DEPROVISIONED"}
+
+
+class user_inactivity_automation_35d_enabled(Check):
+    """Verifies that Okta suspends/deactivates users after 35 days of inactivity.
+
+    A Workflows Automation must exist with:
+    - status ACTIVE,
+    - schedule active,
+    - condition `User Inactivity in Okta = 35 days`,
+    - action that changes the user state to Suspended / Deactivated,
+    - applied to a group covering every user (typically `Everyone`).
+
+    When user sourcing is delegated to an external directory (Active
+    Directory or LDAP), the requirement is N/A on the Okta side — the
+    connected directory is expected to enforce inactivity-based
+    deactivation instead. Threshold override:
+    `okta_user_inactivity_max_days` in the audit config.
+    """
+
+    def execute(self) -> list[CheckReportOkta]:
+        findings: list[CheckReportOkta] = []
+        audit_config = user_client.audit_config or {}
+        threshold_days = audit_config.get(
+            "okta_user_inactivity_max_days", DEFAULT_INACTIVITY_DAYS
+        )
+        org_domain = user_client.provider.identity.org_domain
+
+        for scope_key in ("automations", "identity_providers"):
+            missing_scope = user_client.missing_scope.get(scope_key)
+            if missing_scope:
+                findings.append(
+                    missing_user_scope_finding(
+                        self.metadata(), org_domain, missing_scope
+                    )
+                )
+                return findings
+
+        # External-directory N/A path.
+        if user_client.external_directory_idps:
+            idp_names = ", ".join(
+                f"'{idp.name}' (type={idp.type})"
+                for idp in user_client.external_directory_idps.values()
+            )
+            placeholder = UserAutomation(
+                id="okta-user-inactivity-na-external-directory",
+                name="(external directory enforces inactivity)",
+                status="N/A",
+            )
+            report = CheckReportOkta(
+                metadata=self.metadata(),
+                resource=placeholder,
+                org_domain=org_domain,
+            )
+            report.status = "MANUAL"
+            report.status_extended = (
+                "User sourcing is delegated to an external directory "
+                f"({idp_names}). The 35-day inactivity disable requirement is "
+                "expected to be enforced by the connected directory rather "
+                "than by an Okta automation. Confirm out-of-band that the "
+                "external directory disables accounts after "
+                f"{threshold_days} days of inactivity."
+            )
+            findings.append(report)
+            return findings
+
+        compliant_automations = [
+            automation
+            for automation in user_client.automations.values()
+            if _is_compliant(automation, threshold_days)
+        ]
+
+        if not user_client.automations:
+            placeholder = UserAutomation(
+                id="okta-user-inactivity-no-automations",
+                name="(no automations configured)",
+                status="MISSING",
+            )
+            report = CheckReportOkta(
+                metadata=self.metadata(),
+                resource=placeholder,
+                org_domain=org_domain,
+            )
+            report.status = "FAIL"
+            report.status_extended = (
+                "No Okta Workflows automations are configured. Create an "
+                "automation that suspends or deactivates users after "
+                f"{threshold_days} days of inactivity, scoped to a group "
+                "covering every user (typically 'Everyone'), with an active "
+                "schedule."
+            )
+            findings.append(report)
+            return findings
+
+        if compliant_automations:
+            for automation in compliant_automations:
+                report = CheckReportOkta(
+                    metadata=self.metadata(),
+                    resource=automation,
+                    org_domain=org_domain,
+                )
+                report.status = "PASS"
+                groups_label = ", ".join(automation.applies_to_groups)
+                report.status_extended = (
+                    f"Okta automation '{automation.name}' is ACTIVE with an "
+                    f"active schedule, triggers after "
+                    f"{automation.inactivity_days} days of inactivity, and "
+                    f"changes the user state to "
+                    f"{automation.lifecycle_action or 'unset'}. "
+                    f"Applied to group(s): {groups_label}. Verify that these "
+                    "group(s) cover every user. Okta has no built-in "
+                    "'Everyone' group ID, so tenant-wide coverage cannot be "
+                    "asserted automatically."
+                )
+                findings.append(report)
+            return findings
+
+        # Automations exist but none satisfy the predicate — surface the
+        # closest candidate for the auditor.
+        candidate = _closest_candidate(user_client.automations.values())
+        report = CheckReportOkta(
+            metadata=self.metadata(),
+            resource=candidate
+            or UserAutomation(
+                id="okta-user-inactivity-noncompliant",
+                name="(no compliant automation)",
+                status="MISSING",
+            ),
+            org_domain=org_domain,
+        )
+        report.status = "FAIL"
+        report.status_extended = _failure_message(candidate, threshold_days)
+        findings.append(report)
+        return findings
+
+
+def _is_compliant(automation: UserAutomation, threshold_days: int) -> bool:
+    # `applies_to_groups` must be non-empty — Okta USER_LIFECYCLE policies
+    # do not implicitly cover every user; the scope is whatever group IDs
+    # the operator put in `people.groups.include`. An empty scope means
+    # the automation runs against nobody. Operator must still verify those
+    # group(s) cover the intended user population (surfaced in the PASS
+    # status_extended).
+    return bool(
+        automation.status.upper() == "ACTIVE"
+        and automation.schedule_status.upper() == "ACTIVE"
+        and automation.inactivity_days is not None
+        and automation.inactivity_days <= threshold_days
+        and (automation.lifecycle_action or "").upper() in SUSPENSION_LIFECYCLE_ACTIONS
+        and bool(automation.applies_to_groups)
+    )
+
+
+def _closest_candidate(automations):
+    automations = list(automations)
+    if not automations:
+        return None
+    automations.sort(
+        key=lambda a: (
+            0 if a.status.upper() == "ACTIVE" else 1,
+            0 if a.schedule_status.upper() == "ACTIVE" else 1,
+            (
+                abs(a.inactivity_days - DEFAULT_INACTIVITY_DAYS)
+                if a.inactivity_days is not None
+                else 10_000
+            ),
+            a.name,
+        )
+    )
+    return automations[0]
+
+
+def _failure_message(automation, threshold_days):
+    if automation is None:
+        return f"No Okta automation enforces {threshold_days}-day inactivity disable."
+    issues = []
+    if automation.status.upper() != "ACTIVE":
+        issues.append(f"status {automation.status or 'unset'}")
+    if automation.schedule_status.upper() != "ACTIVE":
+        issues.append(f"schedule {automation.schedule_status or 'unset'}")
+    if automation.inactivity_days is None:
+        issues.append("no inactivity condition")
+    elif automation.inactivity_days > threshold_days:
+        issues.append(
+            f"inactivity {automation.inactivity_days}d (max {threshold_days}d)"
+        )
+    action = (automation.lifecycle_action or "").upper()
+    if action not in SUSPENSION_LIFECYCLE_ACTIONS:
+        issues.append(f"action {automation.lifecycle_action or 'unset'}")
+    if not automation.applies_to_groups:
+        issues.append("no group scope")
+    detail = ", ".join(issues) if issues else "incomplete"
+    return (
+        f"Okta automation '{automation.name}' fails {threshold_days}d "
+        f"inactivity: {detail}."
+    )

prowler/providers/okta/services/user/user_service.py

@@ -0,0 +1,455 @@
+from typing import Optional
+
+from pydantic import BaseModel, ValidationError
+
+from prowler.lib.logger import logger
+from prowler.providers.okta.lib.service.pagination import paginate
+from prowler.providers.okta.lib.service.raw_fetch import (
+    get_json_paginated as raw_get_json_paginated,
+)
+from prowler.providers.okta.lib.service.service import OktaService
+
+# External-directory IdP `type` values that delegate user sourcing to a
+# separate identity store. When any of these is present and ACTIVE, the
+# STIG's 35-day inactivity disable requirement is N/A on the Okta side —
+# the connected directory is expected to enforce it instead.
+EXTERNAL_DIRECTORY_IDP_TYPES = {"ACTIVE_DIRECTORY", "LDAP"}
+
+# Okta exposes "Workflow > Automations" as USER_LIFECYCLE policies with
+# inactivity rule conditions, not as a standalone `/api/v1/automations`
+# resource. The SDK's `UserPolicyRuleCondition.inactivity` and
+# `ScheduledUserLifecycleAction` models confirm this; the API rejects
+# every other `type` candidate.
+USER_LIFECYCLE_POLICY_TYPE = "USER_LIFECYCLE"
+
+REQUIRED_SCOPES: dict[str, str] = {
+    "automations": "okta.policies.read",
+    "identity_providers": "okta.idps.read",
+}
+
+
+class User(OktaService):
+    """Fetches Okta User Lifecycle Automations and external-directory IdPs.
+
+    Populates:
+    - `self.automations` — keyed by USER_LIFECYCLE policy rule id. Each
+      entry projects the fields the 35-day inactivity check evaluates:
+      identity (`id`, `name` — taken from the rule), `status`,
+      `schedule_status` (inherited from the parent policy), the
+      `inactivity_days` condition and `applies_to_groups` scope from the
+      parent policy, and the `lifecycle_action` from the rule.
+    - `self.external_directory_idps` — keyed by IdP id. Used to short
+      circuit the STIG to N/A when user sourcing is delegated to an
+      external directory (Active Directory, LDAP).
+
+    The Okta Admin Console's "Workflow > Automations" page is rendered
+    on top of `USER_LIFECYCLE` policies in the Management API
+    (`list_policies(type='USER_LIFECYCLE')` + `list_policy_rules(...)`).
+    There is no standalone `/api/v1/automations` GET endpoint; the SDK's
+    `InactivityPolicyRuleCondition`, `UserPolicyRuleCondition`, and
+    `ScheduledUserLifecycleAction` models all hang off the policy API.
+
+    Required OAuth scopes (`REQUIRED_SCOPES`) are compared against the
+    access token's granted scopes (`provider.identity.granted_scopes`).
+    Missing scopes are recorded in `self.missing_scope` so the check
+    can emit an explicit MANUAL finding.
+    """
+
+    def __init__(self, provider):
+        super().__init__(__class__.__name__, provider)
+        granted = set(getattr(provider.identity, "granted_scopes", None) or [])
+        self.missing_scope: dict[str, Optional[str]] = {
+            resource: (scope if granted and scope not in granted else None)
+            for resource, scope in REQUIRED_SCOPES.items()
+        }
+
+        self.automations: dict[str, UserAutomation] = (
+            {} if self.missing_scope["automations"] else self._list_automations()
+        )
+        self.external_directory_idps: dict[str, ExternalDirectoryIdp] = (
+            {}
+            if self.missing_scope["identity_providers"]
+            else self._list_external_directory_idps()
+        )
+
+    def _list_automations(self) -> dict:
+        logger.info("User - Listing USER_LIFECYCLE policies and rules...")
+        try:
+            return self._run(self._fetch_automations())
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return {}
+
+    async def _fetch_automations(self) -> dict:
+        result: dict[str, UserAutomation] = {}
+        try:
+            all_policies, err = await paginate(
+                lambda after: self.client.list_policies(
+                    type=USER_LIFECYCLE_POLICY_TYPE, after=after
+                )
+            )
+        except (ValueError, ValidationError) as ex:
+            # Upstream okta-sdk-python bug: `Policy.from_dict` uses a
+            # discriminator dispatch that maps `type` → concrete Policy
+            # subclass, and `USER_LIFECYCLE` is not in the map. The SDK
+            # raises ValueError ("failed to lookup discriminator value")
+            # even though the API returns a valid policy. Fall back to
+            # raw JSON. Remove once okta-sdk-python adds
+            # USER_LIFECYCLE → UserLifecyclePolicy to the mapping.
+            logger.warning(
+                f"Okta SDK raised {type(ex).__name__} parsing USER_LIFECYCLE "
+                "policies — falling back to raw-JSON parse. This is an "
+                "okta-sdk-python deserialization bug "
+                "(missing discriminator mapping)."
+            )
+            return await self._fetch_automations_raw()
+
+        if err is not None:
+            logger.error(f"Error listing USER_LIFECYCLE policies: {err}")
+            return result
+
+        for policy in all_policies:
+            policy_id = getattr(policy, "id", "") or ""
+            if not policy_id:
+                continue
+            policy_status = _stringify_enum(getattr(policy, "status", None)) or ""
+            policy_name = getattr(policy, "name", "") or ""
+            rules = await self._fetch_rules(policy_id)
+            if rules is None:
+                # Rule typed parsing tripped an SDK validator. Re-run the
+                # whole automation discovery via raw JSON so we don't lose
+                # the rule data for this — or any other — policy. Cheaper
+                # than mixing typed and raw projections.
+                logger.warning(
+                    f"Rule typed parsing failed for USER_LIFECYCLE policy "
+                    f"{policy_id} — re-running all automations via raw-JSON."
+                )
+                return await self._fetch_automations_raw()
+            if not rules:
+                # A policy with no rules exists in the Admin Console UI as
+                # an "Automation" the operator hasn't finished configuring
+                # (no conditions, no actions). Emit a placeholder so the
+                # check FAILs with a specific message naming every missing
+                # piece, instead of pretending the policy doesn't exist.
+                result[policy_id] = _shell_automation(
+                    policy_id, policy_name, policy_status
+                )
+                continue
+            for rule in rules:
+                automation = _rule_to_automation(rule, policy)
+                if automation is None:
+                    continue
+                result[automation.id] = automation
+        return result
+
+    async def _fetch_rules(self, policy_id: str) -> Optional[list]:
+        """Return the policy's typed rules, or None to signal raw fallback.
+
+        The Okta SDK's `list_policy_rules` shares the same brittle typed
+        deserialization as `list_policies` (strict pydantic validators
+        rejecting values the API actually returns). When that happens the
+        caller can't reuse any of the typed projection for this policy —
+        we return None as a sentinel and the caller re-runs the whole
+        discovery via `_fetch_automations_raw`. Returning `[]` would
+        otherwise misclassify the policy as an "unfinished automation"
+        and FAIL it.
+        """
+        rule_fetch_limit = 100
+        try:
+            result = await self.client.list_policy_rules(
+                policy_id, limit=str(rule_fetch_limit)
+            )
+        except (ValueError, ValidationError) as ex:
+            logger.warning(
+                f"Okta SDK raised {type(ex).__name__} parsing rules for "
+                f"USER_LIFECYCLE policy {policy_id} — signaling raw fallback."
+            )
+            return None
+        err = result[-1]
+        if err is not None:
+            logger.error(
+                f"Error listing rules for USER_LIFECYCLE policy {policy_id}: {err}"
+            )
+            return []
+        rules = list(result[0] or [])
+        if len(rules) >= rule_fetch_limit:
+            logger.warning(
+                f"USER_LIFECYCLE policy {policy_id} returned {len(rules)} rules — "
+                f"the per-policy fetch limit ({rule_fetch_limit}) was hit; any "
+                "rules beyond this limit are not evaluated."
+            )
+        return rules
+
+    async def _fetch_automations_raw(self) -> dict:
+        """Raw-JSON fallback for `list_policies(type='USER_LIFECYCLE')`.
+
+        Bypasses the SDK's typed deserialization via the shared
+        `get_json_paginated` helper, then drains each policy's rules
+        via the same path. Projects everything onto our `UserAutomation`
+        snapshot which only validates the fields the check reads.
+        """
+        result: dict[str, UserAutomation] = {}
+        policies_data = await raw_get_json_paginated(
+            self.client,
+            f"/api/v1/policies?type={USER_LIFECYCLE_POLICY_TYPE}",
+            page_size=200,
+            context="USER_LIFECYCLE policies",
+        )
+        if policies_data is None:
+            return result
+
+        for policy_dict in policies_data:
+            if not isinstance(policy_dict, dict):
+                continue
+            policy_id = policy_dict.get("id")
+            if not policy_id:
+                continue
+            policy_status = (policy_dict.get("status") or "").upper()
+            policy_name = policy_dict.get("name") or ""
+
+            rules_data = await raw_get_json_paginated(
+                self.client,
+                f"/api/v1/policies/{policy_id}/rules",
+                page_size=100,
+                context=f"USER_LIFECYCLE policy {policy_id} rules",
+            )
+            if not rules_data:
+                # No rules under the policy → emit placeholder. Same
+                # rationale as the typed path: surface the unfinished
+                # automation so the check can name what's missing.
+                result[policy_id] = _shell_automation(
+                    policy_id, policy_name, policy_status
+                )
+                continue
+            for rule_dict in rules_data:
+                automation = _raw_rule_to_automation(
+                    rule_dict, policy_dict, policy_id, policy_name, policy_status
+                )
+                if automation is None:
+                    continue
+                result[automation.id] = automation
+        return result
+
+    def _list_external_directory_idps(self) -> dict:
+        logger.info("User - Listing Okta IdPs for external-directory detection...")
+        try:
+            return self._run(self._fetch_external_directory_idps())
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return {}
+
+    async def _fetch_external_directory_idps(self) -> dict:
+        result: dict[str, ExternalDirectoryIdp] = {}
+        all_idps, err = await paginate(
+            lambda after: self.client.list_identity_providers(after=after)
+        )
+        if err is not None:
+            logger.error(f"Error listing identity providers: {err}")
+            return result
+
+        for idp in all_idps:
+            idp_type = _stringify_enum(getattr(idp, "type", None)) or ""
+            if idp_type.upper() not in EXTERNAL_DIRECTORY_IDP_TYPES:
+                continue
+            idp_status = _stringify_enum(getattr(idp, "status", None)) or ""
+            if idp_status.upper() != "ACTIVE":
+                continue
+            idp_id = getattr(idp, "id", "") or ""
+            if not idp_id:
+                continue
+            result[idp_id] = ExternalDirectoryIdp(
+                id=idp_id,
+                name=getattr(idp, "name", "") or "",
+                type=idp_type,
+                status=idp_status,
+            )
+        return result
+
+
+def _rule_to_automation(rule, policy) -> Optional["UserAutomation"]:
+    """Project a typed USER_LIFECYCLE policy + rule pair onto our snapshot.
+
+    Important: in the actual API response, an Okta "Automation" is split
+    across two resources — the **inactivity condition + group scope**
+    live on the *policy* (`policy.conditions.people.users.inactivity`,
+    `policy.conditions.people.groups.include`), and the **lifecycle
+    action** lives on the *rule* (`rule.actions.user_lifecycle.action`
+    on the typed model; `updateUserLifecycle.targetStatus` on raw JSON).
+    The rule's own `conditions` is typically empty. Projecting requires
+    both — kept aligned with `_raw_rule_to_automation` so the two paths
+    yield identical snapshots.
+    """
+    rule_id = getattr(rule, "id", "") or ""
+    if not rule_id:
+        return None
+
+    policy_id = getattr(policy, "id", "") or ""
+    policy_name = getattr(policy, "name", "") or ""
+    policy_status = (_stringify_enum(getattr(policy, "status", None)) or "").upper()
+
+    # Inactivity + groups live on the POLICY in the API response.
+    inactivity_days: Optional[int] = None
+    applies_to_groups: list[str] = []
+    conditions = getattr(policy, "conditions", None)
+    people = getattr(conditions, "people", None) if conditions else None
+    users = getattr(people, "users", None) if people else None
+    inactivity = getattr(users, "inactivity", None) if users else None
+    if inactivity is not None:
+        number = getattr(inactivity, "number", None)
+        unit = (_stringify_enum(getattr(inactivity, "unit", None)) or "").upper()
+        if isinstance(number, int) and unit in {"DAYS", "DAY"}:
+            inactivity_days = number
+    groups = getattr(people, "groups", None) if people else None
+    include_groups = getattr(groups, "include", None) if groups else None
+    if include_groups:
+        applies_to_groups = [str(g) for g in include_groups if g]
+
+    # Lifecycle action lives on the RULE.
+    actions = getattr(rule, "actions", None)
+    user_lifecycle = (
+        getattr(actions, "user_lifecycle", None) if actions else None
+    ) or (getattr(actions, "userLifecycle", None) if actions else None)
+    lifecycle_action: Optional[str] = None
+    if user_lifecycle is not None:
+        for attr in ("action", "status"):
+            value = _stringify_enum(getattr(user_lifecycle, attr, None))
+            if value:
+                lifecycle_action = value.upper()
+                break
+
+    rule_name = getattr(rule, "name", "") or policy_name or "(unnamed)"
+    rule_status = _stringify_enum(getattr(rule, "status", None)) or ""
+
+    return UserAutomation(
+        id=rule_id,
+        name=rule_name,
+        status=rule_status.upper(),
+        schedule_status=policy_status,
+        inactivity_days=inactivity_days,
+        lifecycle_action=lifecycle_action,
+        applies_to_groups=applies_to_groups,
+        policy_id=policy_id,
+        policy_name=policy_name,
+    )
+
+
+def _raw_rule_to_automation(
+    rule_dict,
+    policy_dict,
+    policy_id: str,
+    policy_name: str,
+    policy_status: str,
+) -> Optional["UserAutomation"]:
+    """Project a raw USER_LIFECYCLE policy+rule pair onto our snapshot.
+
+    Important: in the actual API response, an Okta "Automation" is split
+    across two resources — the **inactivity condition + group scope**
+    live on the *policy* (`policy.conditions.people.users.inactivity`,
+    `policy.conditions.people.groups.include`), and the **lifecycle
+    action** lives on the *rule*
+    (`rule.actions.updateUserLifecycle.targetStatus`). The rule's own
+    `conditions` is typically empty `{}`. Projecting requires both.
+
+    Schedule isn't exposed by the API on either resource. Okta runs an
+    automation on its UI-configured schedule iff the policy is ACTIVE,
+    so we treat `policy.status` as the schedule proxy.
+    """
+    if not isinstance(rule_dict, dict):
+        return None
+    rule_id = rule_dict.get("id")
+    if not rule_id:
+        return None
+
+    # Inactivity + groups live on the POLICY in the API response.
+    inactivity_days: Optional[int] = None
+    applies_to_groups: list[str] = []
+    if isinstance(policy_dict, dict):
+        policy_conditions = policy_dict.get("conditions") or {}
+        people = policy_conditions.get("people") or {}
+        users = people.get("users") or {}
+        inactivity = users.get("inactivity")
+        if isinstance(inactivity, dict):
+            number = inactivity.get("number")
+            unit = (inactivity.get("unit") or "").upper()
+            if isinstance(number, int) and unit in {"DAYS", "DAY"}:
+                inactivity_days = number
+        groups = people.get("groups") or {}
+        include_groups = groups.get("include")
+        if isinstance(include_groups, list):
+            applies_to_groups = [str(g) for g in include_groups if g]
+
+    # Lifecycle action lives on the RULE under
+    # `actions.updateUserLifecycle.targetStatus` (the API uses
+    # "updateUserLifecycle" rather than the SDK's `user_lifecycle`).
+    rule_actions = rule_dict.get("actions") or {}
+    update_user_lifecycle = rule_actions.get("updateUserLifecycle") or {}
+    lifecycle_action: Optional[str] = None
+    if isinstance(update_user_lifecycle, dict):
+        target = update_user_lifecycle.get("targetStatus")
+        if isinstance(target, str) and target:
+            lifecycle_action = target.upper()
+
+    return UserAutomation(
+        id=rule_id,
+        name=(rule_dict.get("name") or policy_name or "(unnamed)"),
+        status=(rule_dict.get("status") or "").upper(),
+        schedule_status=policy_status,
+        inactivity_days=inactivity_days,
+        lifecycle_action=lifecycle_action,
+        applies_to_groups=applies_to_groups,
+        policy_id=policy_id,
+        policy_name=policy_name,
+    )
+
+
+def _shell_automation(
+    policy_id: str, policy_name: str, policy_status: str
+) -> "UserAutomation":
+    """Placeholder UserAutomation for a USER_LIFECYCLE policy with no rules.
+
+    Surfaces the unfinished automation in `self.automations` so the check
+    can list every missing piece in its FAIL message (no inactivity
+    condition, no lifecycle action, status inactive, etc.) instead of
+    silently dropping the policy.
+    """
+    upper_status = (policy_status or "").upper()
+    return UserAutomation(
+        id=policy_id,
+        name=policy_name or "(unnamed automation)",
+        status=upper_status,
+        schedule_status=upper_status,
+        inactivity_days=None,
+        lifecycle_action=None,
+        applies_to_groups=[],
+        policy_id=policy_id,
+        policy_name=policy_name,
+    )
+
+
+def _stringify_enum(value) -> Optional[str]:
+    if value is None:
+        return None
+    return getattr(value, "value", None) or str(value)
+
+
+class UserAutomation(BaseModel):
+    id: str
+    name: str = ""
+    status: str = ""
+    schedule_status: str = ""
+    inactivity_days: Optional[int] = None
+    lifecycle_action: Optional[str] = None
+    applies_to_groups: list[str] = []
+    policy_id: str = ""
+    policy_name: str = ""
+
+
+class ExternalDirectoryIdp(BaseModel):
+    id: str
+    name: str = ""
+    type: str = ""
+    status: str = ""

skills/prowler-tour/SKILL.md

@@ -0,0 +1,99 @@
+---
+name: prowler-tour
+description: >
+  Keeps product-tour definitions aligned with the UI features they describe.
+  Trigger: When modifying UI components that have associated tours, editing tour
+  definition files, or renaming data-tour-id attributes.
+license: Apache-2.0
+metadata:
+  author: prowler-cloud
+  version: "1.0"
+  scope: [root, ui]
+  auto_invoke:
+    - "Editing a UI file containing data-tour-id attributes"
+    - "Adding, updating, or removing a tour definition (*.tour.ts)"
+    - "Renaming or removing a data-tour-id attribute value"
+    - "Changing button labels or section headings on a tour-covered page"
+    - "Restructuring routes or layouts covered by a tour"
+allowed-tools: Read, Glob, Grep
+---
+
+# prowler-tour
+
+**Report-only.** This skill never edits tour files or UI files; it inspects
+the change, reports drift it finds between tours and the covered UI, and
+recommends actions for the developer to apply.
+
+## Early-exit rule
+
+Run this check first. Most UI edits are not tour-related — exit cheaply.
+
+1. Glob `ui/lib/tours/*.tour.ts`.
+2. For each tour, check whether any `coversFiles` glob pattern matches any
+   file in the current change.
+3. If no tour matches, respond **exactly**:
+
+   > No tour affected — skipping alignment check
+
+   and exit. Do not proceed to the checklist.
+4. If at least one tour matches, continue to "Drift checklist" for that tour.
+
+## Drift checklist
+
+For each affected tour, evaluate every item. Skip items that obviously do
+not apply, but list explicitly which items were checked.
+
+1. **Orphan selectors** — every step's `target` (which composes to
+   `data-tour-id="<tour-id>-<step.target>"`) must resolve to a real element
+   in the codebase. Grep `ui/` for the expected attribute value; report
+   any step whose target is missing.
+2. **Renamed selectors** — a `data-tour-id` attribute was edited in this
+   change. Match it back to any tour step referencing the old value.
+3. **Outdated copy** — a popover `title`/`description` references a button
+   label, heading, or term that no longer exists on the covered page.
+4. **Obsolete steps** — a step describes a section, panel, or workflow
+   that was removed.
+5. **Missing steps** — a new feature was added on the covered surface
+   without a corresponding step (e.g. a new panel, a new primary action,
+   a new wizard stage).
+6. **Reordered flow** — the user's path through the feature changed (e.g.
+   query builder moved before scan selection) and the step order no
+   longer reflects it.
+
+## Version-bump decision tree
+
+Apply per tour after listing drift:
+
+- **NO bump** when the change is cosmetic. Examples: fix a typo, soften
+  copy, rename a `data-tour-id` selector while keeping the same step,
+  swap one screenshot for another, tighten wording.
+- **BUMP `version`** when the user-visible flow changes materially.
+  Examples: a new step was added or removed; the order changed; an
+  anchored target was retargeted to a different panel; the tour now
+  covers a new feature on the surface.
+
+When in doubt, ask: "Would a user who already saw the previous version
+miss something useful by not seeing this one?" If yes, bump.
+
+## Output format
+
+When emitting a report, follow the exact structure in
+`references/output-format.md`. The structure is mandatory because the
+report is consumed downstream and tolerates no field reordering.
+
+## What this skill MUST NOT do
+
+- Do not edit `*.tour.ts` files. This skill is report-only.
+- Do not edit UI files to add or rename `data-tour-id` attributes.
+- Do not invent new tours. Authoring a new tour is a separate, deliberate
+  decision — the developer makes it, not the skill.
+- Do not flag drift in tours whose `coversFiles` do not match any file
+  in the current change. Stick to the early-exit rule.
+
+## See also
+
+- `references/output-format.md` — exact report template (read when
+  emitting a report).
+- `references/tours-architecture.md` — code map for the tour abstraction
+  under `ui/lib/tours/`.
+- `assets/tour-template.ts` — boilerplate for authoring a new `*.tour.ts`.

skills/prowler-tour/assets/tour-template.ts

@@ -0,0 +1,51 @@
+// @ts-nocheck -- template only; resolves once copied into `ui/lib/tours/`
+/**
+ * Tour template — copy this file to `ui/lib/tours/<your-id>.tour.ts` and
+ * fill in the placeholders. See `references/tours-architecture.md` for the
+ * design context.
+ *
+ * Conventions:
+ *   - Declare via `defineTour({...})` (NOT `: TourDefinition`) so TS
+ *     preserves the literal union of `target` values. `useDriverTour` uses
+ *     that union to validate `stepHandlers` keys and `waitForStep` args.
+ *   - `id` is kebab-case and unique across all tours.
+ *   - Anchored steps reference DOM via `data-tour-id="<id>-<step.target>"`;
+ *     the hook composes the CSS selector automatically.
+ *   - `coversFiles` lists the globs that describe the tour's surface; the
+ *     `prowler-tour` skill consumes this to decide whether to evaluate
+ *     drift on a given change.
+ *   - Material flow changes bump `version`; cosmetic edits do not.
+ */
+import {
+  defineTour,
+  TOUR_STEP_ALIGNMENTS,
+  TOUR_STEP_SIDES,
+} from "@/lib/tours/tour-types";
+
+export const yourTour = defineTour({
+  id: "your-tour-id",
+  version: 1,
+  coversFiles: [
+    // List the UI files this tour describes, using globs under `ui/`.
+    // Example: "ui/app/(prowler)/your-feature/**"
+  ],
+  steps: [
+    {
+      // Modal step — no anchor. Use for intros, outros, and any step
+      // that does not point at a specific DOM element.
+      title: "Welcome",
+      description: "Short, plain-English description.",
+    },
+    {
+      // Anchored step. The hook resolves
+      // `[data-tour-id="your-tour-id-step-name"]` lazily, so the element
+      // can be conditionally rendered as long as it exists when the step
+      // becomes active.
+      target: "step-name",
+      side: TOUR_STEP_SIDES.BOTTOM,
+      align: TOUR_STEP_ALIGNMENTS.START,
+      title: "Where the action is",
+      description: "Tell the user what to look at here and why.",
+    },
+  ],
+});

skills/prowler-tour/references/output-format.md

@@ -0,0 +1,31 @@
+# Tour Alignment Report — output format
+
+The report is consumed downstream. Field names, order, and headings are
+load-bearing — do not rename, reorder, or omit them.
+
+## Template
+
+```text
+## Tour Alignment Report
+**Tour:** `<tour-id>@v<version>`
+**Files touched:** <comma-separated list of files in the change>
+
+### Drift detected
+- <one bullet per drift item; include file:line where available>
+
+### Recommended actions
+1. <numbered, actionable steps the developer should take>
+
+### Version bump verdict
+- <BUMP | NO bump> — <one-line rationale>
+```
+
+## Rules
+
+- One report per affected tour. If multiple tours are affected, separate
+  reports with a `---` line.
+- If no drift is detected for an affected tour, still emit the report:
+  put "No drift detected." under "Drift detected" and "None required."
+  under "Recommended actions". The verdict line is still mandatory.
+- The verdict is exactly one of `BUMP` or `NO bump` — see the
+  version-bump decision tree in `SKILL.md`.

skills/prowler-tour/references/tours-architecture.md

@@ -0,0 +1,44 @@
+# Tours Architecture
+
+The product-tour abstraction lives under [`ui/lib/tours/`](../../../ui/lib/tours/).
+This skill operates on tour definitions that follow this architecture.
+
+## Code map
+
+| File | Purpose |
+|---|---|
+| `ui/lib/tours/tour-types.ts` | Public type surface: `TourDefinition`, `TourStep`, `TourId`, `TourCompletionRecord`, completion-state const map. Also exports `defineTour(...)` — the required authoring helper that preserves literal step `target`s so `useDriverTour` can type-check `stepHandlers` keys and `waitForStep` arguments. |
+| `ui/lib/tours/tour-config.ts` | `baseDriverConfig`, `getDriverConfig(theme, overrides?)`, overlay-color map. |
+| `ui/lib/tours/store/tour-completion-store.ts` | Persistence interface — the swap point for future API adapters. |
+| `ui/lib/tours/store/local-storage-adapter.ts` | The only adapter in the PoC. Key format: `prowler.tour.<id>.v<version>`. |
+| `ui/lib/tours/use-driver-tour.ts` | React hook. Initializes driver.js, derives `overlayColor` from `useTheme()`, persists completion. |
+| `ui/lib/tours/<id>.tour.ts` | One file per tour. Declared via `defineTour({...})` (not `: TourDefinition`) and imported by the page that opts the user in. |
+| `ui/styles/tours.css` | `.driver-popover.prowler-theme` — every color resolved via `var(--...)` from `globals.css`. |
+
+## Selector convention
+
+Tour steps anchor via `data-tour-id="<tour-id>-<step.target>"`. The hook
+composes the CSS selector at runtime; tour authors only provide the step
+name in `step.target`. Class-based, ID-based, structural selectors are
+forbidden — they couple tours to styling decisions that legitimately
+change.
+
+## Identity and versioning
+
+A tour is `{ id, version }`. The localStorage key composes both. A
+**material content change** bumps `version`; cosmetic edits do not. The
+decision tree lives in the parent SKILL.md.
+
+## Persistence scope
+
+Per-user, cross-tenant. A user who completed `attack-paths@v1` in tenant
+A does not see the tour again in tenant B, even if they can access the
+feature there. The future `UserTourState` model (documented in
+`design.md`, not built) is FK to `User`, not `Membership`.
+
+## Drift = #1 risk
+
+Without the maintenance skill + the optional CI gate
+(`ui/scripts/check-tour-alignment.mjs`), tours decay silently as the
+covered UI evolves. The parent SKILL.md enumerates the six drift
+categories the skill checks for.

tests/lib/outputs/jira/jira_test.py

@@ -1004,6 +1004,89 @@ class TestJiraIntegration:
             for mark in node.get("marks", [])
         )
 
+    @staticmethod
+    def _find_empty_text_nodes(node) -> List[str]:
+        # ADF forbids empty text nodes; collect any to assert the document is valid.
+        empties: List[str] = []
+
+        def walk(current) -> None:
+            if isinstance(current, dict):
+                if current.get("type") == "text" and current.get("text", "") == "":
+                    empties.append(current.get("text", ""))
+                for value in current.values():
+                    walk(value)
+            elif isinstance(current, list):
+                for item in current:
+                    walk(item)
+
+        walk(node)
+        return empties
+
+    def test_get_adf_description_empty_resource_name_has_no_empty_text_nodes(self):
+        # A resource without a name (e.g. an AWS-managed IAM policy) used to emit an
+        # empty ADF text node, making Jira reject the issue with 400 INVALID_INPUT.
+        adf_description = self.jira_integration.get_adf_description(
+            check_id="CHECK-1",
+            check_title="Sample check",
+            severity="CRITICAL",
+            severity_color="#FF0000",
+            status="FAIL",
+            status_color="#FF0000",
+            status_extended="Some status",
+            provider="aws",
+            region="eu-west-1",
+            resource_uid="arn:aws:iam::aws:policy/AdministratorAccess",
+            resource_name="",
+            recommendation_text="",
+        )
+
+        assert self._find_empty_text_nodes(adf_description) == []
+
+        table = adf_description["content"][1]
+        resource_name_row = self._find_table_row(table["content"], "Resource Name")
+        value_cell = resource_name_row["content"][1]
+        assert self._collect_text_from_cell(value_cell) == "-"
+
+    @pytest.mark.parametrize(
+        "field, header",
+        [
+            ("check_id", "Check Id"),
+            ("check_title", "Check Title"),
+            ("status_extended", "Status Extended"),
+            ("provider", "Provider"),
+            ("region", "Region"),
+            ("resource_uid", "Resource UID"),
+            ("resource_name", "Resource Name"),
+        ],
+    )
+    def test_get_adf_description_empty_plain_text_fields_render_placeholder(
+        self, field, header
+    ):
+        base_kwargs = dict(
+            check_id="CHECK-1",
+            check_title="Sample check",
+            severity="HIGH",
+            severity_color="#FF0000",
+            status="FAIL",
+            status_color="#00FF00",
+            status_extended="Some status",
+            provider="aws",
+            region="us-east-1",
+            resource_uid="resource-1",
+            resource_name="resource-name",
+            recommendation_text="",
+        )
+        base_kwargs[field] = ""
+
+        adf_description = self.jira_integration.get_adf_description(**base_kwargs)
+
+        assert self._find_empty_text_nodes(adf_description) == []
+
+        table = adf_description["content"][1]
+        row = self._find_table_row(table["content"], header)
+        value_cell = row["content"][1]
+        assert self._collect_text_from_cell(value_cell) == "-"
+
     @patch.object(Jira, "get_access_token", return_value="valid_access_token")
     @patch.object(
         Jira, "get_available_issue_types", return_value=["Bug", "Task", "Story"]

tests/providers/aws/services/elbv2/elbv2_alb_drop_invalid_header_fields_enabled/elbv2_alb_drop_invalid_header_fields_enabled_test.py

@@ -0,0 +1,254 @@
+from importlib import import_module
+from unittest import mock
+
+from boto3 import client, resource
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_REGION_EU_WEST_1,
+    AWS_REGION_EU_WEST_1_AZA,
+    AWS_REGION_EU_WEST_1_AZB,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+CHECK_MODULE = (
+    "prowler.providers.aws.services.elbv2."
+    "elbv2_alb_drop_invalid_header_fields_enabled."
+    "elbv2_alb_drop_invalid_header_fields_enabled"
+)
+ELBV2_CLIENT_PATCH = f"{CHECK_MODULE}.elbv2_client"
+GLOBAL_PROVIDER_PATCH = ".".join(
+    [
+        "prowler.providers.common.provider.Provider",
+        "get_global_provider",
+    ]
+)
+PASS_STATUS_EXTENDED = " ".join(
+    [
+        "ELBv2 ALB my-lb is configured to drop invalid",
+        "header fields.",
+    ]
+)
+FAIL_STATUS_EXTENDED = (
+    "ELBv2 ALB my-lb is not configured to drop invalid header fields."
+)
+
+
+def get_check_class():
+    return getattr(
+        import_module(CHECK_MODULE),
+        "elbv2_alb_drop_invalid_header_fields_enabled",
+    )
+
+
+class Test_elbv2_alb_drop_invalid_header_fields_enabled:
+    @mock_aws
+    def test_elb_no_balancers(self):
+        from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
+
+        with (
+            mock.patch(
+                GLOBAL_PROVIDER_PATCH,
+                return_value=set_mocked_aws_provider(
+                    [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+                ),
+            ),
+            mock.patch(
+                ELBV2_CLIENT_PATCH,
+                new=ELBv2(
+                    set_mocked_aws_provider(
+                        [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
+                        create_default_organization=False,
+                    )
+                ),
+            ),
+        ):
+            check = get_check_class()()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    @mock_aws
+    def test_elbv2_dropping_invalid_header_fields(self):
+        conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
+        ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
+
+        security_group = ec2.create_security_group(
+            GroupName="a-security-group", Description="First One"
+        )
+        vpc = ec2.create_vpc(
+            CidrBlock="172.28.7.0/24",
+            InstanceTenancy="default",
+        )
+        subnet1 = ec2.create_subnet(
+            VpcId=vpc.id,
+            CidrBlock="172.28.7.192/26",
+            AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
+        )
+        subnet2 = ec2.create_subnet(
+            VpcId=vpc.id,
+            CidrBlock="172.28.7.0/26",
+            AvailabilityZone=AWS_REGION_EU_WEST_1_AZB,
+        )
+
+        lb = conn.create_load_balancer(
+            Name="my-lb",
+            Subnets=[subnet1.id, subnet2.id],
+            SecurityGroups=[security_group.id],
+            Scheme="internal",
+            Type="application",
+        )["LoadBalancers"][0]
+
+        conn.modify_load_balancer_attributes(
+            LoadBalancerArn=lb["LoadBalancerArn"],
+            Attributes=[
+                {
+                    "Key": "routing.http.drop_invalid_header_fields.enabled",
+                    "Value": "true",
+                },
+            ],
+        )
+
+        from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
+
+        with (
+            mock.patch(
+                GLOBAL_PROVIDER_PATCH,
+                return_value=set_mocked_aws_provider(
+                    [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+                ),
+            ),
+            mock.patch(
+                ELBV2_CLIENT_PATCH,
+                new=ELBv2(
+                    set_mocked_aws_provider(
+                        [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
+                        create_default_organization=False,
+                    )
+                ),
+            ),
+        ):
+            check = get_check_class()()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == PASS_STATUS_EXTENDED
+            assert result[0].resource_id == "my-lb"
+            assert result[0].resource_arn == lb["LoadBalancerArn"]
+
+    @mock_aws
+    def test_elbv2_not_dropping_invalid_header_fields(self):
+        conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
+        ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
+
+        security_group = ec2.create_security_group(
+            GroupName="a-security-group", Description="First One"
+        )
+        vpc = ec2.create_vpc(
+            CidrBlock="172.28.7.0/24",
+            InstanceTenancy="default",
+        )
+        subnet1 = ec2.create_subnet(
+            VpcId=vpc.id,
+            CidrBlock="172.28.7.192/26",
+            AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
+        )
+        subnet2 = ec2.create_subnet(
+            VpcId=vpc.id,
+            CidrBlock="172.28.7.0/26",
+            AvailabilityZone=AWS_REGION_EU_WEST_1_AZB,
+        )
+
+        lb = conn.create_load_balancer(
+            Name="my-lb",
+            Subnets=[subnet1.id, subnet2.id],
+            SecurityGroups=[security_group.id],
+            Scheme="internal",
+            Type="application",
+        )["LoadBalancers"][0]
+
+        conn.modify_load_balancer_attributes(
+            LoadBalancerArn=lb["LoadBalancerArn"],
+            Attributes=[
+                {
+                    "Key": "routing.http.drop_invalid_header_fields.enabled",
+                    "Value": "false",
+                },
+            ],
+        )
+
+        from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
+
+        with (
+            mock.patch(
+                GLOBAL_PROVIDER_PATCH,
+                return_value=set_mocked_aws_provider(
+                    [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+                ),
+            ),
+            mock.patch(
+                ELBV2_CLIENT_PATCH,
+                new=ELBv2(
+                    set_mocked_aws_provider(
+                        [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
+                        create_default_organization=False,
+                    )
+                ),
+            ),
+        ):
+            check = get_check_class()()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == FAIL_STATUS_EXTENDED
+            assert result[0].resource_id == "my-lb"
+            assert result[0].resource_arn == lb["LoadBalancerArn"]
+
+    @mock_aws
+    def test_elbv2_network_load_balancer_ignored(self):
+        conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
+        ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
+
+        vpc = ec2.create_vpc(
+            CidrBlock="172.28.7.0/24",
+            InstanceTenancy="default",
+        )
+        subnet1 = ec2.create_subnet(
+            VpcId=vpc.id,
+            CidrBlock="172.28.7.192/26",
+            AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
+        )
+
+        conn.create_load_balancer(
+            Name="my-nlb",
+            Subnets=[subnet1.id],
+            Scheme="internal",
+            Type="network",
+        )
+
+        from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
+
+        with (
+            mock.patch(
+                GLOBAL_PROVIDER_PATCH,
+                return_value=set_mocked_aws_provider(
+                    [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+                ),
+            ),
+            mock.patch(
+                ELBV2_CLIENT_PATCH,
+                new=ELBv2(
+                    set_mocked_aws_provider(
+                        [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
+                        create_default_organization=False,
+                    )
+                ),
+            ),
+        ):
+            check = get_check_class()()
+            result = check.execute()
+
+            assert len(result) == 0

tests/providers/gcp/services/iam/iam_service_account_unused/iam_service_account_unused_test.py

@@ -179,3 +179,60 @@ class Test_iam_service_account_unused:
             assert result[1].project_id == GCP_PROJECT_ID
             assert result[1].location == GCP_US_CENTER1_LOCATION
             assert result[1].resource == iam_client.service_accounts[1]
+
+    def test_iam_service_account_disabled(self):
+        iam_client = mock.MagicMock()
+        monitoring_client = mock.MagicMock()
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.gcp.services.iam.iam_service_account_unused.iam_service_account_unused.iam_client",
+                new=iam_client,
+            ),
+            mock.patch(
+                "prowler.providers.gcp.services.iam.iam_service_account_unused.iam_service_account_unused.monitoring_client",
+                new=monitoring_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.iam.iam_service import ServiceAccount
+            from prowler.providers.gcp.services.iam.iam_service_account_unused.iam_service_account_unused import (
+                iam_service_account_unused,
+            )
+
+            iam_client.project_ids = [GCP_PROJECT_ID]
+            iam_client.region = GCP_US_CENTER1_LOCATION
+
+            iam_client.service_accounts = [
+                ServiceAccount(
+                    name="projects/my-project/serviceAccounts/disabled-sa@my-project.iam.gserviceaccount.com",
+                    email="disabled-sa@my-project.iam.gserviceaccount.com",
+                    display_name="Disabled service account",
+                    keys=[],
+                    project_id=GCP_PROJECT_ID,
+                    uniqueId="999888877776666",
+                    disabled=True,
+                )
+            ]
+
+            # The account is absent from the usage metrics, so a non-disabled
+            # account here would FAIL. Being disabled must take precedence and
+            # PASS, since a disabled account cannot authenticate or be used.
+            monitoring_client.sa_api_metrics = set()
+            monitoring_client.audit_config = {"max_unused_account_days": 30}
+
+            check = iam_service_account_unused()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Service Account {iam_client.service_accounts[0].email} is disabled and cannot be used."
+            )
+            assert result[0].resource_id == iam_client.service_accounts[0].email
+            assert result[0].project_id == GCP_PROJECT_ID
+            assert result[0].location == GCP_US_CENTER1_LOCATION
+            assert result[0].resource == iam_client.service_accounts[0]

tests/providers/gcp/services/logging/logging_service_test.py

@@ -1,4 +1,4 @@
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from prowler.providers.gcp.services.logging.logging_service import Logging
 from tests.providers.gcp.gcp_fixtures import (
@@ -66,3 +66,74 @@ class TestLoggingService:
                 == "resource.type=gae_app AND severity>=ERROR"
             )
             assert logging_client.metrics[1].project_id == GCP_PROJECT_ID
+
+    def test_org_sinks_fetched_when_project_has_organization(self):
+        """_get_org_sinks() appends org-level sinks when projects have an org."""
+        from prowler.providers.gcp.models import GCPOrganization, GCPProject
+
+        org_id = "999888777"
+        provider = set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID])
+        provider.projects = {
+            GCP_PROJECT_ID: GCPProject(
+                id=GCP_PROJECT_ID,
+                number="123456789012",
+                name="test",
+                labels={},
+                lifecycle_state="ACTIVE",
+                organization=GCPOrganization(id=org_id, name=f"organizations/{org_id}"),
+            )
+        }
+
+        mock_client = MagicMock()
+        mock_client.sinks().list().execute.return_value = {
+            "sinks": [
+                {
+                    "name": "org-sink",
+                    "destination": "storage.googleapis.com/org-bucket",
+                    "filter": "all",
+                    "includeChildren": True,
+                }
+            ]
+        }
+        mock_client.sinks().list_next.return_value = None
+        mock_client.projects().metrics().list().execute.return_value = {"metrics": []}
+        mock_client.projects().metrics().list_next.return_value = None
+
+        with (
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__is_api_active__",
+                new=mock_is_api_active,
+            ),
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__generate_client__",
+                return_value=mock_client,
+            ),
+        ):
+            logging_svc = Logging(provider)
+
+        org_sinks = [
+            s for s in logging_svc.sinks if s.project_id == f"organizations/{org_id}"
+        ]
+        assert len(org_sinks) == 1
+        assert org_sinks[0].name == "org-sink"
+        assert org_sinks[0].include_children is True
+        assert org_sinks[0].filter == "all"
+
+    def test_org_sinks_skipped_when_no_organization(self):
+        """_get_org_sinks() adds nothing when projects have no organization."""
+        with (
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__is_api_active__",
+                new=mock_is_api_active,
+            ),
+            patch(
+                "prowler.providers.gcp.lib.service.service.GCPService.__generate_client__",
+                new=mock_api_client,
+            ),
+        ):
+            logging_svc = Logging(set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID]))
+
+        org_sinks = [
+            s for s in logging_svc.sinks if s.project_id.startswith("organizations/")
+        ]
+        assert org_sinks == []

tests/providers/gcp/services/logging/logging_sink_created/logging_sink_created_test.py

@@ -1,6 +1,6 @@
 from unittest.mock import MagicMock, patch
 
-from prowler.providers.gcp.models import GCPProject
+from prowler.providers.gcp.models import GCPOrganization, GCPProject
 from tests.providers.gcp.gcp_fixtures import (
     GCP_EU1_LOCATION,
     GCP_PROJECT_ID,
@@ -268,6 +268,7 @@ class Test_logging_sink_created:
             sink.name = None
             sink.filter = "all"
             sink.project_id = GCP_PROJECT_ID
+            sink.include_children = False
 
             logging_client.project_ids = [GCP_PROJECT_ID]
             logging_client.region = GCP_EU1_LOCATION
@@ -311,9 +312,10 @@ class Test_logging_sink_created:
             )
 
             # Create a MagicMock sink object without name attribute
-            sink = MagicMock(spec=["filter", "project_id"])
+            sink = MagicMock(spec=["filter", "project_id", "include_children"])
             sink.filter = "all"
             sink.project_id = GCP_PROJECT_ID
+            sink.include_children = False
 
             logging_client.project_ids = [GCP_PROJECT_ID]
             logging_client.region = GCP_EU1_LOCATION
@@ -336,3 +338,175 @@ class Test_logging_sink_created:
             assert result[0].resource_id == "unknown"
             assert result[0].project_id == GCP_PROJECT_ID
             assert result[0].location == GCP_EU1_LOCATION
+
+    def test_org_level_sink_with_include_children_passes(self):
+        """Projects covered by an org-level sink with includeChildren=True should PASS."""
+        logging_client = MagicMock()
+        org_id = "111222333"
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            patch(
+                "prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
+                new=logging_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.logging.logging_service import Sink
+            from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
+                logging_sink_created,
+            )
+
+            logging_client.project_ids = [GCP_PROJECT_ID]
+            logging_client.region = GCP_EU1_LOCATION
+            logging_client.sinks = [
+                Sink(
+                    name="org-sink",
+                    destination="storage.googleapis.com/org-bucket",
+                    filter="all",
+                    project_id=f"organizations/{org_id}",
+                    include_children=True,
+                )
+            ]
+            logging_client.projects = {
+                GCP_PROJECT_ID: GCPProject(
+                    id=GCP_PROJECT_ID,
+                    number="123456789012",
+                    name="test",
+                    labels={},
+                    lifecycle_state="ACTIVE",
+                    organization=GCPOrganization(
+                        id=org_id, name=f"organizations/{org_id}"
+                    ),
+                )
+            }
+
+            check = logging_sink_created()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Sink org-sink at organization level is exporting copies of all the log entries in project {GCP_PROJECT_ID}."
+            )
+            assert result[0].resource_id == "org-sink"
+            assert result[0].project_id == GCP_PROJECT_ID
+            assert result[0].location == GCP_EU1_LOCATION
+
+    def test_org_level_sink_without_include_children_fails(self):
+        """Projects NOT covered by includeChildren should still FAIL if no direct project sink."""
+        logging_client = MagicMock()
+        org_id = "111222333"
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            patch(
+                "prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
+                new=logging_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.logging.logging_service import Sink
+            from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
+                logging_sink_created,
+            )
+
+            logging_client.project_ids = [GCP_PROJECT_ID]
+            logging_client.region = GCP_EU1_LOCATION
+            logging_client.sinks = [
+                Sink(
+                    name="org-sink-no-children",
+                    destination="storage.googleapis.com/org-bucket",
+                    filter="all",
+                    project_id=f"organizations/{org_id}",
+                    include_children=False,
+                )
+            ]
+            logging_client.projects = {
+                GCP_PROJECT_ID: GCPProject(
+                    id=GCP_PROJECT_ID,
+                    number="123456789012",
+                    name="test",
+                    labels={},
+                    lifecycle_state="ACTIVE",
+                    organization=GCPOrganization(
+                        id=org_id, name=f"organizations/{org_id}"
+                    ),
+                )
+            }
+
+            check = logging_sink_created()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"There are no logging sinks to export copies of all the log entries in project {GCP_PROJECT_ID}."
+            )
+            assert result[0].resource_id == GCP_PROJECT_ID
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_project_sink_takes_precedence_over_org_sink(self):
+        """A direct project sink should be reported even when an org-level sink also covers the project."""
+        logging_client = MagicMock()
+        org_id = "111222333"
+
+        with (
+            patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_gcp_provider(),
+            ),
+            patch(
+                "prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created.logging_client",
+                new=logging_client,
+            ),
+        ):
+            from prowler.providers.gcp.services.logging.logging_service import Sink
+            from prowler.providers.gcp.services.logging.logging_sink_created.logging_sink_created import (
+                logging_sink_created,
+            )
+
+            logging_client.project_ids = [GCP_PROJECT_ID]
+            logging_client.region = GCP_EU1_LOCATION
+            logging_client.sinks = [
+                Sink(
+                    name="project-sink",
+                    destination="storage.googleapis.com/project-bucket",
+                    filter="all",
+                    project_id=GCP_PROJECT_ID,
+                ),
+                Sink(
+                    name="org-sink",
+                    destination="storage.googleapis.com/org-bucket",
+                    filter="all",
+                    project_id=f"organizations/{org_id}",
+                    include_children=True,
+                ),
+            ]
+            logging_client.projects = {
+                GCP_PROJECT_ID: GCPProject(
+                    id=GCP_PROJECT_ID,
+                    number="123456789012",
+                    name="test",
+                    labels={},
+                    lifecycle_state="ACTIVE",
+                    organization=GCPOrganization(
+                        id=org_id, name=f"organizations/{org_id}"
+                    ),
+                )
+            }
+
+            check = logging_sink_created()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Sink project-sink is enabled exporting copies of all the log entries in project {GCP_PROJECT_ID}."
+            )
+            assert result[0].resource_id == "project-sink"
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/okta/lib/service/raw_fetch_test.py

@@ -0,0 +1,152 @@
+"""Tests for the raw-JSON HTTP helpers in
+`prowler.providers.okta.lib.service.raw_fetch`.
+
+Covers `get_json` (single-shot) and `get_json_paginated`
+(drains list endpoints via the `Link: rel="next"` cursor).
+"""
+
+import asyncio
+import json
+from unittest import mock
+
+from prowler.providers.okta.lib.service.raw_fetch import (
+    get_json,
+    get_json_paginated,
+)
+
+
+def _run(coro):
+    return asyncio.run(coro)
+
+
+def _mock_response(headers: dict = None):
+    r = mock.MagicMock()
+    r.headers = headers or {}
+    return r
+
+
+class Test_get_json:
+    def test_returns_parsed_json_on_success(self):
+        client = mock.MagicMock()
+
+        async def create(*_a, **_k):
+            return ({"url": "/x"}, None)
+
+        async def execute(_req):
+            return (_mock_response(), json.dumps({"hello": "world"}), None)
+
+        client._request_executor.create_request = create
+        client._request_executor.execute = execute
+
+        assert _run(get_json(client, "/x")) == {"hello": "world"}
+
+    def test_returns_none_on_create_request_error(self):
+        client = mock.MagicMock()
+
+        async def create(*_a, **_k):
+            return (None, Exception("boom"))
+
+        client._request_executor.create_request = create
+        assert _run(get_json(client, "/x")) is None
+
+    def test_returns_none_on_execute_error(self):
+        client = mock.MagicMock()
+
+        async def create(*_a, **_k):
+            return ({"url": "/x"}, None)
+
+        async def execute(_req):
+            return (_mock_response(), None, Exception("boom"))
+
+        client._request_executor.create_request = create
+        client._request_executor.execute = execute
+        assert _run(get_json(client, "/x")) is None
+
+
+class Test_get_json_paginated:
+    def test_drains_all_pages_following_link_rel_next(self):
+        # Two pages: first carries `Link: <…?after=cur1>; rel="next"`,
+        # second has no `next`, so iteration stops.
+        client = mock.MagicMock()
+
+        page1 = [{"id": "a"}, {"id": "b"}]
+        page2 = [{"id": "c"}]
+        page1_headers = {
+            "link": '<https://acme.okta.com/api/v1/items?after=cur1>; rel="next"'
+        }
+
+        seen_urls = []
+
+        async def create(**kwargs):
+            seen_urls.append(kwargs["url"])
+            return ({"url": kwargs["url"]}, None)
+
+        async def execute(request):
+            if "after=cur1" in request["url"]:
+                return (_mock_response({}), json.dumps(page2), None)
+            return (_mock_response(page1_headers), json.dumps(page1), None)
+
+        client._request_executor.create_request = create
+        client._request_executor.execute = execute
+
+        items = _run(get_json_paginated(client, "/api/v1/items", page_size=2))
+
+        assert items == [{"id": "a"}, {"id": "b"}, {"id": "c"}]
+        assert len(seen_urls) == 2
+        assert "limit=2" in seen_urls[0]
+        # The cursor was carried into the second request.
+        assert "after=cur1" in seen_urls[1]
+        assert "limit=2" in seen_urls[1]
+
+    def test_single_page_terminates_immediately(self):
+        client = mock.MagicMock()
+
+        async def create(**kwargs):
+            return ({"url": kwargs["url"]}, None)
+
+        async def execute(_req):
+            return (_mock_response({}), json.dumps([{"id": "only"}]), None)
+
+        client._request_executor.create_request = create
+        client._request_executor.execute = execute
+
+        assert _run(get_json_paginated(client, "/api/v1/items")) == [{"id": "only"}]
+
+    def test_returns_none_when_response_is_not_a_list(self):
+        client = mock.MagicMock()
+
+        async def create(**kwargs):
+            return ({"url": kwargs["url"]}, None)
+
+        async def execute(_req):
+            return (_mock_response({}), json.dumps({"error": "nope"}), None)
+
+        client._request_executor.create_request = create
+        client._request_executor.execute = execute
+
+        assert _run(get_json_paginated(client, "/api/v1/items")) is None
+
+    def test_preserves_existing_query_string_and_overrides_limit(self):
+        # Caller already passes `type=USER_LIFECYCLE` — pagination must
+        # merge `limit` without clobbering existing params.
+        client = mock.MagicMock()
+        seen = []
+
+        async def create(**kwargs):
+            seen.append(kwargs["url"])
+            return ({"url": kwargs["url"]}, None)
+
+        async def execute(_req):
+            return (_mock_response({}), "[]", None)
+
+        client._request_executor.create_request = create
+        client._request_executor.execute = execute
+
+        _run(
+            get_json_paginated(
+                client, "/api/v1/policies?type=USER_LIFECYCLE", page_size=50
+            )
+        )
+
+        assert "type=USER_LIFECYCLE" in seen[0]
+        assert "limit=50" in seen[0]

tests/providers/okta/okta_fixtures.py

@@ -16,7 +16,13 @@ def set_mocked_okta_provider(
         session = OktaSession(
             org_domain=OKTA_ORG_DOMAIN,
             client_id=OKTA_CLIENT_ID,
-            scopes=["okta.policies.read", "okta.brands.read", "okta.apps.read"],
+            scopes=[
+                "okta.policies.read",
+                "okta.brands.read",
+                "okta.apps.read",
+                "okta.logStreams.read",
+                "okta.idps.read",
+            ],
             private_key=OKTA_PRIVATE_KEY,
         )
     if identity is None:
@@ -27,6 +33,8 @@ def set_mocked_okta_provider(
                 "okta.policies.read",
                 "okta.brands.read",
                 "okta.apps.read",
+                "okta.logStreams.read",
+                "okta.idps.read",
             ],
         )
 

tests/providers/okta/services/idp/idp_fixtures.py

@@ -0,0 +1,44 @@
+"""Shared helpers for `idp` service check tests."""
+
+from unittest import mock
+
+from prowler.providers.okta.services.idp.idp_service import OktaIdentityProvider
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+
+
+def build_idp_client(
+    identity_providers: dict = None,
+    missing_scope: dict = None,
+):
+    client = mock.MagicMock()
+    client.identity_providers = identity_providers or {}
+    client.provider = set_mocked_okta_provider()
+    client.audit_config = {}
+    client.missing_scope = missing_scope or {"identity_providers": None}
+    return client
+
+
+def smart_card_idp(
+    idp_id: str = "0oa-x509",
+    name: str = "CAC IdP",
+    status: str = "ACTIVE",
+    issuer: str = "CN=DOD ROOT CA 6",
+    kid: str = "kid-abc-123",
+):
+    return OktaIdentityProvider(
+        id=idp_id,
+        name=name,
+        type="X509",
+        status=status,
+        trust_issuer=issuer,
+        trust_kid=kid,
+    )
+
+
+def non_smart_card_idp(
+    idp_id: str = "0oa-saml",
+    name: str = "Corporate SAML",
+    type: str = "SAML2",
+    status: str = "ACTIVE",
+):
+    return OktaIdentityProvider(id=idp_id, name=name, type=type, status=status)

tests/providers/okta/services/idp/idp_service_test.py

@@ -0,0 +1,80 @@
+import json
+from unittest import mock
+
+from okta.models.identity_provider_protocol import IdentityProviderProtocol
+
+from prowler.providers.okta.services.idp.idp_service import Idp, OktaIdentityProvider
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+
+
+def _resp(headers: dict = None):
+    r = mock.MagicMock()
+    r.headers = headers or {}
+    return r
+
+
+def _fake_idp(idp_id, name, type_, status="ACTIVE", issuer=None, kid=None):
+    # Build a real `IdentityProviderProtocol` when issuer/kid are provided
+    # so the test exercises the SDK's Pydantic v2 oneOf wrapper — credentials
+    # live on `actual_instance`, not directly on the wrapper. MagicMock
+    # auto-attribute-creation would otherwise hide a missed unwrap.
+    idp = mock.MagicMock()
+    idp.id = idp_id
+    idp.name = name
+    idp.type = type_
+    idp.status = status
+    if issuer is None and kid is None:
+        idp.protocol = None
+    else:
+        idp.protocol = IdentityProviderProtocol.from_json(
+            json.dumps(
+                {
+                    "type": "MTLS",
+                    "credentials": {"trust": {"issuer": issuer, "kid": kid}},
+                }
+            )
+        )
+    return idp
+
+
+def _patch_sdk(**methods):
+    return mock.patch(
+        "prowler.providers.okta.lib.service.service.OktaSDKClient",
+        return_value=mock.MagicMock(**methods),
+    )
+
+
+class Test_Idp_service:
+    def test_fetches_idps_with_trust_fields(self):
+        provider = set_mocked_okta_provider()
+        x509 = _fake_idp(
+            "0oa1",
+            "CAC",
+            "X509",
+            issuer="CN=DOD ROOT CA 6",
+            kid="kid-1",
+        )
+        saml = _fake_idp("0oa2", "Corp", "SAML2")
+
+        async def fake_list(*_a, **_k):
+            return ([x509, saml], _resp({}), None)
+
+        with _patch_sdk(list_identity_providers=fake_list):
+            service = Idp(provider)
+
+        assert set(service.identity_providers.keys()) == {"0oa1", "0oa2"}
+        assert isinstance(service.identity_providers["0oa1"], OktaIdentityProvider)
+        assert service.identity_providers["0oa1"].trust_issuer == "CN=DOD ROOT CA 6"
+        assert service.identity_providers["0oa1"].trust_kid == "kid-1"
+        assert service.identity_providers["0oa2"].trust_issuer is None
+
+    def test_returns_empty_on_api_error(self):
+        provider = set_mocked_okta_provider()
+
+        async def failing(*_a, **_k):
+            return ([], _resp({}), Exception("API failure"))
+
+        with _patch_sdk(list_identity_providers=failing):
+            service = Idp(provider)
+
+        assert service.identity_providers == {}

tests/providers/okta/services/idp/idp_smart_card_dod_approved_ca/idp_smart_card_dod_approved_ca_test.py

@@ -0,0 +1,125 @@
+from unittest import mock
+
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+from tests.providers.okta.services.idp.idp_fixtures import (
+    build_idp_client,
+    non_smart_card_idp,
+    smart_card_idp,
+)
+
+CHECK_PATH = (
+    "prowler.providers.okta.services.idp."
+    "idp_smart_card_dod_approved_ca.idp_smart_card_dod_approved_ca.idp_client"
+)
+
+DOD_PKI_ISSUER = "CN=DoD ID CA-59, OU=PKI, OU=DoD, O=U.S. Government, C=US"
+ECA_ISSUER = "CN=ECA Root CA 4, OU=ECA, O=U.S. Government, C=US"
+NON_DOD_ISSUER = "CN=ACME Internal Root, O=Acme Corp, C=US"
+
+
+def _run_check(client):
+    with (
+        mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_okta_provider(),
+        ),
+        mock.patch(CHECK_PATH, new=client),
+    ):
+        from prowler.providers.okta.services.idp.idp_smart_card_dod_approved_ca.idp_smart_card_dod_approved_ca import (
+            idp_smart_card_dod_approved_ca,
+        )
+
+        return idp_smart_card_dod_approved_ca().execute()
+
+
+class Test_idp_smart_card_dod_approved_ca:
+    def test_pass_when_active_idp_chain_matches_dod_pki_pattern(self):
+        idp = smart_card_idp(name="CAC", issuer=DOD_PKI_ISSUER, kid="kid-x")
+        client = build_idp_client(identity_providers={idp.id: idp})
+        findings = _run_check(client)
+        assert len(findings) == 1
+        assert findings[0].status == "PASS"
+        assert "OU=DoD" in findings[0].status_extended
+        assert DOD_PKI_ISSUER in findings[0].status_extended
+
+    def test_pass_when_active_idp_chain_matches_eca_pattern(self):
+        idp = smart_card_idp(name="ECA Partner", issuer=ECA_ISSUER, kid="kid-e")
+        client = build_idp_client(identity_providers={idp.id: idp})
+        findings = _run_check(client)
+        assert findings[0].status == "PASS"
+        assert "OU=ECA" in findings[0].status_extended
+
+    def test_manual_when_active_but_issuer_does_not_match_any_pattern(self):
+        idp = smart_card_idp(name="Custom", issuer=NON_DOD_ISSUER, kid="kid-c")
+        client = build_idp_client(identity_providers={idp.id: idp})
+        findings = _run_check(client)
+        assert findings[0].status == "MANUAL"
+        assert NON_DOD_ISSUER in findings[0].status_extended
+        assert "okta_dod_approved_ca_issuer_patterns" in findings[0].status_extended
+
+    def test_pass_when_audit_config_pattern_matches(self):
+        idp = smart_card_idp(name="Custom DOD", issuer=NON_DOD_ISSUER, kid="kid-c")
+        client = build_idp_client(identity_providers={idp.id: idp})
+        client.audit_config = {
+            "okta_dod_approved_ca_issuer_patterns": [r"CN=ACME Internal Root"]
+        }
+        findings = _run_check(client)
+        assert findings[0].status == "PASS"
+
+    def test_audit_config_overrides_bundled_defaults(self):
+        # When the operator supplies a list, the bundled DEFAULT patterns
+        # are replaced (not merged) so customers can carve out a strict set.
+        idp = smart_card_idp(name="DoD", issuer=DOD_PKI_ISSUER, kid="kid-x")
+        client = build_idp_client(identity_providers={idp.id: idp})
+        client.audit_config = {
+            "okta_dod_approved_ca_issuer_patterns": [r"CN=YourTenantCustomDodCA"]
+        }
+        findings = _run_check(client)
+        assert findings[0].status == "MANUAL"
+
+    def test_malformed_audit_config_pattern_skipped(self):
+        # An invalid regex from the operator must not crash the whole check.
+        idp = smart_card_idp(name="CAC", issuer=DOD_PKI_ISSUER, kid="kid-x")
+        client = build_idp_client(identity_providers={idp.id: idp})
+        client.audit_config = {
+            "okta_dod_approved_ca_issuer_patterns": [r"[invalid(regex", r"OU=DoD"]
+        }
+        findings = _run_check(client)
+        assert findings[0].status == "PASS"
+
+    def test_fail_when_x509_idp_is_inactive(self):
+        idp = smart_card_idp(status="INACTIVE", issuer=DOD_PKI_ISSUER)
+        client = build_idp_client(identity_providers={idp.id: idp})
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert "INACTIVE" in findings[0].status_extended
+
+    def test_fail_when_no_smart_card_idp_configured(self):
+        client = build_idp_client(identity_providers={"saml": non_smart_card_idp()})
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert (
+            "No Smart Card (X509) Identity Providers are configured"
+            in findings[0].status_extended
+        )
+        assert "mutelist" in findings[0].status_extended
+
+    def test_manual_when_idps_scope_missing(self):
+        client = build_idp_client(
+            missing_scope={"identity_providers": "okta.idps.read"}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "MANUAL"
+        assert "okta.idps.read" in findings[0].status_extended
+
+    def test_multiple_x509_idps_yield_one_finding_each(self):
+        idp_a = smart_card_idp(idp_id="0oa-a", name="A", issuer=DOD_PKI_ISSUER)
+        idp_b = smart_card_idp(
+            idp_id="0oa-b", name="B", status="INACTIVE", issuer=DOD_PKI_ISSUER
+        )
+        client = build_idp_client(identity_providers={idp_a.id: idp_a, idp_b.id: idp_b})
+        findings = _run_check(client)
+        assert len(findings) == 2
+        # We don't strictly assert ordering — just that both are covered.
+        statuses = sorted(f.status for f in findings)
+        assert statuses == ["FAIL", "PASS"]

tests/providers/okta/services/signon/signon_service_test.py

@@ -1,12 +1,14 @@
 from unittest import mock
 
+from prowler.providers.okta.lib.service.pagination import (
+    next_after_cursor as _next_after_cursor,
+)
 from prowler.providers.okta.models import OktaIdentityInfo
 from prowler.providers.okta.services.signon.signon_service import (
     GlobalSessionPolicy,
     GlobalSessionPolicyRule,
     SignInPage,
     Signon,
-    _next_after_cursor,
 )
 from tests.providers.okta.okta_fixtures import (
     OKTA_CLIENT_ID,

tests/providers/okta/services/systemlog/systemlog_fixtures.py

@@ -0,0 +1,27 @@
+"""Shared helpers for `systemlog` service check tests."""
+
+from unittest import mock
+
+from prowler.providers.okta.services.systemlog.systemlog_service import LogStream
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+
+
+def build_systemlog_client(
+    log_streams: dict = None,
+    missing_scope: dict = None,
+):
+    client = mock.MagicMock()
+    client.log_streams = log_streams or {}
+    client.provider = set_mocked_okta_provider()
+    client.audit_config = {}
+    client.missing_scope = missing_scope or {"log_streams": None}
+    return client
+
+
+def log_stream(
+    stream_id: str = "log-1",
+    name: str = "EventBridge stream",
+    status: str = "ACTIVE",
+    type: str = "AWS_EVENTBRIDGE",
+):
+    return LogStream(id=stream_id, name=name, status=status, type=type)

tests/providers/okta/services/systemlog/systemlog_service_test.py

@@ -0,0 +1,185 @@
+import json
+from unittest import mock
+
+from prowler.providers.okta.models import OktaIdentityInfo
+from prowler.providers.okta.services.systemlog.systemlog_service import (
+    LogStream,
+    SystemLog,
+)
+from tests.providers.okta.okta_fixtures import (
+    OKTA_CLIENT_ID,
+    OKTA_ORG_DOMAIN,
+    set_mocked_okta_provider,
+)
+
+
+def _resp(headers: dict = None):
+    r = mock.MagicMock()
+    r.headers = headers or {}
+    return r
+
+
+def _fake_stream(
+    stream_id: str, name: str, status: str = "ACTIVE", type_: str = "AWS_EVENTBRIDGE"
+):
+    s = mock.MagicMock()
+    s.id = stream_id
+    s.name = name
+    s.status = status
+    s.type = type_
+    return s
+
+
+def _patch_sdk(**methods):
+    return mock.patch(
+        "prowler.providers.okta.lib.service.service.OktaSDKClient",
+        return_value=mock.MagicMock(**methods),
+    )
+
+
+class Test_SystemLog_service:
+    def test_fetches_active_streams(self):
+        provider = set_mocked_okta_provider()
+        s1 = _fake_stream("log-1", "EventBridge")
+        s2 = _fake_stream("log-2", "Splunk", type_="SPLUNK_CLOUD_LOGSTREAMING")
+
+        async def fake_list(*_a, **_k):
+            return ([s1, s2], _resp({}), None)
+
+        with _patch_sdk(list_log_streams=fake_list):
+            service = SystemLog(provider)
+
+        assert set(service.log_streams.keys()) == {"log-1", "log-2"}
+        assert isinstance(service.log_streams["log-1"], LogStream)
+        assert service.log_streams["log-2"].type == "SPLUNK_CLOUD_LOGSTREAMING"
+
+    def test_returns_empty_on_api_error(self):
+        provider = set_mocked_okta_provider()
+
+        async def failing(*_a, **_k):
+            return ([], _resp({}), Exception("E0000007"))
+
+        with _patch_sdk(list_log_streams=failing):
+            service = SystemLog(provider)
+
+        assert service.log_streams == {}
+
+    def test_skips_fetch_when_scope_missing(self):
+        identity = OktaIdentityInfo(
+            org_domain=OKTA_ORG_DOMAIN,
+            client_id=OKTA_CLIENT_ID,
+            granted_scopes=["okta.policies.read"],  # no logStreams scope
+        )
+        provider = set_mocked_okta_provider(identity=identity)
+
+        called = False
+
+        async def fake_list(*_a, **_k):
+            nonlocal called
+            called = True
+            return ([], _resp({}), None)
+
+        with _patch_sdk(list_log_streams=fake_list):
+            service = SystemLog(provider)
+
+        assert called is False
+        assert service.log_streams == {}
+        assert service.missing_scope["log_streams"] == "okta.logStreams.read"
+
+
+class Test_SystemLog_service_sdk_validation_fallback:
+    """Verifies the raw-JSON fallback when the Okta SDK rejects API values.
+
+    The SDK's `LogStreamSettingsAws.eventSourceName` validator uses the
+    regex `^[a-zA-Z0-9.\\-_]$` — missing the `+` quantifier, so every
+    multi-character name raises pydantic `ValidationError`. Without the
+    fallback the whole stream list is lost; with it, the raw JSON path
+    still surfaces each stream's id/name/status/type.
+    """
+
+    def test_raw_fallback_projects_streams_when_sdk_raises(self):
+        from pydantic import ValidationError
+
+        provider = set_mocked_okta_provider()
+
+        raw_payload = [
+            {
+                "id": "log-1",
+                "name": "EventBridge prod",
+                "status": "ACTIVE",
+                "type": "AWS_EVENTBRIDGE",
+            },
+            {
+                "id": "log-2",
+                "name": "Splunk staging",
+                "status": "INACTIVE",
+                "type": "SPLUNK_CLOUD_LOGSTREAMING",
+            },
+        ]
+
+        async def failing_list_log_streams(*_a, **_k):
+            try:
+                # Trigger a real pydantic ValidationError so we exercise
+                # the exact exception type the SDK raises in production.
+                from okta.models.log_stream_settings_aws import LogStreamSettingsAws
+
+                LogStreamSettingsAws(
+                    accountId="123456789012",
+                    eventSourceName="MultiCharacter",
+                    region="us-east-1",
+                )
+            except ValidationError as ve:
+                raise ve
+            return ([], _resp({}), None)
+
+        async def fake_raw_create(*_a, **_k):
+            return ({"url": "/api/v1/logStreams"}, None)
+
+        async def fake_raw_execute(_request):
+            return (None, json.dumps(raw_payload), None)
+
+        sdk = mock.MagicMock()
+        sdk.list_log_streams = failing_list_log_streams
+        sdk._request_executor.create_request = fake_raw_create
+        sdk._request_executor.execute = fake_raw_execute
+
+        with mock.patch(
+            "prowler.providers.okta.lib.service.service.OktaSDKClient",
+            return_value=sdk,
+        ):
+            service = SystemLog(provider)
+
+        assert set(service.log_streams.keys()) == {"log-1", "log-2"}
+        assert service.log_streams["log-1"].status == "ACTIVE"
+        assert service.log_streams["log-2"].status == "INACTIVE"
+        assert service.log_streams["log-2"].type == "SPLUNK_CLOUD_LOGSTREAMING"
+
+    def test_raw_fallback_handles_empty_list(self):
+        from pydantic import ValidationError
+
+        provider = set_mocked_okta_provider()
+
+        async def failing(*_a, **_k):
+            raise ValidationError.from_exception_data(
+                title="LogStreamSettingsAws",
+                line_errors=[],
+            )
+
+        async def fake_create(*_a, **_k):
+            return ({"url": "/api/v1/logStreams"}, None)
+
+        async def fake_execute(_req):
+            return (None, "[]", None)
+
+        sdk = mock.MagicMock()
+        sdk.list_log_streams = failing
+        sdk._request_executor.create_request = fake_create
+        sdk._request_executor.execute = fake_execute
+
+        with mock.patch(
+            "prowler.providers.okta.lib.service.service.OktaSDKClient",
+            return_value=sdk,
+        ):
+            service = SystemLog(provider)
+
+        assert service.log_streams == {}

tests/providers/okta/services/systemlog/systemlog_streaming_enabled/systemlog_streaming_enabled_test.py

@@ -0,0 +1,73 @@
+from unittest import mock
+
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+from tests.providers.okta.services.systemlog.systemlog_fixtures import (
+    build_systemlog_client,
+    log_stream,
+)
+
+CHECK_PATH = (
+    "prowler.providers.okta.services.systemlog."
+    "systemlog_streaming_enabled.systemlog_streaming_enabled.systemlog_client"
+)
+
+
+def _run_check(client):
+    with (
+        mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_okta_provider(),
+        ),
+        mock.patch(CHECK_PATH, new=client),
+    ):
+        from prowler.providers.okta.services.systemlog.systemlog_streaming_enabled.systemlog_streaming_enabled import (
+            systemlog_streaming_enabled,
+        )
+
+        return systemlog_streaming_enabled().execute()
+
+
+class Test_systemlog_streaming_enabled:
+    def test_pass_when_active_stream_exists(self):
+        client = build_systemlog_client(
+            log_streams={"log-1": log_stream(name="EventBridge prod")}
+        )
+        findings = _run_check(client)
+        assert len(findings) == 1
+        assert findings[0].status == "PASS"
+        assert "EventBridge prod" in findings[0].status_extended
+
+    def test_pass_when_multiple_active_streams(self):
+        client = build_systemlog_client(
+            log_streams={
+                "log-1": log_stream(stream_id="log-1", name="A"),
+                "log-2": log_stream(stream_id="log-2", name="B"),
+            }
+        )
+        findings = _run_check(client)
+        assert len(findings) == 2
+        assert all(f.status == "PASS" for f in findings)
+
+    def test_fail_when_all_streams_inactive(self):
+        client = build_systemlog_client(
+            log_streams={"log-1": log_stream(name="A", status="INACTIVE")}
+        )
+        findings = _run_check(client)
+        assert len(findings) == 1
+        assert findings[0].status == "FAIL"
+        assert "none are ACTIVE" in findings[0].status_extended
+
+    def test_fail_when_no_streams_configured(self):
+        client = build_systemlog_client(log_streams={})
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert "No Okta Log Streams are configured" in findings[0].status_extended
+        assert "mutelist" in findings[0].status_extended
+
+    def test_manual_when_scope_missing(self):
+        client = build_systemlog_client(
+            missing_scope={"log_streams": "okta.logStreams.read"}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "MANUAL"
+        assert "okta.logStreams.read" in findings[0].status_extended

tests/providers/okta/services/user/user_fixtures.py

@@ -0,0 +1,55 @@
+"""Shared helpers for `user` service check tests."""
+
+from unittest import mock
+
+from prowler.providers.okta.services.user.user_service import (
+    ExternalDirectoryIdp,
+    UserAutomation,
+)
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+
+
+def build_user_client(
+    automations: dict = None,
+    external_directory_idps: dict = None,
+    audit_config: dict = None,
+    missing_scope: dict = None,
+):
+    client = mock.MagicMock()
+    client.automations = automations or {}
+    client.external_directory_idps = external_directory_idps or {}
+    client.provider = set_mocked_okta_provider()
+    client.audit_config = audit_config or {}
+    client.missing_scope = missing_scope or {
+        "automations": None,
+        "identity_providers": None,
+    }
+    return client
+
+
+def automation(
+    automation_id: str = "auto-1",
+    name: str = "User Inactivity",
+    status: str = "ACTIVE",
+    schedule_status: str = "ACTIVE",
+    inactivity_days: int = 35,
+    lifecycle_action: str = "SUSPENDED",
+    groups: list = None,
+):
+    # `groups is None` keeps the "Everyone-equivalent" default; passing
+    # `groups=[]` lets a test exercise the empty-scope FAIL path.
+    return UserAutomation(
+        id=automation_id,
+        name=name,
+        status=status,
+        schedule_status=schedule_status,
+        inactivity_days=inactivity_days,
+        lifecycle_action=lifecycle_action,
+        applies_to_groups=["everyone"] if groups is None else groups,
+    )
+
+
+def ad_idp(idp_id: str = "0oa-ad", name: str = "Corp AD"):
+    return ExternalDirectoryIdp(
+        id=idp_id, name=name, type="ACTIVE_DIRECTORY", status="ACTIVE"
+    )

tests/providers/okta/services/user/user_inactivity_automation_35d_enabled/user_inactivity_automation_35d_enabled_test.py

@@ -0,0 +1,165 @@
+from unittest import mock
+
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+from tests.providers.okta.services.user.user_fixtures import (
+    ad_idp,
+    automation,
+    build_user_client,
+)
+
+CHECK_PATH = (
+    "prowler.providers.okta.services.user."
+    "user_inactivity_automation_35d_enabled."
+    "user_inactivity_automation_35d_enabled.user_client"
+)
+
+
+def _run_check(client):
+    with (
+        mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_okta_provider(),
+        ),
+        mock.patch(CHECK_PATH, new=client),
+    ):
+        from prowler.providers.okta.services.user.user_inactivity_automation_35d_enabled.user_inactivity_automation_35d_enabled import (
+            user_inactivity_automation_35d_enabled,
+        )
+
+        return user_inactivity_automation_35d_enabled().execute()
+
+
+class Test_user_inactivity_automation_35d_enabled:
+    def test_pass_when_compliant_automation_present(self):
+        client = build_user_client(
+            automations={"auto-1": automation(name="Inactivity 35d")}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "PASS"
+        assert "Inactivity 35d" in findings[0].status_extended
+        assert "SUSPENDED" in findings[0].status_extended
+
+    def test_pass_message_names_groups_and_asks_for_coverage_verification(self):
+        # Okta has no built-in Everyone group ID and group names vary by
+        # tenant (e.g. "pepito"), so we can't assert tenant-wide coverage
+        # automatically — surface the group IDs and let the operator verify.
+        client = build_user_client(
+            automations={"auto-1": automation(groups=["grp-A", "grp-B"])}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "PASS"
+        assert "grp-A, grp-B" in findings[0].status_extended
+        assert "cover every user" in findings[0].status_extended
+
+    def test_fail_when_applies_to_no_group(self):
+        # An automation with empty `people.groups.include` runs against
+        # nobody — Okta does not implicitly cover every user.
+        client = build_user_client(automations={"auto-1": automation(groups=[])})
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert "no group scope" in findings[0].status_extended
+
+    def test_pass_when_lower_threshold(self):
+        # Inactivity threshold lower than the default is still compliant.
+        client = build_user_client(
+            automations={"auto-1": automation(inactivity_days=14)}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "PASS"
+
+    def test_fail_when_threshold_too_high(self):
+        client = build_user_client(
+            automations={"auto-1": automation(inactivity_days=90)}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert "inactivity 90d (max 35d)" in findings[0].status_extended
+
+    def test_fail_when_status_inactive(self):
+        client = build_user_client(
+            automations={"auto-1": automation(status="INACTIVE")}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert "status INACTIVE" in findings[0].status_extended
+
+    def test_fail_when_schedule_inactive(self):
+        client = build_user_client(
+            automations={"auto-1": automation(schedule_status="INACTIVE")}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert "schedule INACTIVE" in findings[0].status_extended
+
+    def test_fail_when_wrong_lifecycle_action(self):
+        client = build_user_client(
+            automations={"auto-1": automation(lifecycle_action="ACTIVE")}
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert "action ACTIVE" in findings[0].status_extended
+
+    def test_fail_when_no_automations(self):
+        client = build_user_client(automations={})
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        assert "No Okta Workflows automations" in findings[0].status_extended
+
+    def test_fail_lists_every_missing_piece_for_unfinished_automation(self):
+        # Mirrors the real-world case where an admin clicks "Add Automation"
+        # in the UI but never configures conditions or actions. The service
+        # emits a placeholder UserAutomation so the check FAILs with a
+        # specific message instead of pretending the policy doesn't exist.
+        from prowler.providers.okta.services.user.user_service import UserAutomation
+
+        shell = UserAutomation(
+            id="pol-1",
+            name="TestCheck",
+            status="INACTIVE",
+            schedule_status="INACTIVE",
+            inactivity_days=None,
+            lifecycle_action=None,
+            applies_to_groups=[],
+            policy_id="pol-1",
+            policy_name="TestCheck",
+        )
+        client = build_user_client(automations={"pol-1": shell})
+        findings = _run_check(client)
+        assert findings[0].status == "FAIL"
+        msg = findings[0].status_extended
+        assert "TestCheck" in msg
+        assert "status INACTIVE" in msg
+        assert "schedule INACTIVE" in msg
+        assert "no inactivity condition" in msg
+        assert "action unset" in msg
+
+    def test_manual_na_when_external_directory_idp_present(self):
+        client = build_user_client(
+            automations={"auto-1": automation(inactivity_days=90)},  # non-compliant
+            external_directory_idps={"0oa-ad": ad_idp(name="Corp AD")},
+        )
+        findings = _run_check(client)
+        # External directory short-circuits to MANUAL N/A regardless of
+        # the automations state.
+        assert findings[0].status == "MANUAL"
+        assert "ACTIVE_DIRECTORY" in findings[0].status_extended
+        assert "Corp AD" in findings[0].status_extended
+
+    def test_manual_when_scope_missing(self):
+        client = build_user_client(
+            missing_scope={
+                "automations": "okta.policies.read",
+                "identity_providers": None,
+            }
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "MANUAL"
+        assert "okta.policies.read" in findings[0].status_extended
+
+    def test_threshold_overridden_via_audit_config(self):
+        client = build_user_client(
+            automations={"auto-1": automation(inactivity_days=60)},
+            audit_config={"okta_user_inactivity_max_days": 90},
+        )
+        findings = _run_check(client)
+        assert findings[0].status == "PASS"

tests/providers/okta/services/user/user_service_test.py

@@ -0,0 +1,477 @@
+import json
+from unittest import mock
+
+from prowler.providers.okta.services.user.user_service import (
+    ExternalDirectoryIdp,
+    User,
+    UserAutomation,
+    _raw_rule_to_automation,
+    _rule_to_automation,
+)
+from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
+
+
+def _resp(headers: dict = None):
+    r = mock.MagicMock()
+    r.headers = headers or {}
+    return r
+
+
+def _fake_policy(
+    policy_id,
+    name="Inactivity Policy",
+    status="ACTIVE",
+    inactivity_days=35,
+    inactivity_unit="DAYS",
+    groups=None,
+):
+    # In the actual API response, the inactivity condition and the
+    # group scope live on the *policy*, not on its rules — keep the
+    # typed fixture aligned with that shape so it mirrors raw JSON.
+    p = mock.MagicMock()
+    p.id = policy_id
+    p.name = name
+    p.status = status
+    if inactivity_days is None:
+        p.conditions.people.users.inactivity = None
+    else:
+        p.conditions.people.users.inactivity.number = inactivity_days
+        p.conditions.people.users.inactivity.unit = inactivity_unit
+    p.conditions.people.groups.include = ["everyone"] if groups is None else groups
+    return p
+
+
+def _fake_rule(
+    rule_id="rule-1",
+    name="Inactivity",
+    status="ACTIVE",
+    lifecycle_action="SUSPENDED",
+):
+    # A USER_LIFECYCLE policy rule carries only the lifecycle action;
+    # its `conditions` is typically empty.
+    r = mock.MagicMock()
+    r.id = rule_id
+    r.name = name
+    r.status = status
+    r.actions.user_lifecycle.action = lifecycle_action
+    return r
+
+
+def _fake_idp(idp_type, status="ACTIVE", idp_id="0oa-1", name="x"):
+    idp = mock.MagicMock()
+    idp.id = idp_id
+    idp.name = name
+    idp.type = idp_type
+    idp.status = status
+    return idp
+
+
+def _patch_sdk(**methods):
+    return mock.patch(
+        "prowler.providers.okta.lib.service.service.OktaSDKClient",
+        return_value=mock.MagicMock(**methods),
+    )
+
+
+class Test_rule_to_automation:
+    def test_parses_inactivity_and_lifecycle(self):
+        policy = _fake_policy("pol-1", name="Inactivity Policy")
+        rule = _fake_rule(rule_id="rule-1", name="Inactivity")
+        m = _rule_to_automation(rule, policy)
+        assert isinstance(m, UserAutomation)
+        assert m.id == "rule-1"
+        assert m.status == "ACTIVE"
+        assert m.schedule_status == "ACTIVE"
+        assert m.inactivity_days == 35
+        assert m.lifecycle_action == "SUSPENDED"
+        assert m.applies_to_groups == ["everyone"]
+        assert m.policy_id == "pol-1"
+        assert m.policy_name == "Inactivity Policy"
+
+    def test_returns_none_when_id_missing(self):
+        policy = _fake_policy("pol")
+        bad = _fake_rule()
+        bad.id = ""
+        assert _rule_to_automation(bad, policy) is None
+
+    def test_ignores_non_days_unit(self):
+        policy = _fake_policy("pol", inactivity_unit="WEEKS")
+        rule = _fake_rule()
+        m = _rule_to_automation(rule, policy)
+        assert m.inactivity_days is None
+
+    def test_reads_inactivity_and_groups_from_policy_not_rule(self):
+        # The typed path used to read inactivity/groups from the rule;
+        # an SDK update that started populating `policy.conditions`
+        # exposed the mismatch. Locking the policy-shaped projection in.
+        policy = _fake_policy("pol", inactivity_days=21, groups=["grp-x"])
+        rule = _fake_rule()
+        # Sanity: nothing inactivity-ish on the rule.
+        del rule.conditions
+        m = _rule_to_automation(rule, policy)
+        assert m.inactivity_days == 21
+        assert m.applies_to_groups == ["grp-x"]
+
+
+class Test_User_service:
+    def test_fetches_automations_via_policy_api(self):
+        provider = set_mocked_okta_provider()
+        policy = _fake_policy("pol-1")
+        rule = _fake_rule(rule_id="rule-1")
+
+        async def fake_list_policies(*_a, **_k):
+            return ([policy], _resp({}), None)
+
+        async def fake_list_rules(*_a, **_k):
+            return ([rule], _resp({}), None)
+
+        async def fake_list_idps(*_a, **_k):
+            return ([], _resp({}), None)
+
+        sdk = mock.MagicMock()
+        sdk.list_policies = fake_list_policies
+        sdk.list_policy_rules = fake_list_rules
+        sdk.list_identity_providers = fake_list_idps
+
+        with mock.patch(
+            "prowler.providers.okta.lib.service.service.OktaSDKClient",
+            return_value=sdk,
+        ):
+            service = User(provider)
+
+        assert "rule-1" in service.automations
+        assert service.automations["rule-1"].inactivity_days == 35
+        assert service.external_directory_idps == {}
+
+    def test_returns_empty_on_policies_api_error(self):
+        provider = set_mocked_okta_provider()
+
+        async def failing(*_a, **_k):
+            return ([], _resp({}), Exception("E0000007"))
+
+        async def fake_list_idps(*_a, **_k):
+            return ([], _resp({}), None)
+
+        sdk = mock.MagicMock()
+        sdk.list_policies = failing
+        sdk.list_identity_providers = fake_list_idps
+
+        with mock.patch(
+            "prowler.providers.okta.lib.service.service.OktaSDKClient",
+            return_value=sdk,
+        ):
+            service = User(provider)
+
+        assert service.automations == {}
+
+    def test_detects_external_directory_idp(self):
+        provider = set_mocked_okta_provider()
+
+        async def empty_policies(*_a, **_k):
+            return ([], _resp({}), None)
+
+        ad = _fake_idp("ACTIVE_DIRECTORY", idp_id="0oa-ad", name="Corp AD")
+        saml = _fake_idp("SAML2", idp_id="0oa-saml")
+
+        async def fake_list_idps(*_a, **_k):
+            return ([ad, saml], _resp({}), None)
+
+        sdk = mock.MagicMock()
+        sdk.list_policies = empty_policies
+        sdk.list_identity_providers = fake_list_idps
+
+        with mock.patch(
+            "prowler.providers.okta.lib.service.service.OktaSDKClient",
+            return_value=sdk,
+        ):
+            service = User(provider)
+
+        assert "0oa-ad" in service.external_directory_idps
+        assert "0oa-saml" not in service.external_directory_idps
+        assert isinstance(
+            service.external_directory_idps["0oa-ad"], ExternalDirectoryIdp
+        )
+
+
+class Test_raw_rule_to_automation:
+    def test_projects_inactivity_and_lifecycle(self):
+        # Real API shape: inactivity + groups live on the POLICY,
+        # lifecycle action lives on the RULE under
+        # `actions.updateUserLifecycle.targetStatus`.
+        policy = {
+            "id": "pol-1",
+            "name": "TestCheck",
+            "status": "ACTIVE",
+            "conditions": {
+                "people": {
+                    "users": {"inactivity": {"number": 35, "unit": "DAYS"}},
+                    "groups": {"include": ["everyone"]},
+                }
+            },
+            "type": "USER_LIFECYCLE",
+        }
+        rule = {
+            "id": "rule-1",
+            "name": "lifecycle-rule-1",
+            "status": "ACTIVE",
+            "conditions": {},
+            "actions": {
+                "updateUserLifecycle": {
+                    "targetStatus": "SUSPENDED",
+                    "quietPeriod": {"number": 0, "unit": "DAYS"},
+                }
+            },
+        }
+        m = _raw_rule_to_automation(rule, policy, "pol-1", "TestCheck", "ACTIVE")
+        assert isinstance(m, UserAutomation)
+        assert m.id == "rule-1"
+        assert m.status == "ACTIVE"
+        assert m.schedule_status == "ACTIVE"
+        assert m.inactivity_days == 35
+        assert m.lifecycle_action == "SUSPENDED"
+        assert m.applies_to_groups == ["everyone"]
+        assert m.policy_id == "pol-1"
+        assert m.policy_name == "TestCheck"
+
+    def test_returns_none_when_id_missing(self):
+        assert _raw_rule_to_automation({"name": "x"}, {}, "pol", "P", "ACTIVE") is None
+
+    def test_ignores_non_days_unit(self):
+        policy = {
+            "id": "pol",
+            "conditions": {
+                "people": {"users": {"inactivity": {"number": 5, "unit": "WEEKS"}}}
+            },
+        }
+        rule = {"id": "rule-2", "actions": {}}
+        m = _raw_rule_to_automation(rule, policy, "pol", "P", "ACTIVE")
+        assert m.inactivity_days is None
+
+    def test_missing_policy_dict_gives_empty_inactivity_and_groups(self):
+        rule = {
+            "id": "rule-3",
+            "actions": {"updateUserLifecycle": {"targetStatus": "SUSPENDED"}},
+        }
+        m = _raw_rule_to_automation(rule, None, "pol", "P", "ACTIVE")
+        assert m.inactivity_days is None
+        assert m.applies_to_groups == []
+        assert m.lifecycle_action == "SUSPENDED"
+
+
+class Test_User_service_sdk_discriminator_fallback:
+    """Verifies the raw-JSON fallback when the SDK can't deserialize USER_LIFECYCLE.
+
+    Okta SDK 3.4.2 ships a `Policy.from_dict` discriminator mapping that
+    omits `USER_LIFECYCLE`, so the typed call raises ValueError. Without
+    the fallback the whole automations list is lost; with it the raw
+    JSON path projects each rule onto a `UserAutomation` snapshot.
+    """
+
+    def test_raw_fallback_projects_user_lifecycle_policy_rules(self):
+        provider = set_mocked_okta_provider()
+
+        # Real API shape: inactivity + groups on POLICY, lifecycle
+        # action on RULE under `actions.updateUserLifecycle.targetStatus`.
+        policy_payload = [
+            {
+                "id": "pol-1",
+                "name": "TestCheck",
+                "status": "ACTIVE",
+                "type": "USER_LIFECYCLE",
+                "conditions": {
+                    "people": {
+                        "users": {"inactivity": {"number": 35, "unit": "DAYS"}},
+                        "groups": {"include": ["everyone"]},
+                    }
+                },
+            }
+        ]
+        rules_payload = [
+            {
+                "id": "rule-1",
+                "name": "lifecycle-rule-1",
+                "status": "ACTIVE",
+                "conditions": {},
+                "actions": {
+                    "updateUserLifecycle": {
+                        "targetStatus": "SUSPENDED",
+                        "quietPeriod": {"number": 0, "unit": "DAYS"},
+                    }
+                },
+            }
+        ]
+
+        async def failing_list_policies(*_a, **_k):
+            raise ValueError(
+                "Policy failed to lookup discriminator value from {...}. "
+                "Discriminator property name: type, mapping: {...}"
+            )
+
+        async def fake_list_idps(*_a, **_k):
+            return ([], _resp({}), None)
+
+        async def fake_raw_create(*_a, **kwargs):
+            url = kwargs.get("url", "") or ""
+            return ({"url": url}, None)
+
+        async def fake_raw_execute(request):
+            url = request.get("url", "")
+            if "/api/v1/policies/pol-1/rules" in url:
+                return (None, json.dumps(rules_payload), None)
+            if "/api/v1/policies" in url:
+                return (None, json.dumps(policy_payload), None)
+            return (None, "[]", None)
+
+        sdk = mock.MagicMock()
+        sdk.list_policies = failing_list_policies
+        sdk.list_identity_providers = fake_list_idps
+        sdk._request_executor.create_request = fake_raw_create
+        sdk._request_executor.execute = fake_raw_execute
+
+        with mock.patch(
+            "prowler.providers.okta.lib.service.service.OktaSDKClient",
+            return_value=sdk,
+        ):
+            service = User(provider)
+
+        assert "rule-1" in service.automations
+        a = service.automations["rule-1"]
+        assert a.inactivity_days == 35
+        assert a.lifecycle_action == "SUSPENDED"
+        assert a.schedule_status == "ACTIVE"
+        assert a.policy_id == "pol-1"
+        assert a.policy_name == "TestCheck"
+
+    def test_raw_fallback_emits_shell_for_policy_with_no_rules(self):
+        # Mirrors the real-world tenant state where an admin clicked
+        # "Add Automation" in the UI but never configured conditions or
+        # actions. The policy exists; it has zero rules. The raw fallback
+        # must surface the policy as a shell UserAutomation so the check
+        # FAILs with a specific message instead of dropping it.
+        provider = set_mocked_okta_provider()
+
+        async def failing_list_policies(*_a, **_k):
+            raise ValueError("missing discriminator mapping")
+
+        async def fake_list_idps(*_a, **_k):
+            return ([], _resp({}), None)
+
+        async def fake_raw_create(*_a, **kwargs):
+            return ({"url": kwargs.get("url", "") or ""}, None)
+
+        async def fake_raw_execute(request):
+            url = request.get("url", "")
+            if "/api/v1/policies/pol-empty/rules" in url:
+                return (None, "[]", None)
+            if "/api/v1/policies" in url:
+                return (
+                    None,
+                    json.dumps(
+                        [
+                            {
+                                "id": "pol-empty",
+                                "name": "TestCheck",
+                                "status": "INACTIVE",
+                                "type": "USER_LIFECYCLE",
+                            }
+                        ]
+                    ),
+                    None,
+                )
+            return (None, "[]", None)
+
+        sdk = mock.MagicMock()
+        sdk.list_policies = failing_list_policies
+        sdk.list_identity_providers = fake_list_idps
+        sdk._request_executor.create_request = fake_raw_create
+        sdk._request_executor.execute = fake_raw_execute
+
+        with mock.patch(
+            "prowler.providers.okta.lib.service.service.OktaSDKClient",
+            return_value=sdk,
+        ):
+            service = User(provider)
+
+        assert "pol-empty" in service.automations
+        shell = service.automations["pol-empty"]
+        assert shell.name == "TestCheck"
+        assert shell.status == "INACTIVE"
+        assert shell.schedule_status == "INACTIVE"
+        assert shell.inactivity_days is None
+        assert shell.lifecycle_action is None
+        assert shell.applies_to_groups == []
+        assert shell.policy_id == "pol-empty"
+
+    def test_rule_typed_failure_triggers_raw_fallback_for_all_policies(self):
+        # When the typed `list_policies` succeeds but the typed
+        # `list_policy_rules` fails for a policy, the previous behavior
+        # was to emit a shell automation — silently misclassifying a
+        # valid automation as "unfinished". Now `_fetch_rules` returns
+        # None as a sentinel and the caller re-runs the entire
+        # discovery via raw JSON so no rule data is lost.
+        provider = set_mocked_okta_provider()
+
+        typed_policy = _fake_policy(
+            "pol-1", name="TestCheck", inactivity_days=35, groups=["everyone"]
+        )
+
+        async def fake_list_policies(*_a, **_k):
+            return ([typed_policy], _resp({}), None)
+
+        async def failing_list_policy_rules(*_a, **_k):
+            raise ValueError("KnowledgeConstraint.types expected uppercase")
+
+        async def fake_list_idps(*_a, **_k):
+            return ([], _resp({}), None)
+
+        raw_policy_payload = [
+            {
+                "id": "pol-1",
+                "name": "TestCheck",
+                "status": "ACTIVE",
+                "type": "USER_LIFECYCLE",
+                "conditions": {
+                    "people": {
+                        "users": {"inactivity": {"number": 35, "unit": "DAYS"}},
+                        "groups": {"include": ["everyone"]},
+                    }
+                },
+            }
+        ]
+        raw_rules_payload = [
+            {
+                "id": "rule-1",
+                "name": "lifecycle-rule-1",
+                "status": "ACTIVE",
+                "actions": {"updateUserLifecycle": {"targetStatus": "SUSPENDED"}},
+            }
+        ]
+
+        async def fake_raw_create(*_a, **kwargs):
+            return ({"url": kwargs.get("url", "") or ""}, None)
+
+        async def fake_raw_execute(request):
+            url = request.get("url", "")
+            if "/api/v1/policies/pol-1/rules" in url:
+                return (None, json.dumps(raw_rules_payload), None)
+            if "/api/v1/policies" in url:
+                return (None, json.dumps(raw_policy_payload), None)
+            return (None, "[]", None)
+
+        sdk = mock.MagicMock()
+        sdk.list_policies = fake_list_policies
+        sdk.list_policy_rules = failing_list_policy_rules
+        sdk.list_identity_providers = fake_list_idps
+        sdk._request_executor.create_request = fake_raw_create
+        sdk._request_executor.execute = fake_raw_execute
+
+        with mock.patch(
+            "prowler.providers.okta.lib.service.service.OktaSDKClient",
+            return_value=sdk,
+        ):
+            service = User(provider)
+
+        # Raw-projected automation, not a shell.
+        assert "rule-1" in service.automations
+        assert service.automations["rule-1"].inactivity_days == 35
+        assert service.automations["rule-1"].lifecycle_action == "SUSPENDED"

ui/AGENTS.md

@@ -14,40 +14,46 @@
 > - [`playwright`](../skills/playwright/SKILL.md) - Page Object Model, selectors
 > - [`vitest`](../skills/vitest/SKILL.md) - Unit testing with React Testing Library
 > - [`tdd`](../skills/tdd/SKILL.md) - TDD workflow (MANDATORY for UI tasks)
+> - [`prowler-tour`](../skills/prowler-tour/SKILL.md) - Keep product-tour definitions aligned with the UI
 
 ## Auto-invoke Skills
 
 When performing these actions, ALWAYS invoke the corresponding skill FIRST:
 
-| Action                                                         | Skill               |
-| -------------------------------------------------------------- | ------------------- |
-| Add changelog entry for a PR or feature                        | `prowler-changelog` |
-| App Router / Server Actions                                    | `nextjs-16`         |
-| Building AI chat features                                      | `ai-sdk-5`          |
-| Committing changes                                             | `prowler-commit`    |
-| Create PR that requires changelog entry                        | `prowler-changelog` |
-| Creating Zod schemas                                           | `zod-4`             |
-| Creating a git commit                                          | `prowler-commit`    |
-| Creating/modifying Prowler UI components                       | `prowler-ui`        |
-| Fixing bug                                                     | `tdd`               |
-| Implementing feature                                           | `tdd`               |
-| Modifying component                                            | `tdd`               |
-| Refactoring code                                               | `tdd`               |
-| Review changelog format and conventions                        | `prowler-changelog` |
-| Testing hooks or utilities                                     | `vitest`            |
-| Update CHANGELOG.md in any component                           | `prowler-changelog` |
-| Using Zustand stores                                           | `zustand-5`         |
-| Working on Prowler UI structure (actions/adapters/types/hooks) | `prowler-ui`        |
-| Working on task                                                | `tdd`               |
-| Working with Prowler UI test helpers/pages                     | `prowler-test-ui`   |
-| Working with Tailwind classes                                  | `tailwind-4`        |
-| Writing Playwright E2E tests                                   | `playwright`        |
-| Writing Prowler UI E2E tests                                   | `prowler-test-ui`   |
-| Writing React component tests                                  | `vitest`            |
-| Writing React components                                       | `react-19`          |
-| Writing TypeScript types/interfaces                            | `typescript`        |
-| Writing Vitest tests                                           | `vitest`            |
-| Writing unit tests for UI                                      | `vitest`            |
+| Action                                                            | Skill               |
+| ----------------------------------------------------------------- | ------------------- |
+| Add changelog entry for a PR or feature                           | `prowler-changelog` |
+| Adding, updating, or removing a tour definition (\*.tour.ts)      | `prowler-tour`      |
+| App Router / Server Actions                                       | `nextjs-16`         |
+| Building AI chat features                                         | `ai-sdk-5`          |
+| Changing button labels or section headings on a tour-covered page | `prowler-tour`      |
+| Committing changes                                                | `prowler-commit`    |
+| Create PR that requires changelog entry                           | `prowler-changelog` |
+| Creating Zod schemas                                              | `zod-4`             |
+| Creating a git commit                                             | `prowler-commit`    |
+| Creating/modifying Prowler UI components                          | `prowler-ui`        |
+| Editing a UI file containing data-tour-id attributes              | `prowler-tour`      |
+| Fixing bug                                                        | `tdd`               |
+| Implementing feature                                              | `tdd`               |
+| Modifying component                                               | `tdd`               |
+| Refactoring code                                                  | `tdd`               |
+| Renaming or removing a data-tour-id attribute value               | `prowler-tour`      |
+| Restructuring routes or layouts covered by a tour                 | `prowler-tour`      |
+| Review changelog format and conventions                           | `prowler-changelog` |
+| Testing hooks or utilities                                        | `vitest`            |
+| Update CHANGELOG.md in any component                              | `prowler-changelog` |
+| Using Zustand stores                                              | `zustand-5`         |
+| Working on Prowler UI structure (actions/adapters/types/hooks)    | `prowler-ui`        |
+| Working on task                                                   | `tdd`               |
+| Working with Prowler UI test helpers/pages                        | `prowler-test-ui`   |
+| Working with Tailwind classes                                     | `tailwind-4`        |
+| Writing Playwright E2E tests                                      | `playwright`        |
+| Writing Prowler UI E2E tests                                      | `prowler-test-ui`   |
+| Writing React component tests                                     | `vitest`            |
+| Writing React components                                          | `react-19`          |
+| Writing TypeScript types/interfaces                               | `typescript`        |
+| Writing Vitest tests                                              | `vitest`            |
+| Writing unit tests for UI                                         | `vitest`            |
 
 ---
 

ui/CHANGELOG.md

@@ -7,6 +7,32 @@ All notable changes to the **Prowler UI** are documented in this file.
 ### 🚀 Added
 
 - DORA compliance framework support [(#11131)](https://github.com/prowler-cloud/prowler/pull/11131)
+- Guided product onboarding for new users — step-by-step tours covering providers, scans, findings, compliance, and attack paths, replayable anytime from the info icon in the page header [(#11430)](https://github.com/prowler-cloud/prowler/pull/11430)
+
+---
+
+## [1.29.3] (Prowler UNRELEASED)
+
+### 🐞 Fixed
+
+- Finding drawer tabs now keep the active tab text and underline styling when tooltip state changes [(#11493)](https://github.com/prowler-cloud/prowler/pull/11493)
+
+---
+
+## [1.29.2] (Prowler v5.29.2)
+
+### 🔄 Changed
+
+- Account and provider-type selector triggers now show the provider icon, with a non-deduped icon stack [(#11424)](https://github.com/prowler-cloud/prowler/pull/11424)
+
+### 🐞 Fixed
+
+- Add Provider modal now closes without reloading the providers page [(#11424)](https://github.com/prowler-cloud/prowler/pull/11424)
+- Users page now shows the "Delete User" action only on the current user's row, matching the backend rule that a user can only delete their own account [(#11447)](https://github.com/prowler-cloud/prowler/pull/11447)
+
+### 🔐 Security
+
+- Vitest toolchain upgraded `4.0.18` → `4.1.8` to clear two critical `pnpm audit` advisories [(#11424)](https://github.com/prowler-cloud/prowler/pull/11424)
 
 ---
 

ui/__tests__/msw/handlers/attack-paths.ts

@@ -10,7 +10,7 @@ import type {
   QueryResultAttributes,
 } from "@/types/attack-paths";
 
-const API = process.env.NEXT_PUBLIC_API_BASE_URL!;
+const API = process.env.NEXT_PUBLIC_API_BASE_URL;
 
 type JsonApiErrorBody = {
   errors: Array<{ detail: string; status: string }>;

ui/actions/scans/scans.ts

@@ -15,6 +15,7 @@ import {
 } from "@/lib/provider-filters";
 import { addScanOperation } from "@/lib/sentry-breadcrumbs";
 import { handleApiError, handleApiResponse } from "@/lib/server-actions-helper";
+import { SCAN_STATES } from "@/types/attack-paths";
 
 const ORGANIZATION_SCAN_CONCURRENCY_LIMIT = 5;
 export const getScans = async ({
@@ -64,6 +65,10 @@ export const getScansByState = async () => {
     "filter[provider_type__in]",
     sanitizeProviderTypesCsv(),
   );
+  // Only need to know whether at least one completed scan exists; filter server-side
+  // and cap to a single row so the answer is correct regardless of total scan count.
+  url.searchParams.append("filter[state]", SCAN_STATES.COMPLETED);
+  url.searchParams.append("page[size]", "1");
 
   try {
     const response = await fetch(url.toString(), {

ui/app/(prowler)/_overview/_components/accounts-selector.test.tsx

@@ -1,4 +1,4 @@
-import { render, screen } from "@testing-library/react";
+import { render, screen, within } from "@testing-library/react";
 import userEvent from "@testing-library/user-event";
 import { describe, expect, it, vi } from "vitest";
 
@@ -57,7 +57,7 @@ vi.mock("@/components/shadcn/select/multiselect", () => ({
     );
   },
   MultiSelectTrigger: ({ children }: { children: React.ReactNode }) => (
-    <div>{children}</div>
+    <div data-testid="trigger">{children}</div>
   ),
   MultiSelectValue: ({ placeholder }: { placeholder: string }) => (
     <span>{placeholder}</span>
@@ -220,4 +220,45 @@ describe("AccountsSelector", () => {
 
     expect(multiSelectSpy).toHaveBeenLastCalledWith({ open: false });
   });
+
+  it("shows the provider icon next to the name in the trigger for a single selection", async () => {
+    render(
+      <AccountsSelector
+        providers={providers}
+        onBatchChange={vi.fn()}
+        selectedValues={["provider-1"]}
+      />,
+    );
+
+    const trigger = screen.getByTestId("trigger");
+    expect(await within(trigger).findByText("AWS")).toBeInTheDocument();
+    expect(within(trigger).getByText("Production AWS")).toBeInTheDocument();
+  });
+
+  it("renders one icon per selected account without deduping by provider type", async () => {
+    const secondAws = {
+      ...providers[0],
+      id: "provider-2",
+      attributes: {
+        ...providers[0].attributes,
+        uid: "999999999999",
+        alias: "Staging AWS",
+      },
+    };
+
+    render(
+      <AccountsSelector
+        providers={[providers[0], secondAws]}
+        onBatchChange={vi.fn()}
+        selectedValues={["provider-1", "provider-2"]}
+      />,
+    );
+
+    const trigger = screen.getByTestId("trigger");
+    // Two AWS accounts -> two AWS icons in the trigger (no dedupe).
+    expect(await within(trigger).findAllByText("AWS")).toHaveLength(2);
+    expect(
+      within(trigger).getByText("2 Providers selected"),
+    ).toBeInTheDocument();
+  });
 });

ui/app/(prowler)/_overview/_components/accounts-selector.tsx

@@ -1,26 +1,12 @@
 "use client";
 
 import { useSearchParams } from "next/navigation";
-import { ReactNode, useState } from "react";
+import { useState } from "react";
 
 import {
-  AlibabaCloudProviderBadge,
-  AWSProviderBadge,
-  AzureProviderBadge,
-  CloudflareProviderBadge,
-  GCPProviderBadge,
-  GitHubProviderBadge,
-  GoogleWorkspaceProviderBadge,
-  IacProviderBadge,
-  ImageProviderBadge,
-  KS8ProviderBadge,
-  M365ProviderBadge,
-  MongoDBAtlasProviderBadge,
-  OktaProviderBadge,
-  OpenStackProviderBadge,
-  OracleCloudProviderBadge,
-  VercelProviderBadge,
-} from "@/components/icons/providers-badge";
+  ProviderTypeIcon,
+  ProviderTypeIconStack,
+} from "@/components/icons/providers-badge/provider-type-icon";
 import { Badge } from "@/components/shadcn";
 import {
   MultiSelect,
@@ -45,25 +31,6 @@ const ACCOUNT_SELECTOR_FILTER = {
 type AccountSelectorFilter =
   (typeof ACCOUNT_SELECTOR_FILTER)[keyof typeof ACCOUNT_SELECTOR_FILTER];
 
-const PROVIDER_ICON: Record<ProviderType, ReactNode> = {
-  aws: <AWSProviderBadge width={18} height={18} />,
-  azure: <AzureProviderBadge width={18} height={18} />,
-  gcp: <GCPProviderBadge width={18} height={18} />,
-  kubernetes: <KS8ProviderBadge width={18} height={18} />,
-  m365: <M365ProviderBadge width={18} height={18} />,
-  github: <GitHubProviderBadge width={18} height={18} />,
-  googleworkspace: <GoogleWorkspaceProviderBadge width={18} height={18} />,
-  iac: <IacProviderBadge width={18} height={18} />,
-  image: <ImageProviderBadge width={18} height={18} />,
-  oraclecloud: <OracleCloudProviderBadge width={18} height={18} />,
-  mongodbatlas: <MongoDBAtlasProviderBadge width={18} height={18} />,
-  alibabacloud: <AlibabaCloudProviderBadge width={18} height={18} />,
-  cloudflare: <CloudflareProviderBadge width={18} height={18} />,
-  openstack: <OpenStackProviderBadge width={18} height={18} />,
-  vercel: <VercelProviderBadge width={18} height={18} />,
-  okta: <OktaProviderBadge width={18} height={18} />,
-};
-
 /** Common props shared by both batch and instant modes. */
 interface AccountsSelectorBaseProps {
   providers: ProviderProps[];
@@ -158,10 +125,36 @@ export function AccountsSelector({
     if (selectedIds.length === 1) {
       const p = providers.find((pr) => getProviderValue(pr) === selectedIds[0]);
       const name = p ? p.attributes.alias || p.attributes.uid : selectedIds[0];
-      return <span className="truncate">{name}</span>;
+      return (
+        <span className="flex min-w-0 items-center gap-2">
+          {p && (
+            <span aria-hidden="true">
+              <ProviderTypeIcon type={p.attributes.provider} />
+            </span>
+          )}
+          <span className="truncate">{name}</span>
+        </span>
+      );
     }
+    // One icon per selected account (no dedupe): two accounts of the same
+    // provider show two icons, disambiguated by the UID tooltip on hover.
+    const items = selectedIds
+      .map((selectedId) =>
+        providers.find((pr) => getProviderValue(pr) === selectedId),
+      )
+      .filter((p): p is ProviderProps => Boolean(p))
+      .map((p) => ({
+        key: p.id,
+        type: p.attributes.provider as ProviderType,
+        tooltip: p.attributes.uid,
+      }));
     return (
-      <span className="truncate">{selectedIds.length} Providers selected</span>
+      <span className="flex min-w-0 items-center gap-2">
+        <ProviderTypeIconStack items={items} />
+        <span className="truncate">
+          {selectedIds.length} Providers selected
+        </span>
+      </span>
     );
   };
 
@@ -208,7 +201,6 @@ export function AccountsSelector({
                 const isDisabled = disabledValuesSet.has(value);
                 const displayName = p.attributes.alias || p.attributes.uid;
                 const providerType = p.attributes.provider as ProviderType;
-                const icon = PROVIDER_ICON[providerType];
                 const searchKeywords = [
                   displayName,
                   p.attributes.alias,
@@ -228,7 +220,9 @@ export function AccountsSelector({
                       if (closeOnSelect) setSelectorOpen(false);
                     }}
                   >
-                    <span aria-hidden="true">{icon}</span>
+                    <span aria-hidden="true">
+                      <ProviderTypeIcon type={providerType} />
+                    </span>
                     <span className="flex min-w-0 flex-1 items-center gap-2">
                       <span className="truncate">{displayName}</span>
                       {isDisabled && <Badge variant="tag">Disconnected</Badge>}

ui/app/(prowler)/_overview/_components/provider-type-selector.test.tsx

@@ -1,4 +1,4 @@
-import { render, screen } from "@testing-library/react";
+import { render, screen, within } from "@testing-library/react";
 import { describe, expect, it, vi } from "vitest";
 
 import { ProviderTypeSelector } from "./provider-type-selector";
@@ -39,7 +39,7 @@ vi.mock("@/components/shadcn/select/multiselect", () => ({
     <div>{children}</div>
   ),
   MultiSelectTrigger: ({ children }: { children: React.ReactNode }) => (
-    <div>{children}</div>
+    <div data-testid="trigger">{children}</div>
   ),
   MultiSelectValue: ({ placeholder }: { placeholder: string }) => (
     <span>{placeholder}</span>
@@ -145,4 +145,26 @@ describe("ProviderTypeSelector", () => {
     ).toHaveAttribute("aria-disabled", "true");
     expect(screen.getByText("All selected")).toBeInTheDocument();
   });
+
+  it("shows one icon per selected type and a count in the trigger", async () => {
+    const azure = {
+      ...providers[0],
+      id: "provider-2",
+      attributes: { ...providers[0].attributes, provider: "azure" as const },
+    };
+
+    render(
+      <ProviderTypeSelector
+        providers={[providers[0], azure]}
+        onBatchChange={vi.fn()}
+        selectedValues={["aws", "azure"]}
+      />,
+    );
+
+    const trigger = screen.getByTestId("trigger");
+    expect(await within(trigger).findByText("AWS")).toBeInTheDocument();
+    expect(
+      within(trigger).getByText("2 Provider Types selected"),
+    ).toBeInTheDocument();
+  });
 });

ui/app/(prowler)/_overview/_components/provider-type-selector.tsx

@@ -1,8 +1,12 @@
 "use client";
 
 import { useSearchParams } from "next/navigation";
-import { type ComponentType, lazy, Suspense } from "react";
 
+import {
+  PROVIDER_TYPE_DATA,
+  ProviderTypeIcon,
+  ProviderTypeIconStack,
+} from "@/components/icons/providers-badge/provider-type-icon";
 import {
   MultiSelect,
   MultiSelectContent,
@@ -14,163 +18,6 @@ import {
 import { useUrlFilters } from "@/hooks/use-url-filters";
 import { type ProviderProps, ProviderType } from "@/types/providers";
 
-const AWSProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.AWSProviderBadge,
-  })),
-);
-const AzureProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.AzureProviderBadge,
-  })),
-);
-const GCPProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.GCPProviderBadge,
-  })),
-);
-const KS8ProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.KS8ProviderBadge,
-  })),
-);
-const M365ProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.M365ProviderBadge,
-  })),
-);
-const GitHubProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.GitHubProviderBadge,
-  })),
-);
-const IacProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.IacProviderBadge,
-  })),
-);
-const ImageProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.ImageProviderBadge,
-  })),
-);
-const OracleCloudProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.OracleCloudProviderBadge,
-  })),
-);
-const MongoDBAtlasProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.MongoDBAtlasProviderBadge,
-  })),
-);
-const AlibabaCloudProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.AlibabaCloudProviderBadge,
-  })),
-);
-const CloudflareProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.CloudflareProviderBadge,
-  })),
-);
-const OpenStackProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.OpenStackProviderBadge,
-  })),
-);
-const GoogleWorkspaceProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.GoogleWorkspaceProviderBadge,
-  })),
-);
-const VercelProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.VercelProviderBadge,
-  })),
-);
-const OktaProviderBadge = lazy(() =>
-  import("@/components/icons/providers-badge").then((m) => ({
-    default: m.OktaProviderBadge,
-  })),
-);
-
-type IconProps = { width: number; height: number };
-
-const IconPlaceholder = ({ width, height }: IconProps) => (
-  <div style={{ width, height }} />
-);
-
-const PROVIDER_DATA: Record<
-  ProviderType,
-  { label: string; icon: ComponentType<IconProps> }
-> = {
-  aws: {
-    label: "Amazon Web Services",
-    icon: AWSProviderBadge,
-  },
-  azure: {
-    label: "Microsoft Azure",
-    icon: AzureProviderBadge,
-  },
-  gcp: {
-    label: "Google Cloud Platform",
-    icon: GCPProviderBadge,
-  },
-  kubernetes: {
-    label: "Kubernetes",
-    icon: KS8ProviderBadge,
-  },
-  m365: {
-    label: "Microsoft 365",
-    icon: M365ProviderBadge,
-  },
-  github: {
-    label: "GitHub",
-    icon: GitHubProviderBadge,
-  },
-  googleworkspace: {
-    label: "Google Workspace",
-    icon: GoogleWorkspaceProviderBadge,
-  },
-  iac: {
-    label: "Infrastructure as Code",
-    icon: IacProviderBadge,
-  },
-  image: {
-    label: "Container Registry",
-    icon: ImageProviderBadge,
-  },
-  oraclecloud: {
-    label: "Oracle Cloud Infrastructure",
-    icon: OracleCloudProviderBadge,
-  },
-  mongodbatlas: {
-    label: "MongoDB Atlas",
-    icon: MongoDBAtlasProviderBadge,
-  },
-  alibabacloud: {
-    label: "Alibaba Cloud",
-    icon: AlibabaCloudProviderBadge,
-  },
-  cloudflare: {
-    label: "Cloudflare",
-    icon: CloudflareProviderBadge,
-  },
-  openstack: {
-    label: "OpenStack",
-    icon: OpenStackProviderBadge,
-  },
-  vercel: {
-    label: "Vercel",
-    icon: VercelProviderBadge,
-  },
-  okta: {
-    label: "Okta",
-    icon: OktaProviderBadge,
-  },
-};
-
 /** Common props shared by both batch and instant modes. */
 interface ProviderTypeSelectorBaseProps {
   providers: ProviderProps[];
@@ -247,34 +94,38 @@ export const ProviderTypeSelector = ({
         .map((p) => p.attributes.provider),
     ),
   )
-    .filter((type): type is ProviderType => type in PROVIDER_DATA)
+    .filter((type): type is ProviderType => type in PROVIDER_TYPE_DATA)
     .sort((a, b) =>
-      PROVIDER_DATA[a].label.localeCompare(PROVIDER_DATA[b].label),
+      PROVIDER_TYPE_DATA[a].label.localeCompare(PROVIDER_TYPE_DATA[b].label),
     );
 
-  const renderIcon = (providerType: ProviderType) => {
-    const IconComponent = PROVIDER_DATA[providerType].icon;
-    return (
-      <Suspense fallback={<IconPlaceholder width={24} height={24} />}>
-        <IconComponent width={24} height={24} />
-      </Suspense>
-    );
-  };
-
   const selectedLabel = () => {
     if (selectedTypes.length === 0) return null;
     if (selectedTypes.length === 1) {
       const providerType = selectedTypes[0] as ProviderType;
       return (
         <span className="flex min-w-0 items-center gap-2">
-          {renderIcon(providerType)}
-          <span className="truncate">{PROVIDER_DATA[providerType].label}</span>
+          <span aria-hidden="true">
+            <ProviderTypeIcon type={providerType} />
+          </span>
+          <span className="truncate">
+            {PROVIDER_TYPE_DATA[providerType].label}
+          </span>
         </span>
       );
     }
     return (
-      <span className="min-w-0 truncate">
-        {selectedTypes.length} Provider Types selected
+      <span className="flex min-w-0 items-center gap-2">
+        <ProviderTypeIconStack
+          items={(selectedTypes as ProviderType[]).map((type) => ({
+            key: type,
+            type,
+            tooltip: PROVIDER_TYPE_DATA[type].label,
+          }))}
+        />
+        <span className="min-w-0 truncate">
+          {selectedTypes.length} Provider Types selected
+        </span>
       </span>
     );
   };
@@ -329,12 +180,17 @@ export const ProviderTypeSelector = ({
                 <MultiSelectItem
                   key={providerType}
                   value={providerType}
-                  badgeLabel={PROVIDER_DATA[providerType].label}
-                  keywords={[providerType, PROVIDER_DATA[providerType].label]}
-                  aria-label={`${PROVIDER_DATA[providerType].label} Provider Type`}
+                  badgeLabel={PROVIDER_TYPE_DATA[providerType].label}
+                  keywords={[
+                    providerType,
+                    PROVIDER_TYPE_DATA[providerType].label,
+                  ]}
+                  aria-label={`${PROVIDER_TYPE_DATA[providerType].label} Provider Type`}
                 >
-                  <span aria-hidden="true">{renderIcon(providerType)}</span>
-                  <span>{PROVIDER_DATA[providerType].label}</span>
+                  <span aria-hidden="true">
+                    <ProviderTypeIcon type={providerType} size={24} />
+                  </span>
+                  <span>{PROVIDER_TYPE_DATA[providerType].label}</span>
                 </MultiSelectItem>
               ))}
             </>

ui/app/(prowler)/attack-paths/(workflow)/query-builder/_components/scan-list-table.test.tsx

@@ -58,6 +58,7 @@ vi.mock("@/components/ui/table", () => ({
     data,
     metadata,
     controlledPage,
+    getRowAttributes,
   }: {
     columns: Array<{
       id?: string;
@@ -74,6 +75,10 @@ vi.mock("@/components/ui/table", () => ({
       };
     };
     controlledPage: number;
+    getRowAttributes?: (row: {
+      index: number;
+      original: AttackPathScan;
+    }) => Record<string, string | undefined>;
   }) => (
     <div>
       <span>{metadata.pagination.count} Total Entries</span>
@@ -95,8 +100,8 @@ vi.mock("@/components/ui/table", () => ({
           </tr>
         </thead>
         <tbody>
-          {data.map((row) => (
-            <tr key={row.id}>
+          {data.map((row, index) => (
+            <tr key={row.id} {...getRowAttributes?.({ index, original: row })}>
               {columns.map((column, index) => (
                 <td key={column.id ?? index}>
                   {column.cell
@@ -176,6 +181,20 @@ describe("ScanListTable", () => {
     );
   });
 
+  it("anchors the attack paths scan tour to the first visible scan row", () => {
+    render(
+      <ScanListTable scans={[createScan(1), createScan(2), createScan(3)]} />,
+    );
+
+    const firstRow = screen
+      .getAllByRole("radio", {
+        name: "Select scan",
+      })[0]
+      .closest("tr");
+
+    expect(firstRow).toHaveAttribute("data-tour-id", "attack-paths-scan-list");
+  });
+
   it("enables the radio button for a failed scan when graph data is ready", async () => {
     const user = userEvent.setup();
     const failedScan: AttackPathScan = {

ui/app/(prowler)/attack-paths/(workflow)/query-builder/_components/scan-list-table.tsx

@@ -295,6 +295,9 @@ export const ScanListTable = ({ scans }: ScanListTableProps) => {
             handleSelectScan(row.original.id);
           }
         }}
+        getRowAttributes={(row) =>
+          row.index === 0 ? { "data-tour-id": "attack-paths-scan-list" } : {}
+        }
         enableRowSelection
         rowSelection={getSelectedRowSelection(paginatedScans, selectedScanId)}
       />

ui/app/(prowler)/attack-paths/(workflow)/query-builder/_hooks/index.ts

@@ -1,3 +1,4 @@
+export { useAttackPathScans } from "./use-attack-path-scans";
 export { useGraphState } from "./use-graph-state";
 export { useQueryBuilder } from "./use-query-builder";
 export { useWizardState } from "./use-wizard-state";

ui/app/(prowler)/attack-paths/(workflow)/query-builder/_hooks/use-attack-path-scans.ts

@@ -0,0 +1,81 @@
+"use client";
+
+import { useState } from "react";
+
+import { getAttackPathScans } from "@/actions/attack-paths";
+import { useMountEffect } from "@/hooks/use-mount-effect";
+import type { AttackPathScan } from "@/types/attack-paths";
+
+export interface UseAttackPathScansOptions {
+  /**
+   * Invoked once the initial load resolves with no scan whose graph data is
+   * ready (including empty results or a fetch failure). The page passes a
+   * redirect only during onboarding replay; an established user gets `undefined`
+   * and stays on the page.
+   */
+  onNoReadyScan?: () => void;
+}
+
+export interface UseAttackPathScansResult {
+  scans: AttackPathScan[];
+  scansLoading: boolean;
+  refreshScans: () => Promise<void>;
+}
+
+/**
+ * `useData`-style hook owning the Attack Paths scan list. The direct
+ * `useEffect` (via `useMountEffect`) lives here, not in the component: the
+ * project forbids `useEffect` in components, but a reusable data hook is the
+ * sanctioned place for a mount-time fetch when no fetching library is wired up.
+ */
+export function useAttackPathScans(
+  options: UseAttackPathScansOptions = {},
+): UseAttackPathScansResult {
+  const { onNoReadyScan } = options;
+
+  const [scans, setScans] = useState<AttackPathScan[]>([]);
+  const [scansLoading, setScansLoading] = useState(true);
+
+  const refreshScans = async () => {
+    try {
+      const scansData = await getAttackPathScans();
+      if (scansData?.data) {
+        setScans(scansData.data);
+      }
+    } catch (error) {
+      console.error("Failed to refresh scans:", error);
+    }
+  };
+
+  useMountEffect(() => {
+    let active = true;
+
+    const loadScans = async () => {
+      setScansLoading(true);
+      try {
+        const scansData = await getAttackPathScans();
+        const nextScans = scansData?.data ?? [];
+        if (!active) return;
+        setScans(nextScans);
+        if (!nextScans.some((scan) => scan.attributes.graph_data_ready)) {
+          onNoReadyScan?.();
+        }
+      } catch (error) {
+        if (!active) return;
+        console.error("Failed to load scans:", error);
+        setScans([]);
+        onNoReadyScan?.();
+      } finally {
+        if (active) setScansLoading(false);
+      }
+    };
+
+    void loadScans();
+
+    return () => {
+      active = false;
+    };
+  });
+
+  return { scans, scansLoading, refreshScans };
+}

ui/app/(prowler)/attack-paths/(workflow)/query-builder/attack-paths-page.test.tsx

@@ -1,19 +0,0 @@
-import { readFileSync } from "node:fs";
-import path from "node:path";
-import { fileURLToPath } from "node:url";
-
-import { describe, expect, it } from "vitest";
-
-describe("AttackPathsPage", () => {
-  const currentDir = path.dirname(fileURLToPath(import.meta.url));
-  const filePath = path.join(currentDir, "attack-paths-page.tsx");
-  const source = readFileSync(filePath, "utf8");
-
-  it("keeps the page description without rendering a duplicate Attack Paths heading", () => {
-    // Then
-    expect(source).not.toContain(">\n          Attack Paths\n        </h2>");
-    expect(source).toContain(
-      "Select a scan, build a query, and visualize Attack Paths in your",
-    );
-  });
-});

ui/app/(prowler)/attack-paths/(workflow)/query-builder/attack-paths-page.tsx

@@ -2,7 +2,7 @@
 
 import { ArrowLeft, Info, Maximize2 } from "lucide-react";
 import Link from "next/link";
-import { useSearchParams } from "next/navigation";
+import { usePathname, useRouter, useSearchParams } from "next/navigation";
 import { Suspense, useEffect, useRef, useState } from "react";
 import { FormProvider } from "react-hook-form";
 
@@ -10,11 +10,11 @@ import {
   buildAttackPathQueries,
   executeCustomQuery,
   executeQuery,
-  getAttackPathScans,
   getAvailableQueries,
 } from "@/actions/attack-paths";
 import { adaptQueryResultToGraphData } from "@/actions/attack-paths/query-result.adapter";
 import { FindingDetailDrawer } from "@/components/findings/table";
+import { PageReady } from "@/components/onboarding";
 import { useFindingDetails } from "@/components/resources/table/use-finding-details";
 import { AutoRefresh } from "@/components/scans";
 import {
@@ -32,10 +32,19 @@ import {
   DialogTrigger,
 } from "@/components/shadcn/dialog";
 import { useToast } from "@/components/ui";
+import { useMountEffect } from "@/hooks/use-mount-effect";
+import { isCloud } from "@/lib/shared/env";
+import {
+  attackPathsTour,
+  type AttackPathsTourTarget,
+  pickDemoQuery,
+  pickDemoScan,
+} from "@/lib/tours/attack-paths.tour";
+import { attackPathsEmptyTour } from "@/lib/tours/attack-paths-empty.tour";
+import { useDriverTour } from "@/lib/tours/use-driver-tour";
 import type {
   AttackPathQuery,
   AttackPathQueryError,
-  AttackPathScan,
   GraphNode,
 } from "@/types/attack-paths";
 import { ATTACK_PATH_QUERY_IDS, SCAN_STATES } from "@/types/attack-paths";
@@ -53,23 +62,30 @@ import {
   ScanListTable,
 } from "./_components";
 import type { GraphHandle } from "./_components/graph/attack-path-graph";
+import { useAttackPathScans } from "./_hooks/use-attack-path-scans";
 import { useGraphState } from "./_hooks/use-graph-state";
 import { useQueryBuilder } from "./_hooks/use-query-builder";
 import { exportGraphAsPNG } from "./_lib";
 
-/**
- * Attack Paths
- * Allows users to select a scan, build a query, and visualize the attack path graph
- */
 export default function AttackPathsPage() {
   const searchParams = useSearchParams();
+  const pathname = usePathname();
+  const router = useRouter();
   const scanId = searchParams.get("scanId");
+  // Onboarding tours are Cloud-only.
+  const onboardingEnabled = isCloud();
+  const isAttackPathsReplay =
+    onboardingEnabled && searchParams.get("onboarding") === "attack-paths";
   const graphState = useGraphState();
   const finding = useFindingDetails();
   const { toast } = useToast();
 
-  const [scansLoading, setScansLoading] = useState(true);
-  const [scans, setScans] = useState<AttackPathScan[]>([]);
+  const { scans, scansLoading, refreshScans } = useAttackPathScans({
+    onNoReadyScan: isAttackPathsReplay
+      ? () => router.push("/scans?onboarding=view-first-scan")
+      : undefined,
+  });
+
   const [queriesLoading, setQueriesLoading] = useState(true);
   const [queriesError, setQueriesError] = useState<string | null>(null);
   const [isFullscreenOpen, setIsFullscreenOpen] = useState(false);
@@ -81,10 +97,62 @@ export default function AttackPathsPage() {
 
   const [queries, setQueries] = useState<AttackPathQuery[]>([]);
 
-  // Use custom hook for query builder form state and validation
   const queryBuilder = useQueryBuilder(queries);
 
-  // Reset graph state when component mounts
+  const hasReadyScan = scans.some((scan) => scan.attributes.graph_data_ready);
+  const hasNoScans = scans.length === 0;
+
+  useDriverTour(attackPathsEmptyTour, {
+    enabled: onboardingEnabled && !scansLoading && hasNoScans,
+  });
+
+  const { start: startAttackPathsTour } = useDriverTour<AttackPathsTourTarget>(
+    attackPathsTour,
+    {
+      enabled: onboardingEnabled && !scansLoading && hasReadyScan,
+      autoOpen: !isAttackPathsReplay,
+      // Page owns tour auto-open; OnboardingSequenceBanner is the sole Continue/Skip control.
+      // pickDemoScan/pickDemoQuery policy lives in attack-paths.tour.ts.
+      stepHandlers: {
+        "scan-list": {
+          onNext: async ({ waitForStep }) => {
+            const selected = pickDemoScan(scans);
+            if (!selected) return;
+            const params = new URLSearchParams(searchParams.toString());
+            params.set("scanId", selected.id);
+            router.push(`${pathname}?${params.toString()}`);
+            await waitForStep("query-selector");
+          },
+        },
+        "query-selector": {
+          onNext: async ({ waitForStep }) => {
+            const selected = pickDemoQuery(queries);
+            if (!selected) return;
+            queryBuilder.handleQueryChange(selected.id);
+            await waitForStep("execute-button");
+          },
+        },
+      },
+    },
+  );
+
+  // Onboarding replay entry: start the tour once and strip the `onboarding`
+  // param. Invoked from <AttackPathsReplayTrigger>, which mounts only when the
+  // replay conditions hold — so `useMountEffect` fires it exactly once and the
+  // old `replayStartedRef` run-once guard is gone.
+  const startAttackPathsReplay = () => {
+    startAttackPathsTour();
+
+    const params = new URLSearchParams(searchParams.toString());
+    params.delete("onboarding");
+    const query = params.toString();
+    window.history.replaceState(
+      null,
+      "",
+      query ? `${pathname}?${query}` : pathname,
+    );
+  };
+
   useEffect(() => {
     if (!hasResetRef.current) {
       hasResetRef.current = true;
@@ -92,60 +160,22 @@ export default function AttackPathsPage() {
     }
   }, [graphState]);
 
-  // Reset graph state when scan changes
   useEffect(() => {
     graphState.resetGraph();
   }, [scanId]); // eslint-disable-line react-hooks/exhaustive-deps -- reset on scanId change only
 
-  // Load available scans on mount
-  useEffect(() => {
-    const loadScans = async () => {
-      setScansLoading(true);
-      try {
-        const scansData = await getAttackPathScans();
-        if (scansData?.data) {
-          setScans(scansData.data);
-        } else {
-          setScans([]);
-        }
-      } catch (error) {
-        console.error("Failed to load scans:", error);
-        setScans([]);
-      } finally {
-        setScansLoading(false);
-      }
-    };
-
-    loadScans();
-  }, []);
-
-  // Check if there's an executing scan for auto-refresh
   const hasExecutingScan = scans.some(
     (scan) =>
       scan.attributes.state === SCAN_STATES.EXECUTING ||
       scan.attributes.state === SCAN_STATES.SCHEDULED,
   );
 
-  // Detect if the selected scan is showing data from a previous cycle
   const selectedScan = scans.find((scan) => scan.id === scanId);
   const isViewingPreviousCycleData =
     selectedScan &&
     selectedScan.attributes.graph_data_ready &&
     selectedScan.attributes.state !== SCAN_STATES.COMPLETED;
 
-  // Callback to refresh scans (used by AutoRefresh component)
-  const refreshScans = async () => {
-    try {
-      const scansData = await getAttackPathScans();
-      if (scansData?.data) {
-        setScans(scansData.data);
-      }
-    } catch (error) {
-      console.error("Failed to refresh scans:", error);
-    }
-  };
-
-  // Load available queries on mount
   useEffect(() => {
     const loadQueries = async () => {
       if (!scanId) {
@@ -205,7 +235,6 @@ export default function AttackPathsPage() {
       return;
     }
 
-    // Validate form before executing query
     const isValid = await queryBuilder.form.trigger();
     if (!isValid) {
       showErrorToast(
@@ -257,7 +286,6 @@ export default function AttackPathsPage() {
           variant: "default",
         });
 
-        // Scroll to graph after successful query execution
         setTimeout(() => {
           graphContainerRef.current?.scrollIntoView({
             behavior: "smooth",
@@ -297,13 +325,9 @@ export default function AttackPathsPage() {
       }
 
       findingNavigationInFlightRef.current = true;
-      // Findings skip the intermediate node-details modal. The finding drawer
-      // is the useful destination, so open it directly from the graph click.
+      // Open finding drawer directly, bypassing the node-details modal.
       graphState.enterFilteredView(node.id);
-      // enterFilteredView stores the filtered node as selected so the graph can
-      // highlight it. Clear the selection right after for findings so the node
-      // details modal does not open before the finding drawer.
-      graphState.selectNode(null);
+      graphState.selectNode(null); // clear so node-details modal doesn't open first
       void handleViewFinding(String(node.properties?.id || node.id));
       return;
     }
@@ -368,14 +392,19 @@ export default function AttackPathsPage() {
 
   return (
     <div className="flex flex-col gap-6">
-      {/* Auto-refresh scans when there's an executing scan */}
       <AutoRefresh
         hasExecutingScan={hasExecutingScan}
         onRefresh={refreshScans}
       />
 
-      {/* Page introduction */}
-      <div>
+      {isAttackPathsReplay && !scansLoading && hasReadyScan && (
+        <AttackPathsReplayTrigger onReplay={startAttackPathsReplay} />
+      )}
+
+      {/* Enables the navbar replay icon once the initial scan load resolves. */}
+      {!scansLoading && <PageReady />}
+
+      <div data-tour-id="attack-paths-intro">
         <p className="text-text-neutral-secondary text-sm">
           Select a scan, build a query, and visualize Attack Paths in your
           infrastructure.
@@ -390,27 +419,27 @@ export default function AttackPathsPage() {
         <div className="minimal-scrollbar rounded-large shadow-small border-border-neutral-secondary bg-bg-neutral-secondary relative z-0 flex w-full flex-col gap-4 overflow-auto border p-4">
           <p className="text-sm">Loading scans...</p>
         </div>
-      ) : scans.length === 0 ? (
-        <Alert variant="info">
-          <Info className="size-4" />
-          <AlertTitle>No scans available</AlertTitle>
-          <AlertDescription>
-            <span>
-              You need to run a scan before you can analyze attack paths.{" "}
-              <Link href="/scans" className="font-medium underline">
-                Go to Scan Jobs
-              </Link>
-            </span>
-          </AlertDescription>
-        </Alert>
+      ) : hasNoScans ? (
+        <div data-tour-id="attack-paths-empty-scans-cta">
+          <Alert variant="info">
+            <Info className="size-4" />
+            <AlertTitle>No scans available</AlertTitle>
+            <AlertDescription>
+              <span>
+                You need to run a scan before you can analyze attack paths.{" "}
+                <Link href="/scans" className="font-medium underline">
+                  Go to Scan Jobs
+                </Link>
+              </span>
+            </AlertDescription>
+          </Alert>
+        </div>
       ) : (
         <>
-          {/* Scans Table */}
           <Suspense fallback={<div>Loading scans...</div>}>
             <ScanListTable scans={scans} />
           </Suspense>
 
-          {/* Banner: viewing data from a previous scan cycle */}
           {isViewingPreviousCycleData && (
             <Alert variant="info">
               <Info className="size-4" />
@@ -425,7 +454,6 @@ export default function AttackPathsPage() {
             </Alert>
           )}
 
-          {/* Query Builder Section - shown only after selecting a scan */}
           {scanId && (
             <div className="minimal-scrollbar rounded-large shadow-small border-border-neutral-secondary bg-bg-neutral-secondary relative z-0 flex w-full flex-col gap-4 overflow-auto border p-4">
               {queriesLoading ? (
@@ -438,11 +466,13 @@ export default function AttackPathsPage() {
               ) : (
                 <>
                   <FormProvider {...queryBuilder.form}>
-                    <QuerySelector
-                      queries={queries}
-                      selectedQueryId={queryBuilder.selectedQuery}
-                      onQueryChange={queryBuilder.handleQueryChange}
-                    />
+                    <div data-tour-id="attack-paths-query-selector">
+                      <QuerySelector
+                        queries={queries}
+                        selectedQueryId={queryBuilder.selectedQuery}
+                        onQueryChange={queryBuilder.handleQueryChange}
+                      />
+                    </div>
 
                     {queryBuilder.selectedQueryData && (
                       <QueryDescription
@@ -457,7 +487,10 @@ export default function AttackPathsPage() {
                     )}
                   </FormProvider>
 
-                  <div className="flex justify-end gap-3">
+                  <div
+                    data-tour-id="attack-paths-execute-button"
+                    className="flex justify-end gap-3"
+                  >
                     <ExecuteButton
                       isLoading={graphState.loading}
                       isDisabled={
@@ -476,7 +509,6 @@ export default function AttackPathsPage() {
             </div>
           )}
 
-          {/* Graph Visualization (Full Width) */}
           {(graphState.loading ||
             (graphState.data &&
               graphState.data.nodes &&
@@ -488,7 +520,6 @@ export default function AttackPathsPage() {
                 graphState.data.nodes &&
                 graphState.data.nodes.length > 0 ? (
                 <>
-                  {/* Info message and controls */}
                   <div className="flex flex-col gap-3 sm:flex-row sm:items-center sm:justify-between">
                     {graphState.isFilteredView ? (
                       <div className="flex items-center gap-3">
@@ -537,7 +568,6 @@ export default function AttackPathsPage() {
                       </div>
                     )}
 
-                    {/* Graph controls and fullscreen button together */}
                     <div className="flex items-center gap-2">
                       <GraphControls
                         onZoomIn={() => graphRef.current?.zoomIn()}
@@ -546,7 +576,6 @@ export default function AttackPathsPage() {
                         onExport={() => handleGraphExport("main")}
                       />
 
-                      {/* Fullscreen button */}
                       <div className="border-border-neutral-primary bg-bg-neutral-tertiary flex gap-1 rounded-lg border p-1">
                         <Dialog
                           open={isFullscreenOpen}
@@ -604,7 +633,6 @@ export default function AttackPathsPage() {
                     </div>
                   </div>
 
-                  {/* Graph in the middle */}
                   <div
                     ref={graphContainerRef}
                     className="h-[calc(100vh-22rem)]"
@@ -619,7 +647,6 @@ export default function AttackPathsPage() {
                     />
                   </div>
 
-                  {/* Legend below */}
                   <div className="flex justify-center overflow-x-auto">
                     <GraphLegend
                       data={graphState.data}
@@ -647,3 +674,26 @@ export default function AttackPathsPage() {
     </div>
   );
 }
+
+interface AttackPathsReplayTriggerProps {
+  onReplay: () => void;
+}
+
+// Conditional-mount trigger: the parent renders this only when the replay
+// should start. The microtask keeps driver.js/flushSync outside React's
+// mount lifecycle while still running before the next browser task.
+function AttackPathsReplayTrigger({ onReplay }: AttackPathsReplayTriggerProps) {
+  useMountEffect(() => {
+    let cancelled = false;
+
+    queueMicrotask(() => {
+      if (!cancelled) onReplay();
+    });
+
+    return () => {
+      cancelled = true;
+    };
+  });
+
+  return null;
+}

ui/app/(prowler)/attack-paths/layout.tsx

@@ -6,7 +6,11 @@ export default function AttackPathsLayout({
   children: React.ReactNode;
 }) {
   return (
-    <ContentLayout title="Attack Paths" icon="lucide:git-branch">
+    <ContentLayout
+      title="Attack Paths"
+      icon="lucide:git-branch"
+      onboardingAction={{ flowId: "attack-paths" }}
+    >
       {children}
     </ContentLayout>
   );

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

@@ -46,16 +46,26 @@ export default async function Compliance({
   });
 
   if (!scansData?.data) {
-    return <NoScansAvailable />;
+    return (
+      <ContentLayout
+        title="Compliance"
+        icon="lucide:shield-check"
+        onboardingAction={{
+          flowId: "view-compliance",
+          fallbackFlowId: "view-first-scan",
+          useFallback: true,
+        }}
+      >
+        <NoScansAvailable />
+      </ContentLayout>
+    );
   }
 
-  // Process scans with provider information from included data
   const expandedScansData: ExpandedScanData[] = scansData.data
     .filter((scan: ScanProps) => scan.relationships?.provider?.data?.id)
     .map((scan: ScanProps) => {
       const providerId = scan.relationships!.provider!.data!.id;
 
-      // Find the provider data in the included array
       const providerData = scansData.included?.find(
         (item: { type: string; id: string }) =>
           item.type === "providers" && item.id === providerId,
@@ -76,15 +86,20 @@ export default async function Compliance({
     })
     .filter(Boolean) as ExpandedScanData[];
 
-  // Use scanId from URL, or select the first scan if not provided
   const scanIdParam = resolvedSearchParams.scanId;
   const scanIdFromUrl = Array.isArray(scanIdParam)
     ? scanIdParam[0]
     : scanIdParam;
   const selectedScanId: string | null =
     scanIdFromUrl || expandedScansData[0]?.id || null;
+  const onboardingAction = selectedScanId
+    ? { flowId: "view-compliance" }
+    : {
+        flowId: "view-compliance",
+        fallbackFlowId: "view-first-scan",
+        useFallback: true,
+      };
 
-  // Find the selected scan
   const selectedScan = expandedScansData.find(
     (scan) => scan.id === selectedScanId,
   );
@@ -100,7 +115,6 @@ export default async function Compliance({
       }
     : undefined;
 
-  // Fetch metadata if we have a selected scan
   const metadataInfoData = selectedScanId
     ? await getComplianceOverviewMetadataInfo({
         filters: {
@@ -111,7 +125,6 @@ export default async function Compliance({
 
   const uniqueRegions = metadataInfoData?.data?.attributes?.regions || [];
 
-  // Fetch ThreatScore data from API if we have a selected scan
   let threatScoreData = null;
   if (selectedScanId && typeof selectedScanId === "string") {
     const threatScoreResponse = await getThreatScore({
@@ -128,10 +141,13 @@ export default async function Compliance({
   }
 
   return (
-    <ContentLayout title="Compliance" icon="lucide:shield-check">
+    <ContentLayout
+      title="Compliance"
+      icon="lucide:shield-check"
+      onboardingAction={onboardingAction}
+    >
       {selectedScanId ? (
         <>
-          {/* Row 1: Filters */}
           <div className="mb-6">
             <ComplianceFilters
               scans={expandedScansData}
@@ -140,7 +156,6 @@ export default async function Compliance({
             />
           </div>
 
-          {/* Row 2: ThreatScore card — full width, horizontal */}
           {threatScoreData &&
             typeof selectedScanId === "string" &&
             selectedScan && (
@@ -155,7 +170,6 @@ export default async function Compliance({
               </div>
             )}
 
-          {/* Row 3: Compliance grid with client-side search */}
           <Suspense
             key={searchParamsKey}
             fallback={
@@ -189,7 +203,6 @@ const SSRComplianceGrid = async ({
 }) => {
   const regionFilter = searchParams["filter[region__in]"]?.toString() || "";
 
-  // Only fetch compliance data if we have a valid scanId
   const compliancesData =
     scanId && scanId.trim() !== ""
       ? await getCompliancesOverview({
@@ -207,7 +220,6 @@ const SSRComplianceGrid = async ({
       a.attributes.framework.localeCompare(b.attributes.framework),
     );
 
-  // Check if the response contains no data
   if (
     !compliancesData ||
     !compliancesData.data ||
@@ -225,7 +237,6 @@ const SSRComplianceGrid = async ({
     );
   }
 
-  // Handle errors returned by the API
   if (compliancesData?.errors?.length > 0) {
     return (
       <Alert variant="info">
@@ -235,10 +246,7 @@ const SSRComplianceGrid = async ({
     );
   }
 
-  // Compute the set of latest CIS variants per provider once, so each card
-  // can gate its PDF button without re-parsing on every render. The backend
-  // only generates a CIS PDF for the latest version per provider, so any
-  // other CIS card must not expose the PDF download button.
+  // Backend only generates CIS PDFs for the latest version per provider.
   const latestCisIds = pickLatestCisPerProvider(
     compliancesData.data.map(
       (compliance: ComplianceOverviewData) => compliance.id,

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

@@ -59,7 +59,6 @@ export default async function Findings({
     filters: resolvedFilters,
   });
 
-  // Extract unique regions, services, categories, groups from the new endpoint
   const uniqueRegions = metadataInfoData?.data?.attributes?.regions || [];
   const uniqueServices = metadataInfoData?.data?.attributes?.services || [];
   const uniqueResourceTypes =
@@ -67,7 +66,6 @@ export default async function Findings({
   const uniqueCategories = metadataInfoData?.data?.attributes?.categories || [];
   const uniqueGroups = metadataInfoData?.data?.attributes?.groups || [];
 
-  // Extract scan UUIDs with "completed" state and more than one resource
   const completedScans = scansData?.data?.filter(
     (scan: ScanProps) =>
       scan.attributes.state === "completed" &&
@@ -76,6 +74,14 @@ export default async function Findings({
 
   const completedScanIds =
     completedScans?.map((scan: ScanProps) => scan.id) || [];
+  const onboardingAction =
+    completedScanIds.length > 0
+      ? { flowId: "explore-findings" }
+      : {
+          flowId: "explore-findings",
+          fallbackFlowId: "view-first-scan",
+          useFallback: true,
+        };
 
   const scanDetails = createScanDetailsMapping(
     completedScans || [],
@@ -84,7 +90,11 @@ export default async function Findings({
   const alertsEnabled = process.env.NEXT_PUBLIC_IS_CLOUD_ENV === "true";
 
   return (
-    <ContentLayout title="Findings" icon="lucide:tag">
+    <ContentLayout
+      title="Findings"
+      icon="lucide:tag"
+      onboardingAction={onboardingAction}
+    >
       <FilterTransitionWrapper>
         <div className="mb-6">
           <FindingsFilters
@@ -146,9 +156,8 @@ const SSRDataTable = async ({
     pageSize,
   });
 
-  // Transform API response to FindingGroupRow[]
   const groups = adaptFindingGroupsResponse(findingGroupsData);
-  // Key resets all client state (selection, drill-down) when data changes
+  // Key resets client state (selection, drill-down) when data changes.
   const groupKey = groups.map((g) => g.id).join(",");
 
   return (

ui/app/(prowler)/layout.tsx

@@ -2,16 +2,24 @@ import "@/styles/globals.css";
 
 import * as Sentry from "@sentry/nextjs";
 import { Metadata, Viewport } from "next";
-import { ReactNode } from "react";
+import { ReactNode, Suspense } from "react";
 
 import { getProviders } from "@/actions/providers";
+import { getScansByState } from "@/actions/scans/scans";
+import {
+  OnboardingCheckpointWatcher,
+  OnboardingGate,
+  OnboardingSequenceBanner,
+} from "@/components/onboarding";
 import MainLayout from "@/components/ui/main-layout/main-layout";
 import { NavigationProgress } from "@/components/ui/navigation-progress";
 import { Toaster } from "@/components/ui/toast";
 import { fontSans } from "@/config/fonts";
 import { siteConfig } from "@/config/site";
+import { isCloud } from "@/lib/shared/env";
 import { cn } from "@/lib/utils";
 import { StoreInitializer } from "@/store/ui/store-initializer";
+import { SCAN_STATES } from "@/types/attack-paths";
 
 import { Providers } from "../providers";
 
@@ -41,8 +49,30 @@ export default async function RootLayout({
 }: {
   children: ReactNode;
 }) {
-  const providersData = await getProviders({ page: 1, pageSize: 1 });
-  const hasProviders = !!(providersData?.data && providersData.data.length > 0);
+  // Onboarding is Cloud-only; skip its fetches and orchestrators in OSS.
+  const onboardingEnabled = isCloud();
+
+  // Fail-open: unknown scan state is treated as "has data" so the banner never blocks
+  // progression on a fetch error.
+  let hasCompletedScan = true;
+  // Tri-state: true = has providers, false = zero providers, undefined = fetch failed (gate fails open).
+  let hasProviders: boolean | undefined = false;
+
+  if (onboardingEnabled) {
+    const [providersData, scansByState] = await Promise.all([
+      getProviders({ page: 1, pageSize: 1 }),
+      getScansByState(),
+    ]);
+    hasCompletedScan = Array.isArray(scansByState?.data)
+      ? scansByState.data.some(
+          (scan: { attributes?: { state?: string } }) =>
+            scan.attributes?.state === SCAN_STATES.COMPLETED,
+        )
+      : true;
+    hasProviders = Array.isArray(providersData?.data)
+      ? providersData.data.length > 0
+      : undefined;
+  }
 
   return (
     <html suppressHydrationWarning lang="en">
@@ -55,8 +85,22 @@ export default async function RootLayout({
         )}
       >
         <Providers themeProps={{ attribute: "class", defaultTheme: "dark" }}>
-          <NavigationProgress />
-          <StoreInitializer values={{ hasProviders }} />
+          {/* Suspense contains the useSearchParams() CSR bailout so statically
+              prerendered pages don't fail the build (matches the auth layout). */}
+          <Suspense>
+            <NavigationProgress />
+          </Suspense>
+          {/* Store uses boolean; gate receives tri-state to fail open on fetch errors. */}
+          <StoreInitializer values={{ hasProviders: hasProviders ?? false }} />
+          {onboardingEnabled && (
+            <>
+              <OnboardingGate hasProviders={hasProviders} />
+              {/* Single mount point so the watcher survives post-connect navigation. */}
+              <OnboardingCheckpointWatcher />
+              {/* Persistent banner shown only while a guided sequence is active. */}
+              <OnboardingSequenceBanner hasCompletedScan={hasCompletedScan} />
+            </>
+          )}
           <MainLayout>{children}</MainLayout>
           <Toaster />
         </Providers>

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

@@ -22,12 +22,22 @@ export default async function Providers({
   const activeTab = getProviderTab(resolvedSearchParams.tab);
   const isCloudEnvironment = process.env.NEXT_PUBLIC_IS_CLOUD_ENV === "true";
 
-  // Exclude `tab` from the Suspense key so switching tabs doesn't re-suspend
-  const { tab: _, ...paramsWithoutTab } = resolvedSearchParams || {};
-  const searchParamsKey = JSON.stringify(paramsWithoutTab);
+  // Exclude `tab` and `onboarding` from the key: tab switches must not re-suspend,
+  // and `onboarding` is ephemeral (stripped via history.replaceState) — keeping it
+  // would remount ProvidersAccountsView and reset the wizard mid-flow.
+  const {
+    tab: _tab,
+    onboarding: _onboarding,
+    ...stableParams
+  } = resolvedSearchParams || {};
+  const searchParamsKey = JSON.stringify(stableParams);
 
   return (
-    <ContentLayout title="Providers" icon="lucide:cloud-cog">
+    <ContentLayout
+      title="Providers"
+      icon="lucide:cloud-cog"
+      onboardingAction={{ flowId: "add-provider" }}
+    >
       {isCloudEnvironment && <CliImportBanner className="mb-6" />}
       <FilterTransitionWrapper>
         <ProviderPageTabs
@@ -58,15 +68,10 @@ const ProvidersTableFallback = () => {
   return (
     <div className="flex flex-col gap-6">
       <div className="flex flex-wrap items-center gap-4">
-        {/* ProviderTypeSelector */}
         <Skeleton className="h-[52px] min-w-[200px] flex-1 rounded-lg md:max-w-[280px]" />
-        {/* Organizations filter */}
         <Skeleton className="h-[52px] max-w-[240px] min-w-[180px] flex-1 rounded-lg" />
-        {/* Provider Groups filter */}
         <Skeleton className="h-[52px] max-w-[240px] min-w-[180px] flex-1 rounded-lg" />
-        {/* Status filter */}
         <Skeleton className="h-[52px] max-w-[240px] min-w-[180px] flex-1 rounded-lg" />
-        {/* Action buttons */}
         <div className="ml-auto flex flex-wrap gap-4">
           <Skeleton className="h-9 w-[160px] rounded-md" />
           <Skeleton className="h-9 w-[120px] rounded-md" />