chore(merge): master

Co-authored-by: Jasmine Sullivan <20147180@tafe.wa.edu.au>
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

.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/sdk-tests.yml

@@ -540,7 +540,7 @@ jobs:
         with:
           flags: prowler-py${{ matrix.python-version }}-vercel
           files: ./vercel_coverage.xml
-
+      
       # Scaleway Provider
       - name: Check if Scaleway files changed
         if: steps.check-changes.outputs.any_changed == 'true'
@@ -588,7 +588,34 @@ jobs:
         with:
           flags: prowler-py${{ matrix.python-version }}-stackit
           files: ./stackit_coverage.xml
+ 
+      # External Provider (dynamic loading)
+      - name: Check if External Provider files changed
+        if: steps.check-changes.outputs.any_changed == 'true'
+        id: changed-external
+        uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
+        with:
+          files: |
+            ./prowler/providers/common/**
+            ./prowler/config/**
+            ./prowler/lib/**
+            ./tests/providers/external/**
+            ./uv.lock
 
+      - name: Run External Provider tests
+        if: steps.changed-external.outputs.any_changed == 'true'
+        run: uv run pytest -n auto --cov=./prowler/providers/common --cov=./prowler/config --cov=./prowler/lib --cov-report=xml:external_coverage.xml tests/providers/external
+
+      - name: Upload External Provider coverage to Codecov
+        if: steps.changed-external.outputs.any_changed == 'true'
+     
+        uses: codecov/codecov-action@671740ac38dd9b0130fbe1cec585b89eea48d3de # v5.5.2
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+        with:
+          flags: prowler-py${{ matrix.python-version }}-external
+          files: ./external_coverage.xml
+          
       # Lib
       - name: Check if Lib files changed
         if: steps.check-changes.outputs.any_changed == 'true'

.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 }}'

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

@@ -6,19 +6,30 @@ All notable changes to the **Prowler API** are documented in this file.
 
 ### 🚀 Added
 
-- 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)
+- Opt-in automatic recovery of allowlisted idempotent background tasks whose worker died during a deploy or crash: when enabled via `DJANGO_TASK_RECOVERY_ENABLED` (off by default), stuck summary and deletion tasks are detected and re-run instead of staying pending forever (scan and Jira tasks are excluded), with a `reconcile_orphan_tasks` management command for on-demand recovery [(#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
 
 - Allowlisted idempotent background tasks are no longer lost when a worker is stopped or crashes mid-task; tasks with external side effects are marked terminal instead of blindly re-running [(#11416)](https://github.com/prowler-cloud/prowler/pull/11416)
-- A recovered scan rewrites its findings, summaries, attack surface, and compliance data instead of appending to the previous run, so recovery never leaves stale or duplicate materialized rows [(#11416)](https://github.com/prowler-cloud/prowler/pull/11416)
 
 ### 🐞 Fixed
 
 - Workers now shut down gracefully on deploy or restart, finishing or re-queueing in-flight tasks instead of being force-killed and leaving them stuck [(#11416)](https://github.com/prowler-cloud/prowler/pull/11416)
 
+### 🔐 Security
+
+- `dulwich` from 0.23.0 to 1.2.5 and `pyjwt` from 2.12.1 to 2.13.0, patching `GHSA-897w-fcg9-f6xj` (arbitrary file write) and `PYSEC-2026-179` (HMAC/JWK key confusion) flagged by osv-scanner in `api/uv.lock` [(#11499)](https://github.com/prowler-cloud/prowler/pull/11499)
+
+---
+
+## [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)

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/docs/orphan-task-recovery.md

@@ -1,10 +1,11 @@
 # Orphan Celery task recovery
 
 When a worker is terminated mid-task (a deploy, an OOM kill, a node eviction), the
-task it was running can be left non-terminal forever: the `Scan` stays `EXECUTING`,
-the `TaskResult` stays `STARTED`, and nothing re-runs it. This page describes the
-mechanisms that detect and recover allowlisted idempotent orphans so users never
-see a stuck scan and pending-task alerts do not fire.
+task it was running can be left non-terminal forever: the `TaskResult` stays
+`STARTED` and nothing re-runs it. This page describes the mechanisms that detect and
+recover allowlisted idempotent orphans so pending-task alerts do not fire. Scan tasks
+are not auto-recovered (re-running a scan is not safe to do automatically); the
+watchdog covers the summary/aggregation and deletion tasks.
 
 ## How recovery works
 
@@ -13,29 +14,35 @@ see a stuck scan and pending-task alerts do not fire.
    (`worker_prefetch_multiplier = 1`), and an abruptly-lost worker re-queues its task
    (`task_reject_on_worker_lost`). On `SIGTERM` the worker is given a soft-shutdown
    window (`worker_soft_shutdown_timeout`) to finish or re-queue in-flight work
-   before it is force-killed.
+   before it is force-killed. `scan-perform`, `scan-perform-scheduled` and
+   `integration-jira` opt out of redelivery with `acks_late=False`, so a crash drops
+   them rather than re-running and duplicating findings or Jira issues. Other
+   non-recovered side-effect tasks keep `acks_late=True`, so the broker can still
+   re-deliver them after a worker loss: the S3 upload rebuilds from worker-local files
+   that did not survive the crash and so no-ops, but Security Hub re-reads findings from
+   the DB and re-sends them to AWS.
 
 2. **Periodic watchdog.** A Beat task, `reconcile-orphan-tasks`, runs every couple of
    minutes (a `django_celery_beat` periodic task created by migration). For each
    in-flight task result with an allowlisted idempotent task name, it pings the
    worker recorded on the task's `TaskResult`:
    - worker responds -> the task is still running, leave it alone;
-   - worker is gone (and the scan started before a short grace window) -> it is a
+   - worker is gone (and the task started before a short grace window) -> it is a
      real orphan: the stale task is revoked and marked terminal (clearing the
-     pending/started alert), and the scan is re-enqueued from scratch.
+     pending/started alert), and the task is re-enqueued from its stored name and
+     kwargs.
 
-   The re-run is safe because only tasks with proven idempotency are allowlisted.
-   Scan persistence, for example, clears the scan's prior findings and materialized
-   summary/compliance rows before re-writing them. Jira sends are allowlisted too:
-   each finding is reserved in a dispatch table before the external call, so a re-run
-   skips already-ticketed findings (the worst case is one finding missed if a worker
-   is hard-killed mid-send, never a duplicate issue). Other external side effects stay
-   terminal: the S3 upload rebuilds from worker-local files that do not survive a
-   crash, and report/Security Hub recovery is out of scope.
+   The re-run is safe because only tasks with proven idempotency are allowlisted: the
+   summary/aggregation tasks clear and re-write their own rows, and deletions are
+   idempotent. Scan tasks and external side effects are excluded: re-running a scan is
+   not safe to do automatically, Jira sends would create duplicate issues, the S3
+   upload rebuilds from worker-local files that do not survive a crash, and
+   report/Security Hub recovery is out of scope.
 
-3. **Recovery cap.** Each automatic re-enqueue increments `Scan.recovery_count`.
-   After `--max-attempts` recoveries (default 3) the scan is marked `FAILED` instead
-   of re-enqueued, so a task that repeatedly kills its worker cannot loop forever.
+3. **Recovery cap.** A per-task Valkey counter limits how often the same task is
+   re-enqueued. After `--max-attempts` recoveries (default 3) the orphan is marked
+   terminal instead of re-enqueued, so a task that repeatedly kills its worker cannot
+   loop forever.
 
 A Postgres advisory lock ensures that, even with multiple API/worker replicas, only
 one reconciliation runs at a time; the others no-op.
@@ -63,6 +70,18 @@ All settings have safe defaults; override via environment variables.
 | `DJANGO_CELERY_TASK_SOFT_TIME_LIMIT` | hard - 600 | Soft limit; raises `SoftTimeLimitExceeded` for cleanup. |
 | `DJANGO_CELERY_LONG_TASK_TIME_LIMIT` | `172800` (48h) | Hard limit for scans and provider/tenant deletions, which can legitimately run for more than a day. |
 | `DJANGO_CELERY_LONG_TASK_SOFT_TIME_LIMIT` | long hard - 600 | Soft limit for the long-running tasks above. |
+| `DJANGO_TASK_RECOVERY_ENABLED` | `false` | Master switch for orphan-task recovery, disabled by default (opt-in); set to `true` to enable. When off, no orphan is detected, marked terminal, or re-enqueued (attack-paths stale cleanup still runs). |
+| `DJANGO_TASK_RECOVERY_SUMMARIES_ENABLED` | `true` | Auto re-enqueue orphaned scan summary/aggregation tasks. |
+| `DJANGO_TASK_RECOVERY_DELETIONS_ENABLED` | `true` | Auto re-enqueue orphaned provider/tenant deletion tasks. |
+
+Recovery is opt-in: with the master flag off (the default) the sweep does nothing.
+Once enabled, the per-group flags default to on, so every group recovers unless you
+turn one off; a task whose group flag is off is marked terminal instead of
+re-enqueued.
+
+Turning recovery off only disables this watchdog sweep; it does not change Celery's
+broker-level redelivery (`task_acks_late`/`task_reject_on_worker_lost`), which still
+re-delivers tasks that keep `acks_late=True` on worker loss, independently of this flag.
 
 `task_acks_late` and `task_reject_on_worker_lost` are enabled in `config/celery.py`.
 

api/pyproject.toml

@@ -226,7 +226,7 @@ constraint-dependencies = [
   "drf-simple-apikey==2.2.1",
   "drf-spectacular==0.27.2",
   "drf-spectacular-jsonapi==0.5.1",
-  "dulwich==0.23.0",
+  "dulwich==1.2.5",
   "duo-client==5.5.0",
   "durationpy==0.10",
   "email-validator==2.2.0",
@@ -354,7 +354,7 @@ constraint-dependencies = [
   "pydantic-core==2.41.5",
   "pygithub==2.8.0",
   "pygments==2.20.0",
-  "pyjwt==2.12.1",
+  "pyjwt==2.13.0",
   "pylint==3.2.5",
   "pymsalruntime==0.18.1",
   "pynacl==1.6.2",
@@ -443,7 +443,17 @@ constraint-dependencies = [
 # The microsoft-kiota-http security bump to 1.9.9 (GHSA-7j59-v9qr-6fq9) requires
 # microsoft-kiota-abstractions>=1.9.9, which a constraint cannot satisfy against the
 # SDK's hard pin; override it to the patched, kiota-aligned version.
+#
+# prowler@master hard-pins dulwich==0.23.0 and pyjwt==2.12.1 in [project.dependencies].
+# dulwich 1.2.5 patches GHSA-897w-fcg9-f6xj (arbitrary file write) and pyjwt 2.13.0
+# patches PYSEC-2026-179 (HMAC/JWK key-confusion); a constraint cannot satisfy these
+# against the SDK's hard pins, so override them to the patched versions until the SDK
+# bump propagates to the pinned master rev. pyjwt keeps the [crypto] extra because an
+# override replaces the whole requirement; bare pyjwt would drop it from the consumers
+# that request pyjwt[crypto] and leave cryptography (needed for RS256) only transitive.
 override-dependencies = [
   "okta==3.4.2",
-  "microsoft-kiota-abstractions==1.9.9"
+  "microsoft-kiota-abstractions==1.9.9",
+  "dulwich==1.2.5",
+  "pyjwt[crypto]==2.13.0"
 ]

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/management/commands/reconcile_orphan_tasks.py

@@ -20,7 +20,7 @@ class Command(BaseCommand):
             "--max-attempts",
             type=int,
             default=3,
-            help="Give up re-running a task after this many recovery attempts (scans are marked FAILED).",
+            help="Give up re-running a task after this many recovery attempts; it is then left terminal instead of re-enqueued.",
         )
         parser.add_argument(
             "--dry-run",
@@ -39,6 +39,16 @@ class Command(BaseCommand):
             self.stdout.write("Reconcile skipped: another run holds the lock.")
             return
 
+        if result.get("enabled") is False:
+            message = (
+                "Task recovery is disabled (DJANGO_TASK_RECOVERY_ENABLED is off); "
+                "no orphans were recovered."
+            )
+            if result.get("attack_paths") is not None:
+                message += " Attack-paths stale cleanup still ran."
+            self.stdout.write(message)
+            return
+
         self.stdout.write(
             self.style.SUCCESS(
                 "Orphan reconcile complete: "

api/src/backend/api/migrations/0094_scan_recovery_count.py

@@ -1,17 +0,0 @@
-# Generated by Django 5.1.15 on 2026-05-30 17:38
-
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("api", "0093_okta_provider"),
-    ]
-
-    operations = [
-        migrations.AddField(
-            model_name="scan",
-            name="recovery_count",
-            field=models.IntegerField(default=0),
-        ),
-    ]

api/src/backend/api/migrations/0095_reconcile_orphan_tasks_periodic_task.py

@@ -40,7 +40,7 @@ def delete_periodic_task(apps, schema_editor):
 
 class Migration(migrations.Migration):
     dependencies = [
-        ("api", "0094_scan_recovery_count"),
+        ("api", "0093_okta_provider"),
         ("django_celery_beat", "0019_alter_periodictasks_options"),
     ]
 

api/src/backend/api/migrations/0096_jiraissuedispatch.py

@@ -1,64 +0,0 @@
-import uuid
-
-import django.db.models.deletion
-from django.db import migrations, models
-
-import api.rls
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("api", "0095_reconcile_orphan_tasks_periodic_task"),
-    ]
-
-    operations = [
-        migrations.CreateModel(
-            name="JiraIssueDispatch",
-            fields=[
-                (
-                    "id",
-                    models.UUIDField(
-                        default=uuid.uuid4,
-                        editable=False,
-                        primary_key=True,
-                        serialize=False,
-                    ),
-                ),
-                ("inserted_at", models.DateTimeField(auto_now_add=True)),
-                ("finding_id", models.UUIDField()),
-                (
-                    "integration",
-                    models.ForeignKey(
-                        on_delete=django.db.models.deletion.CASCADE,
-                        related_name="jira_dispatches",
-                        to="api.integration",
-                    ),
-                ),
-                (
-                    "tenant",
-                    models.ForeignKey(
-                        on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
-                    ),
-                ),
-            ],
-            options={
-                "db_table": "jira_issue_dispatches",
-                "abstract": False,
-            },
-        ),
-        migrations.AddConstraint(
-            model_name="jiraissuedispatch",
-            constraint=models.UniqueConstraint(
-                fields=("tenant_id", "integration_id", "finding_id"),
-                name="unique_jira_issue_dispatch",
-            ),
-        ),
-        migrations.AddConstraint(
-            model_name="jiraissuedispatch",
-            constraint=api.rls.RowLevelSecurityConstraint(
-                "tenant_id",
-                name="rls_on_jiraissuedispatch",
-                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
-            ),
-        ),
-    ]

api/src/backend/api/models.py

@@ -666,9 +666,6 @@ class Scan(RowLevelSecurityProtectedModel):
     state = StateEnumField(choices=StateChoices.choices, default=StateChoices.AVAILABLE)
     unique_resource_count = models.IntegerField(default=0)
     progress = models.IntegerField(default=0)
-    # Incremented by the scan-specific orphan-recovery path each time this scan is
-    # re-pointed to a fresh task; for observability (the retry cap is a Valkey counter).
-    recovery_count = models.IntegerField(default=0)
     scanner_args = models.JSONField(default=dict)
     duration = models.IntegerField(null=True, blank=True)
     scheduled_at = models.DateTimeField(null=True, blank=True)
@@ -2001,35 +1998,6 @@ class IntegrationProviderRelationship(RowLevelSecurityProtectedModel):
         ]
 
 
-class JiraIssueDispatch(RowLevelSecurityProtectedModel):
-    """Tracks findings already sent to a Jira integration.
-
-    Lets the Jira task be re-run safely (e.g. by orphan recovery): findings with
-    an existing dispatch row are skipped, so no duplicate issues are created.
-    """
-
-    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
-    inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
-    integration = models.ForeignKey(
-        Integration, on_delete=models.CASCADE, related_name="jira_dispatches"
-    )
-    finding_id = models.UUIDField()
-
-    class Meta(RowLevelSecurityProtectedModel.Meta):
-        db_table = "jira_issue_dispatches"
-        constraints = [
-            models.UniqueConstraint(
-                fields=["tenant_id", "integration_id", "finding_id"],
-                name="unique_jira_issue_dispatch",
-            ),
-            RowLevelSecurityConstraint(
-                field="tenant_id",
-                name="rls_on_%(class)s",
-                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
-            ),
-        ]
-
-
 class SAMLToken(models.Model):
     id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
     inserted_at = models.DateTimeField(auto_now_add=True, editable=False)

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,32 @@ SESSION_COOKIE_SECURE = True
 ATTACK_PATHS_SCAN_STALE_THRESHOLD_MINUTES = env.int(
     "ATTACK_PATHS_SCAN_STALE_THRESHOLD_MINUTES", 2880
 )  # 48h
+
+# Orphan task recovery feature flags. The master switch is OFF by default, so task
+# recovery is opt-in; enable it with DJANGO_TASK_RECOVERY_ENABLED=true. The per-group
+# toggles default to enabled, so once the master is on every group recovers unless a
+# group is explicitly turned off.
+TASK_RECOVERY_ENABLED = env.bool("DJANGO_TASK_RECOVERY_ENABLED", False)
+TASK_RECOVERY_SUMMARIES_ENABLED = env.bool(
+    "DJANGO_TASK_RECOVERY_SUMMARIES_ENABLED", True
+)
+TASK_RECOVERY_DELETIONS_ENABLED = env.bool(
+    "DJANGO_TASK_RECOVERY_DELETIONS_ENABLED", True
+)
+
+
+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

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

@@ -34,3 +34,8 @@ DRF_API_KEY = {
 # JWT
 
 SIMPLE_JWT["ALGORITHM"] = "HS256"  # noqa: F405
+# pyjwt >= 2.13.0 rejects an empty HMAC signing key, so HS256 tests need a real
+# key (>= 32 bytes also avoids the InsecureKeyLengthWarning). Production uses RS256.
+SIMPLE_JWT["SIGNING_KEY"] = env.str(  # noqa: F405
+    "DJANGO_TOKEN_SIGNING_KEY", "insecure-testing-jwt-signing-key-do-not-use-in-prod"
+)

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

@@ -11,7 +11,6 @@ from api.db_utils import batch_delete, rls_transaction
 from api.models import (
     AttackPathsScan,
     Finding,
-    JiraIssueDispatch,
     Provider,
     ProviderComplianceScore,
     Resource,
@@ -81,14 +80,6 @@ def delete_provider(tenant_id: str, pk: str):
 
         deletion_steps = [
             ("Scan Summaries", ScanSummary.all_objects.filter(scan__provider=instance)),
-            (
-                "Jira Issue Dispatches",
-                JiraIssueDispatch.objects.filter(
-                    finding_id__in=Finding.all_objects.filter(
-                        scan__provider=instance
-                    ).values_list("id", flat=True)
-                ),
-            ),
             ("Findings", Finding.all_objects.filter(scan__provider=instance)),
             ("Resources", Resource.all_objects.filter(provider=instance)),
             ("Scans", Scan.all_objects.filter(provider=instance)),

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

@@ -9,7 +9,7 @@ from tasks.utils import batched
 
 from api.db_router import READ_REPLICA_ALIAS, MainRouter
 from api.db_utils import REPLICA_MAX_ATTEMPTS, REPLICA_RETRY_BASE_DELAY, rls_transaction
-from api.models import Finding, Integration, JiraIssueDispatch, Provider
+from api.models import Finding, Integration, Provider
 from api.utils import initialize_prowler_integration, initialize_prowler_provider
 from prowler.lib.outputs.asff.asff import ASFF
 from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
@@ -482,115 +482,66 @@ def send_findings_to_jira(
     with rls_transaction(tenant_id):
         integration = Integration.objects.get(id=integration_id)
         jira_integration = initialize_prowler_integration(integration)
-        # Idempotency: findings already ticketed for this integration must not be
-        # sent again on a re-run (e.g. orphan recovery), to avoid duplicate issues
-        already_sent = {
-            str(fid)
-            for fid in JiraIssueDispatch.objects.filter(
-                integration_id=integration_id, finding_id__in=finding_ids
-            ).values_list("finding_id", flat=True)
-        }
 
     num_tickets_created = 0
-    skipped_count = 0
     for finding_id in finding_ids:
-        if str(finding_id) in already_sent:
-            skipped_count += 1
-            continue
-
-        # Reserve the finding BEFORE the external call. The unique constraint on
-        # (tenant, integration, finding) makes the dispatch row the single source of
-        # truth, so a concurrent run or a retry that raced past the bulk pre-check
-        # cannot create a duplicate issue: created=False means another run already
-        # claimed it. The reservation is released below if the send does not succeed.
         with rls_transaction(tenant_id):
-            _, created = JiraIssueDispatch.objects.get_or_create(
-                tenant_id=tenant_id,
-                integration_id=integration_id,
-                finding_id=finding_id,
+            finding_instance = (
+                Finding.all_objects.select_related("scan__provider")
+                .prefetch_related("resources")
+                .get(id=finding_id)
             )
-        if not created:
-            skipped_count += 1
-            continue
 
-        sent = False
-        try:
-            with rls_transaction(tenant_id):
-                finding_instance = (
-                    Finding.all_objects.select_related("scan__provider")
-                    .prefetch_related("resources")
-                    .get(id=finding_id)
-                )
+            # Extract resource information
+            resource = (
+                finding_instance.resources.first()
+                if finding_instance.resources.exists()
+                else None
+            )
+            resource_uid = resource.uid if resource else ""
+            resource_name = resource.name if resource else ""
+            resource_tags = {}
+            if resource and hasattr(resource, "tags"):
+                resource_tags = resource.get_tags(tenant_id)
 
-                # Extract resource information
-                resource = (
-                    finding_instance.resources.first()
-                    if finding_instance.resources.exists()
-                    else None
-                )
-                resource_uid = resource.uid if resource else ""
-                resource_name = resource.name if resource else ""
-                resource_tags = {}
-                if resource and hasattr(resource, "tags"):
-                    resource_tags = resource.get_tags(tenant_id)
+            # Get region
+            region = resource.region if resource and resource.region else ""
 
-                # Get region
-                region = resource.region if resource and resource.region else ""
+            # Extract remediation information from check_metadata
+            check_metadata = finding_instance.check_metadata
+            remediation = check_metadata.get("remediation", {})
+            recommendation = remediation.get("recommendation", {})
+            remediation_code = remediation.get("code", {})
 
-                # Extract remediation information from check_metadata
-                check_metadata = finding_instance.check_metadata
-                remediation = check_metadata.get("remediation", {})
-                recommendation = remediation.get("recommendation", {})
-                remediation_code = remediation.get("code", {})
-
-                # Send the individual finding to Jira
-                sent = bool(
-                    jira_integration.send_finding(
-                        check_id=finding_instance.check_id,
-                        check_title=check_metadata.get("checktitle", ""),
-                        severity=finding_instance.severity,
-                        status=finding_instance.status,
-                        status_extended=finding_instance.status_extended or "",
-                        provider=finding_instance.scan.provider.provider,
-                        region=region,
-                        resource_uid=resource_uid,
-                        resource_name=resource_name,
-                        risk=check_metadata.get("risk", ""),
-                        recommendation_text=recommendation.get("text", ""),
-                        recommendation_url=recommendation.get("url", ""),
-                        remediation_code_native_iac=remediation_code.get(
-                            "nativeiac", ""
-                        ),
-                        remediation_code_terraform=remediation_code.get(
-                            "terraform", ""
-                        ),
-                        remediation_code_cli=remediation_code.get("cli", ""),
-                        remediation_code_other=remediation_code.get("other", ""),
-                        resource_tags=resource_tags,
-                        compliance=finding_instance.compliance or {},
-                        project_key=project_key,
-                        issue_type=issue_type,
-                    )
-                )
-        finally:
-            if not sent:
-                # Release the reservation so a later run can retry this finding: it
-                # was not ticketed (send failed or raised), so the row must not block
-                # a future legitimate send.
-                with rls_transaction(tenant_id):
-                    JiraIssueDispatch.objects.filter(
-                        tenant_id=tenant_id,
-                        integration_id=integration_id,
-                        finding_id=finding_id,
-                    ).delete()
-
-        if sent:
-            num_tickets_created += 1
-        else:
-            logger.error(f"Failed to send finding {finding_id} to Jira")
+            # Send the individual finding to Jira
+            result = jira_integration.send_finding(
+                check_id=finding_instance.check_id,
+                check_title=check_metadata.get("checktitle", ""),
+                severity=finding_instance.severity,
+                status=finding_instance.status,
+                status_extended=finding_instance.status_extended or "",
+                provider=finding_instance.scan.provider.provider,
+                region=region,
+                resource_uid=resource_uid,
+                resource_name=resource_name,
+                risk=check_metadata.get("risk", ""),
+                recommendation_text=recommendation.get("text", ""),
+                recommendation_url=recommendation.get("url", ""),
+                remediation_code_native_iac=remediation_code.get("nativeiac", ""),
+                remediation_code_terraform=remediation_code.get("terraform", ""),
+                remediation_code_cli=remediation_code.get("cli", ""),
+                remediation_code_other=remediation_code.get("other", ""),
+                resource_tags=resource_tags,
+                compliance=finding_instance.compliance or {},
+                project_key=project_key,
+                issue_type=issue_type,
+            )
+            if result:
+                num_tickets_created += 1
+            else:
+                logger.error(f"Failed to send finding {finding_id} to Jira")
 
     return {
         "created_count": num_tickets_created,
-        "failed_count": len(finding_ids) - num_tickets_created - skipped_count,
-        "skipped_count": skipped_count,
+        "failed_count": len(finding_ids) - num_tickets_created,
     }

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

@@ -37,35 +37,52 @@ ORPHAN_RECOVERY_LOCK_KEY = 0x70726F77  # "prow"
 # Non-terminal states that mean "a worker had this and may have died with it".
 IN_FLIGHT_STATES = (states.STARTED, states.RECEIVED)
 
-# Scan tasks are recovered by re-running scan-perform on the EXISTING scan row,
-# not by re-enqueuing the original task: re-enqueuing scan-perform-scheduled would
-# hit its "a scan is already executing" guard and no-op, leaving the scan stuck.
-_SCAN_TASKS = ("scan-perform", "scan-perform-scheduled")
-
-# Tasks with proven idempotency are auto re-enqueued. Scans/summaries clear and
-# rewrite their own rows. integration-jira is safe too: each finding is reserved in
-# JiraIssueDispatch before the external call, so a re-run skips already-ticketed
-# findings (worst case one finding missed on a mid-send crash, never a duplicate).
-# Other external side effects stay terminal: integration-s3 rebuilds its upload from
-# worker-local files that do not survive a crash, and report/Security Hub recovery is
-# out of scope.
-REENQUEUEABLE_TASKS = {
-    *_SCAN_TASKS,
-    "provider-deletion",
-    "tenant-deletion",
-    "scan-summary",
-    "scan-compliance-overviews",
-    "scan-provider-compliance-scores",
-    "scan-daily-severity",
-    "scan-finding-group-summaries",
-    "scan-reset-ephemeral-resources",
-    "integration-jira",
+# Tasks with proven idempotency are eligible for auto re-enqueue, grouped so each
+# group can be toggled independently by a feature flag (see config.django.base).
+# Summaries clear and rewrite their own rows and deletions are idempotent. Tasks with
+# external side effects are never eligible: integration-jira would create duplicate
+# issues, integration-s3 rebuilds its upload from worker-local files that do not
+# survive a crash, and report/Security Hub recovery is out of scope.
+RECOVERY_TASK_GROUPS = {
+    "summaries": {
+        "scan-summary",
+        "scan-compliance-overviews",
+        "scan-provider-compliance-scores",
+        "scan-daily-severity",
+        "scan-finding-group-summaries",
+        "scan-reset-ephemeral-resources",
+    },
+    "deletions": {"provider-deletion", "tenant-deletion"},
 }
 
-# Tasks excluded from generic recovery: attack-paths scans are handled by their own
-# stale-cleanup (which also drops the temp Neo4j db), and the maintenance tasks must
-# not self-recover (they run again on their own schedule).
+
+def reenqueueable_tasks() -> set[str]:
+    """Task names eligible for auto re-enqueue, honoring the per-group feature flags.
+
+    A group whose flag is disabled is dropped, so its orphaned tasks are marked
+    terminal instead of re-enqueued.
+    """
+    from django.conf import settings
+
+    group_enabled = {
+        "summaries": settings.TASK_RECOVERY_SUMMARIES_ENABLED,
+        "deletions": settings.TASK_RECOVERY_DELETIONS_ENABLED,
+    }
+    return {
+        task
+        for group, tasks in RECOVERY_TASK_GROUPS.items()
+        if group_enabled[group]
+        for task in tasks
+    }
+
+
+# Tasks the watchdog ignores entirely (not even marked terminal): scan tasks are not
+# auto-recovered, since re-running a scan is not safe to do automatically; attack-paths
+# scans are handled by their own stale-cleanup (which also drops the temp Neo4j db);
+# and the maintenance tasks must not self-recover (they run again on their own schedule).
 _SKIP_RECOVERY = {
+    "scan-perform",
+    "scan-perform-scheduled",
     "attack-paths-scan-perform",
     "attack-paths-cleanup-stale-scans",
     "reconcile-orphan-tasks",
@@ -166,15 +183,22 @@ def reconcile_orphans(
             logger.info("Orphan reconcile skipped: another run holds the lock")
             return {"acquired": False}
 
-        # Populate the task registry so we can re-enqueue any task by name.
-        import tasks.tasks  # noqa: F401
+        from django.conf import settings
 
-        result = _reconcile_task_results(
-            grace_minutes=grace_minutes,
-            max_attempts=max_attempts,
-            window_hours=window_hours,
-            dry_run=dry_run,
-        )
+        if settings.TASK_RECOVERY_ENABLED:
+            # Populate the task registry so we can re-enqueue any task by name.
+            import tasks.tasks  # noqa: F401
+
+            result = _reconcile_task_results(
+                grace_minutes=grace_minutes,
+                max_attempts=max_attempts,
+                window_hours=window_hours,
+                dry_run=dry_run,
+            )
+            result["enabled"] = True
+        else:
+            logger.info("Orphan task recovery disabled by feature flag")
+            result = {"recovered": [], "failed": [], "skipped": [], "enabled": False}
 
         if not dry_run:
             from tasks.jobs.attack_paths.cleanup import cleanup_stale_attack_paths_scans
@@ -264,34 +288,27 @@ def _recover_task(task_result, max_attempts: int, window_hours: int) -> str:
     task_result.date_done = now
     task_result.save(update_fields=["status", "date_done"])
 
-    attempt = _recovery_attempt_count(name, kwargs_repr, window_hours)
-    if name not in REENQUEUEABLE_TASKS or attempt > max_attempts:
-        reason = (
-            f"{name} is not allowlisted for auto recovery"
-            if name not in REENQUEUEABLE_TASKS
-            else f"recovery cap reached ({attempt}/{max_attempts})"
-        )
-        _fail_domain_row(task_result.task_id, name, now)
+    if name not in reenqueueable_tasks():
         logger.warning(
-            "Orphan %s (%s) not re-enqueued: %s", task_result.task_id, name, reason
+            "Orphan %s (%s) not re-enqueued: not allowlisted for auto recovery",
+            task_result.task_id,
+            name,
         )
         return "failed"
 
-    # Scan tasks: re-run the EXISTING scan row directly via scan-perform, so the
-    # scheduled-scan "already executing" guard cannot turn recovery into a no-op.
-    # Falls through to the generic path only if no scan is linked yet (e.g. a
-    # scheduled task that died before creating one), where re-running it creates one.
-    if name in _SCAN_TASKS:
-        scan = _scan_for_task(task_result.task_id)
-        if scan is not None:
-            if not _reenqueue_scan(task_result.task_id, scan):
-                return "failed"
-            logger.info(
-                "Re-enqueued orphaned scan %s (was task %s)",
-                scan.id,
-                task_result.task_id,
-            )
-            return "recovered"
+    # Count the attempt only once the task is allowlisted, so a task sitting in a
+    # disabled group does not burn its recovery budget while the flag is off (and is
+    # not already over the cap the moment the group is re-enabled).
+    attempt = _recovery_attempt_count(name, kwargs_repr, window_hours)
+    if attempt > max_attempts:
+        logger.warning(
+            "Orphan %s (%s) not re-enqueued: recovery cap reached (%d/%d)",
+            task_result.task_id,
+            name,
+            attempt,
+            max_attempts,
+        )
+        return "failed"
 
     task_obj = current_app.tasks.get(name)
     if task_obj is None:
@@ -311,7 +328,6 @@ def _recover_task(task_result, max_attempts: int, window_hours: int) -> str:
             task_result.task_id,
             name,
         )
-        _fail_domain_row(task_result.task_id, name, now)
         return "failed"
     new_task_id = str(uuid4())
     task_obj.apply_async(
@@ -323,75 +339,3 @@ def _recover_task(task_result, max_attempts: int, window_hours: int) -> str:
         "Re-enqueued orphan %s (%s) as %s", task_result.task_id, name, new_task_id
     )
     return "recovered"
-
-
-def _scan_for_task(task_id: str):
-    """Return the Scan linked to a Celery task id, or None (read across tenants)."""
-    from api.db_router import MainRouter
-    from api.models import Scan
-
-    return Scan.all_objects.using(MainRouter.admin_db).filter(task_id=task_id).first()
-
-
-def _reenqueue_scan(old_task_id: str, scan) -> bool:
-    """Re-run an orphaned scan via scan-perform on the existing row.
-
-    Pre-provisions the new task linkage (TaskResult + api.Task) and relinks the
-    Scan before enqueuing, so the FK is valid and a worker can never outrun the DB.
-    The relink is conditional on the scan still pointing at the old task, so a stale
-    orphan can never clobber a newer linkage.
-    """
-    from django_celery_results.models import TaskResult
-
-    from api.db_utils import rls_transaction
-    from api.models import Scan
-    from api.models import Task as APITask
-    from tasks.tasks import perform_scan_task
-
-    tenant_id = str(scan.tenant_id)
-    new_task_id = str(uuid4())
-    with rls_transaction(tenant_id):
-        locked_scan = Scan.all_objects.select_for_update().filter(id=scan.id).first()
-        if locked_scan is None or str(locked_scan.task_id) != old_task_id:
-            logger.info(
-                "Scan %s no longer points at task %s; skipping recovery re-enqueue",
-                scan.id,
-                old_task_id,
-            )
-            return False
-        task_result_new, _ = TaskResult.objects.get_or_create(
-            task_id=new_task_id,
-            defaults={"status": states.PENDING, "task_name": "scan-perform"},
-        )
-        APITask.objects.update_or_create(
-            id=new_task_id,
-            tenant_id=tenant_id,
-            defaults={"task_runner_task": task_result_new},
-        )
-        locked_scan.task_id = new_task_id
-        locked_scan.recovery_count = (locked_scan.recovery_count or 0) + 1
-        locked_scan.save(update_fields=["task_id", "recovery_count", "updated_at"])
-
-    perform_scan_task.apply_async(
-        kwargs={
-            "tenant_id": tenant_id,
-            "scan_id": str(scan.id),
-            "provider_id": str(scan.provider_id),
-        },
-        task_id=new_task_id,
-    )
-    return True
-
-
-def _fail_domain_row(old_task_id: str, name: str, now: datetime) -> None:
-    """Mark a scan terminal when its task is capped/denylisted instead of re-run."""
-    from api.db_utils import rls_transaction
-    from api.models import Scan, StateChoices
-
-    if name in _SCAN_TASKS:
-        scan = _scan_for_task(old_task_id)
-        if scan is not None:
-            with rls_transaction(str(scan.tenant_id)):
-                Scan.all_objects.filter(id=scan.id, task_id=old_task_id).update(
-                    state=StateChoices.FAILED, completed_at=now
-                )

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

@@ -118,19 +118,6 @@ ATTACK_SURFACE_PROVIDER_COMPATIBILITY = {
 _ATTACK_SURFACE_MAPPING_CACHE: dict[str, dict] = {}
 
 
-def _clear_scan_rerun_state(tenant_id: str, scan_id: str) -> None:
-    """Remove rows derived from a previous execution of this scan."""
-    with rls_transaction(tenant_id):
-        Finding.all_objects.filter(scan_id=scan_id).delete()
-        ResourceScanSummary.objects.filter(scan_id=scan_id).delete()
-        ScanCategorySummary.objects.filter(scan_id=scan_id).delete()
-        ScanGroupSummary.objects.filter(scan_id=scan_id).delete()
-        ScanSummary.objects.filter(scan_id=scan_id).delete()
-        AttackSurfaceOverview.objects.filter(scan_id=scan_id).delete()
-        ComplianceRequirementOverview.objects.filter(scan_id=scan_id).delete()
-        ComplianceOverviewSummary.objects.filter(scan_id=scan_id).delete()
-
-
 def aggregate_category_counts(
     categories: list[str],
     severity: str,
@@ -489,10 +476,9 @@ def _create_compliance_summaries(
             )
         )
 
-    # Idempotent re-run: clear this scan's prior summaries before re-inserting, so
-    # a recovered scan's summary always reflects its own (re-derived) requirement
-    # rows rather than keeping a stale row (bulk_create ignore_conflicts alone would
-    # keep the old one).
+    # Idempotent re-run: clear this scan's prior summaries before re-inserting, so a
+    # recovered scan-compliance-overviews run reflects its own re-derived rows instead
+    # of keeping a stale one (bulk_create ignore_conflicts alone would keep the old).
     with rls_transaction(tenant_id):
         ComplianceOverviewSummary.objects.filter(scan_id=scan_id).delete()
         if summary_objects:
@@ -1039,7 +1025,6 @@ def perform_prowler_scan(
         scan_instance.state = StateChoices.EXECUTING
         scan_instance.started_at = datetime.now(tz=timezone.utc)
         scan_instance.save(update_fields=["state", "started_at", "updated_at"])
-    _clear_scan_rerun_state(tenant_id, scan_id)
 
     # Find the mutelist processor if it exists
     with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):

api/src/backend/tasks/tasks.py

@@ -260,7 +260,9 @@ def delete_provider_task(provider_id: str, tenant_id: str):
     return delete_provider(tenant_id=tenant_id, pk=provider_id)
 
 
-@shared_task(base=RLSTask, name="scan-perform", queue="scans")
+# acks_late=False: a re-run would duplicate findings and the task is not auto-recovered,
+# so a crashed scan is dropped rather than redelivered by the broker (as before #11416).
+@shared_task(base=RLSTask, name="scan-perform", queue="scans", acks_late=False)
 @handle_provider_deletion
 def perform_scan_task(
     tenant_id: str, scan_id: str, provider_id: str, checks_to_execute: list[str] = None
@@ -304,7 +306,14 @@ def perform_scan_task(
     return result
 
 
-@shared_task(base=RLSTask, bind=True, name="scan-perform-scheduled", queue="scans")
+# acks_late=False: like scan-perform; a dropped run is re-fired by Beat on the next tick.
+@shared_task(
+    base=RLSTask,
+    bind=True,
+    name="scan-perform-scheduled",
+    queue="scans",
+    acks_late=False,
+)
 @handle_provider_deletion
 def perform_scheduled_scan_task(self, tenant_id: str, provider_id: str):
     """
@@ -1151,10 +1160,13 @@ def security_hub_integration_task(
     return upload_security_hub_integration(tenant_id, provider_id, scan_id)
 
 
+# acks_late=False: Jira sends are not deduplicated and the task is not auto-recovered,
+# so a crashed send is dropped rather than redelivered (avoids duplicate Jira issues).
 @shared_task(
     base=RLSTask,
     name="integration-jira",
     queue="integrations",
+    acks_late=False,
 )
 def jira_integration_task(
     tenant_id: str,

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

@@ -1,12 +1,11 @@
 from unittest.mock import call, patch
-from uuid import uuid4
 
 import pytest
 from django.core.exceptions import ObjectDoesNotExist
 from tasks.jobs.deletion import delete_provider, delete_tenant
 
 from api.attack_paths import database as graph_database
-from api.models import JiraIssueDispatch, Provider, Tenant, TenantComplianceSummary
+from api.models import Provider, Tenant, TenantComplianceSummary
 
 
 @pytest.mark.django_db
@@ -35,43 +34,6 @@ class TestDeleteProvider:
                 str(instance.id),
             )
 
-    def test_delete_provider_removes_jira_dispatches(
-        self,
-        providers_fixture,
-        findings_fixture,
-        integrations_fixture,
-    ):
-        """Deleting a provider removes JiraIssueDispatch rows for its findings only."""
-        instance = providers_fixture[0]
-        tenant_id = str(instance.tenant_id)
-        finding = findings_fixture[0]
-        integration = integrations_fixture[0]
-
-        # Dispatch for one of the provider's findings: must be removed with it.
-        JiraIssueDispatch.objects.create(
-            tenant_id=tenant_id,
-            integration=integration,
-            finding_id=finding.id,
-        )
-        # Dispatch for an unrelated finding: must survive the provider deletion.
-        unrelated = JiraIssueDispatch.objects.create(
-            tenant_id=tenant_id,
-            integration=integration,
-            finding_id=uuid4(),
-        )
-
-        with (
-            patch(
-                "tasks.jobs.deletion.graph_database.get_database_name",
-                return_value="tenant-db",
-            ),
-            patch("tasks.jobs.deletion.graph_database.drop_subgraph"),
-        ):
-            delete_provider(tenant_id, instance.id)
-
-        assert not JiraIssueDispatch.objects.filter(finding_id=finding.id).exists()
-        assert JiraIssueDispatch.objects.filter(pk=unrelated.pk).exists()
-
     def test_delete_provider_does_not_exist(self, tenants_fixture):
         with (
             patch(

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

@@ -1640,74 +1640,14 @@ class TestJiraIntegration:
     @patch("tasks.jobs.integrations.Finding")
     @patch("tasks.jobs.integrations.Integration")
     @patch("tasks.jobs.integrations.initialize_prowler_integration")
-    @patch("tasks.jobs.integrations.JiraIssueDispatch")
-    def test_send_findings_to_jira_skips_already_dispatched(
-        self,
-        mock_jira_dispatch,
-        mock_initialize_integration,
-        mock_integration_model,
-        mock_finding_model,
-        mock_rls_transaction,
-    ):
-        """A re-run skips findings already ticketed (no duplicate Jira issues)."""
-        mock_rls_transaction.return_value.__enter__ = MagicMock()
-        mock_rls_transaction.return_value.__exit__ = MagicMock()
-        mock_integration_model.objects.get.return_value = MagicMock()
-        # finding-1 was already dispatched in a prior run; finding-2 is new.
-        mock_jira_dispatch.objects.filter.return_value.values_list.return_value = [
-            "finding-1"
-        ]
-        mock_jira_dispatch.objects.get_or_create.return_value = (MagicMock(), True)
-
-        mock_jira_integration = MagicMock()
-        mock_jira_integration.send_finding.return_value = True
-        mock_initialize_integration.return_value = mock_jira_integration
-
-        finding2 = MagicMock()
-        finding2.id = "finding-2"
-        finding2.check_id = "check_002"
-        finding2.severity = "low"
-        finding2.status = "FAIL"
-        finding2.status_extended = ""
-        finding2.compliance = {}
-        finding2.resources.exists.return_value = False
-        finding2.resources.first.return_value = None
-        finding2.scan.provider.provider = "aws"
-        finding2.check_metadata = {
-            "checktitle": "C2",
-            "risk": "",
-            "remediation": {"recommendation": {}, "code": {}},
-        }
-        mock_finding_model.all_objects.select_related.return_value.prefetch_related.return_value.get.return_value = finding2
-
-        result = send_findings_to_jira(
-            "tenant-123", "integration-456", "PROJ", "Task", ["finding-1", "finding-2"]
-        )
-
-        # finding-1 skipped (already sent); only finding-2 sent -> no duplicate.
-        assert result == {"created_count": 1, "failed_count": 0, "skipped_count": 1}
-        mock_jira_integration.send_finding.assert_called_once()
-        assert (
-            mock_jira_integration.send_finding.call_args.kwargs["check_id"]
-            == "check_002"
-        )
-
-    @patch("tasks.jobs.integrations.rls_transaction")
-    @patch("tasks.jobs.integrations.Finding")
-    @patch("tasks.jobs.integrations.Integration")
-    @patch("tasks.jobs.integrations.initialize_prowler_integration")
-    @patch("tasks.jobs.integrations.JiraIssueDispatch")
     def test_send_findings_to_jira_success(
         self,
-        mock_jira_dispatch,
         mock_initialize_integration,
         mock_integration_model,
         mock_finding_model,
         mock_rls_transaction,
     ):
         """Test successful sending of findings to Jira using send_finding method"""
-        mock_jira_dispatch.objects.filter.return_value.values_list.return_value = []
-        mock_jira_dispatch.objects.get_or_create.return_value = (MagicMock(), True)
         tenant_id = "tenant-123"
         integration_id = "integration-456"
         project_key = "PROJ"
@@ -1799,7 +1739,7 @@ class TestJiraIntegration:
         )
 
         # Assertions
-        assert result == {"created_count": 2, "failed_count": 0, "skipped_count": 0}
+        assert result == {"created_count": 2, "failed_count": 0}
 
         # Verify Jira integration was initialized
         mock_initialize_integration.assert_called_once_with(integration)
@@ -1831,10 +1771,8 @@ class TestJiraIntegration:
     @patch("tasks.jobs.integrations.Integration")
     @patch("tasks.jobs.integrations.initialize_prowler_integration")
     @patch("tasks.jobs.integrations.logger")
-    @patch("tasks.jobs.integrations.JiraIssueDispatch")
     def test_send_findings_to_jira_partial_failure(
         self,
-        mock_jira_dispatch,
         mock_logger,
         mock_initialize_integration,
         mock_integration_model,
@@ -1842,8 +1780,6 @@ class TestJiraIntegration:
         mock_rls_transaction,
     ):
         """Test partial failure when sending findings to Jira"""
-        mock_jira_dispatch.objects.filter.return_value.values_list.return_value = []
-        mock_jira_dispatch.objects.get_or_create.return_value = (MagicMock(), True)
         tenant_id = "tenant-123"
         integration_id = "integration-456"
         project_key = "PROJ"
@@ -1897,35 +1833,23 @@ class TestJiraIntegration:
         )
 
         # Assertions
-        assert result == {"created_count": 2, "failed_count": 1, "skipped_count": 0}
+        assert result == {"created_count": 2, "failed_count": 1}
 
         # Verify error was logged for the failed finding
         mock_logger.error.assert_called_with("Failed to send finding finding-2 to Jira")
 
-        # The failed finding's reservation is released so a later run can retry it.
-        mock_jira_dispatch.objects.filter.assert_any_call(
-            tenant_id=tenant_id,
-            integration_id=integration_id,
-            finding_id="finding-2",
-        )
-        mock_jira_dispatch.objects.filter.return_value.delete.assert_called_once()
-
     @patch("tasks.jobs.integrations.rls_transaction")
     @patch("tasks.jobs.integrations.Finding")
     @patch("tasks.jobs.integrations.Integration")
     @patch("tasks.jobs.integrations.initialize_prowler_integration")
-    @patch("tasks.jobs.integrations.JiraIssueDispatch")
     def test_send_findings_to_jira_no_resources(
         self,
-        mock_jira_dispatch,
         mock_initialize_integration,
         mock_integration_model,
         mock_finding_model,
         mock_rls_transaction,
     ):
         """Test sending findings to Jira when finding has no resources"""
-        mock_jira_dispatch.objects.filter.return_value.values_list.return_value = []
-        mock_jira_dispatch.objects.get_or_create.return_value = (MagicMock(), True)
         tenant_id = "tenant-123"
         integration_id = "integration-456"
         project_key = "PROJ"
@@ -1983,7 +1907,7 @@ class TestJiraIntegration:
         )
 
         # Assertions
-        assert result == {"created_count": 1, "failed_count": 0, "skipped_count": 0}
+        assert result == {"created_count": 1, "failed_count": 0}
 
         # Verify send_finding was called with empty resource fields
         call_kwargs = mock_jira_integration.send_finding.call_args.kwargs
@@ -1996,18 +1920,14 @@ class TestJiraIntegration:
     @patch("tasks.jobs.integrations.Finding")
     @patch("tasks.jobs.integrations.Integration")
     @patch("tasks.jobs.integrations.initialize_prowler_integration")
-    @patch("tasks.jobs.integrations.JiraIssueDispatch")
     def test_send_findings_to_jira_with_empty_check_metadata(
         self,
-        mock_jira_dispatch,
         mock_initialize_integration,
         mock_integration_model,
         mock_finding_model,
         mock_rls_transaction,
     ):
         """Test sending findings to Jira when check_metadata is empty or missing fields"""
-        mock_jira_dispatch.objects.filter.return_value.values_list.return_value = []
-        mock_jira_dispatch.objects.get_or_create.return_value = (MagicMock(), True)
         tenant_id = "tenant-123"
         integration_id = "integration-456"
         project_key = "PROJ"
@@ -2050,7 +1970,7 @@ class TestJiraIntegration:
         )
 
         # Assertions
-        assert result == {"created_count": 1, "failed_count": 0, "skipped_count": 0}
+        assert result == {"created_count": 1, "failed_count": 0}
 
         # Verify send_finding was called with default/empty values
         call_kwargs = mock_jira_integration.send_finding.call_args.kwargs
@@ -2063,94 +1983,3 @@ class TestJiraIntegration:
         assert call_kwargs["remediation_code_cli"] == ""
         assert call_kwargs["remediation_code_other"] == ""
         assert call_kwargs["compliance"] == {}
-
-    @patch("tasks.jobs.integrations.rls_transaction")
-    @patch("tasks.jobs.integrations.Finding")
-    @patch("tasks.jobs.integrations.Integration")
-    @patch("tasks.jobs.integrations.initialize_prowler_integration")
-    @patch("tasks.jobs.integrations.JiraIssueDispatch")
-    def test_send_findings_to_jira_reserves_before_sending(
-        self,
-        mock_jira_dispatch,
-        mock_initialize_integration,
-        mock_integration_model,
-        mock_finding_model,
-        mock_rls_transaction,
-    ):
-        """The dispatch row is reserved before the external Jira call (reserve-then-act)."""
-        mock_rls_transaction.return_value.__enter__ = MagicMock()
-        mock_rls_transaction.return_value.__exit__ = MagicMock()
-        mock_integration_model.objects.get.return_value = MagicMock()
-        mock_jira_dispatch.objects.filter.return_value.values_list.return_value = []
-
-        order = []
-        mock_jira_dispatch.objects.get_or_create.side_effect = lambda **kw: (
-            order.append(("reserve", kw)) or (MagicMock(), True)
-        )
-
-        mock_jira_integration = MagicMock()
-        mock_jira_integration.send_finding.side_effect = lambda **kw: (
-            order.append(("send", kw)) or True
-        )
-        mock_initialize_integration.return_value = mock_jira_integration
-
-        finding = MagicMock()
-        finding.id = "finding-1"
-        finding.check_id = "check_001"
-        finding.severity = "low"
-        finding.status = "FAIL"
-        finding.status_extended = ""
-        finding.compliance = {}
-        finding.resources.exists.return_value = False
-        finding.resources.first.return_value = None
-        finding.scan.provider.provider = "aws"
-        finding.check_metadata = {
-            "checktitle": "C1",
-            "risk": "",
-            "remediation": {"recommendation": {}, "code": {}},
-        }
-        mock_finding_model.all_objects.select_related.return_value.prefetch_related.return_value.get.return_value = finding
-
-        result = send_findings_to_jira(
-            "tenant-123", "integration-456", "PROJ", "Task", ["finding-1"]
-        )
-
-        assert result == {"created_count": 1, "failed_count": 0, "skipped_count": 0}
-        # Reservation must precede the external send.
-        assert [entry[0] for entry in order] == ["reserve", "send"]
-        # A successful send keeps the reservation (no rollback delete).
-        mock_jira_dispatch.objects.filter.return_value.delete.assert_not_called()
-
-    @patch("tasks.jobs.integrations.rls_transaction")
-    @patch("tasks.jobs.integrations.Finding")
-    @patch("tasks.jobs.integrations.Integration")
-    @patch("tasks.jobs.integrations.initialize_prowler_integration")
-    @patch("tasks.jobs.integrations.JiraIssueDispatch")
-    def test_send_findings_to_jira_skips_when_already_reserved(
-        self,
-        mock_jira_dispatch,
-        mock_initialize_integration,
-        mock_integration_model,
-        mock_finding_model,
-        mock_rls_transaction,
-    ):
-        """A finding that races past the bulk pre-check but loses the reservation
-        (created=False) is skipped without a second issue, leaving the row intact."""
-        mock_rls_transaction.return_value.__enter__ = MagicMock()
-        mock_rls_transaction.return_value.__exit__ = MagicMock()
-        mock_integration_model.objects.get.return_value = MagicMock()
-        mock_jira_dispatch.objects.filter.return_value.values_list.return_value = []
-        # Another concurrent run already created the dispatch row.
-        mock_jira_dispatch.objects.get_or_create.return_value = (MagicMock(), False)
-
-        mock_jira_integration = MagicMock()
-        mock_initialize_integration.return_value = mock_jira_integration
-
-        result = send_findings_to_jira(
-            "tenant-123", "integration-456", "PROJ", "Task", ["finding-1"]
-        )
-
-        assert result == {"created_count": 0, "failed_count": 0, "skipped_count": 1}
-        mock_jira_integration.send_finding.assert_not_called()
-        # The reservation belongs to the run that won the race; do not delete it.
-        mock_jira_dispatch.objects.filter.return_value.delete.assert_not_called()

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

@@ -4,17 +4,17 @@ from uuid import uuid4
 
 import pytest
 from celery import states
+from django.test import override_settings
 from django_celery_results.models import TaskResult
 
-from api.models import Scan, StateChoices
-from api.models import Task as APITask
 from tasks.jobs.orphan_recovery import (
     _decode_celery_field,
     _reconcile_task_results,
     _recovery_attempt_count,
-    _reenqueue_scan,
     advisory_lock,
     is_worker_alive,
+    reconcile_orphans,
+    reenqueueable_tasks,
 )
 
 
@@ -130,9 +130,83 @@ class TestReconcileTaskResults:
         assert tr.task_id in result["failed"]
         mock_task.apply_async.assert_not_called()
 
-    def test_jira_integration_task_is_reenqueued(self, tenants_fixture):
-        """integration-jira is re-enqueued: its JiraIssueDispatch reservation makes a
-        re-run skip already-ticketed findings, so recovery cannot duplicate issues."""
+    @override_settings(TASK_RECOVERY_SUMMARIES_ENABLED=False)
+    def test_disabled_group_task_is_not_reenqueued(self, tenants_fixture):
+        """A task whose group feature flag is off stays terminal, not re-enqueued."""
+        tr = _orphan_result(
+            name="scan-summary",
+            kwargs={
+                "tenant_id": str(tenants_fixture[0].id),
+                "scan_id": str(uuid4()),
+            },
+            worker="dead@gone",
+            created_minutes_ago=60,
+        )
+        p_alive, p_revoke, p_app, mock_task = self._patches(alive=False)
+        with (
+            p_alive,
+            p_revoke,
+            p_app,
+            patch("tasks.jobs.orphan_recovery._recovery_attempt_count", return_value=1),
+        ):
+            result = _reconcile_task_results(
+                grace_minutes=2, max_attempts=3, window_hours=6, dry_run=False
+            )
+
+        assert tr.task_id in result["failed"]
+        mock_task.apply_async.assert_not_called()
+
+    @override_settings(TASK_RECOVERY_SUMMARIES_ENABLED=False)
+    def test_disabled_group_task_does_not_consume_recovery_attempt(
+        self, tenants_fixture
+    ):
+        """A disabled-group task is failed without incrementing its Valkey attempt
+        counter, so re-enabling the group does not start it at the cap."""
+        tr = _orphan_result(
+            name="scan-summary",
+            kwargs={"tenant_id": str(tenants_fixture[0].id), "scan_id": str(uuid4())},
+            worker="dead@gone",
+            created_minutes_ago=60,
+        )
+        p_alive, p_revoke, p_app, mock_task = self._patches(alive=False)
+        with (
+            p_alive,
+            p_revoke,
+            p_app,
+            patch("tasks.jobs.orphan_recovery._recovery_attempt_count") as mock_count,
+        ):
+            result = _reconcile_task_results(
+                grace_minutes=2, max_attempts=3, window_hours=6, dry_run=False
+            )
+
+        assert tr.task_id in result["failed"]
+        mock_count.assert_not_called()
+
+    def test_scan_task_is_skipped_entirely(self, tenants_fixture):
+        """Scan tasks are excluded from recovery: the watchdog never touches them."""
+        tr = _orphan_result(
+            name="scan-perform",
+            kwargs={
+                "tenant_id": str(tenants_fixture[0].id),
+                "scan_id": str(uuid4()),
+            },
+            worker="dead@gone",
+            created_minutes_ago=60,
+        )
+        p_alive, p_revoke, p_app, mock_task = self._patches(alive=False)
+        with p_alive, p_revoke, p_app:
+            result = _reconcile_task_results(
+                grace_minutes=2, max_attempts=3, window_hours=6, dry_run=False
+            )
+
+        assert tr.task_id not in result["recovered"]
+        assert tr.task_id not in result["failed"]
+        assert tr.task_id not in result["skipped"]
+        mock_task.apply_async.assert_not_called()
+
+    def test_jira_integration_task_is_not_reenqueued(self, tenants_fixture):
+        """integration-jira stays terminal: re-running it would create duplicate Jira
+        issues, so an orphaned send is failed instead of re-enqueued."""
         tenant = tenants_fixture[0]
         kwargs = {
             "tenant_id": str(tenant.id),
@@ -158,13 +232,10 @@ class TestReconcileTaskResults:
                 grace_minutes=2, max_attempts=3, window_hours=6, dry_run=False
             )
 
-        assert tr.task_id in result["recovered"]
+        assert tr.task_id in result["failed"]
         tr.refresh_from_db()
         assert tr.status == states.REVOKED  # stale result cleared (no pending alert)
-        mock_task.apply_async.assert_called_once()
-        call = mock_task.apply_async.call_args.kwargs
-        assert call["kwargs"] == kwargs
-        assert call["task_id"] != tr.task_id  # fresh task id
+        mock_task.apply_async.assert_not_called()
 
     def test_skips_live_worker(self, tenants_fixture):
         tr = _orphan_result(
@@ -246,98 +317,6 @@ class TestReconcileTaskResults:
         mock_task.apply_async.assert_not_called()
 
 
-@pytest.mark.django_db
-class TestScanRecovery:
-    """Scans are recovered by re-running scan-perform on the EXISTING scan row,
-    so even a scheduled-scan orphan (whose own task would no-op on its guard) is
-    actually re-executed."""
-
-    def _scan_orphan(self, tenant, provider, name):
-        old_id = str(uuid4())
-        tr = TaskResult.objects.create(
-            task_id=old_id,
-            status=states.STARTED,
-            task_name=name,
-            worker="dead@gone",
-            task_kwargs=repr(
-                {"tenant_id": str(tenant.id), "provider_id": str(provider.id)}
-            ),
-            task_args=repr([]),
-        )
-        TaskResult.objects.filter(pk=tr.pk).update(
-            date_created=datetime.now(tz=timezone.utc) - timedelta(minutes=60)
-        )
-        APITask.objects.create(id=old_id, tenant_id=tenant.id, task_runner_task=tr)
-        scan = Scan.objects.create(
-            name="scan-orphan",
-            provider=provider,
-            trigger=Scan.TriggerChoices.SCHEDULED,
-            state=StateChoices.EXECUTING,
-            tenant_id=tenant.id,
-            task_id=old_id,
-            recovery_count=0,
-        )
-        return old_id, scan
-
-    @pytest.mark.parametrize("name", ["scan-perform", "scan-perform-scheduled"])
-    def test_scan_recovered_via_scan_perform(
-        self, tenants_fixture, providers_fixture, name
-    ):
-        tenant, provider = tenants_fixture[0], providers_fixture[0]
-        old_id, scan = self._scan_orphan(tenant, provider, name)
-
-        with (
-            patch("tasks.jobs.orphan_recovery.is_worker_alive", return_value=False),
-            patch("tasks.jobs.orphan_recovery.revoke_task"),
-            patch("tasks.jobs.orphan_recovery._recovery_attempt_count", return_value=1),
-            patch("tasks.tasks.perform_scan_task") as mock_scan_task,
-        ):
-            result = _reconcile_task_results(
-                grace_minutes=2, max_attempts=3, window_hours=6, dry_run=False
-            )
-
-        assert old_id in result["recovered"]
-        scan.refresh_from_db()
-        assert str(scan.task_id) != old_id  # relinked to a fresh task
-        assert scan.recovery_count == 1
-        assert TaskResult.objects.get(task_id=old_id).status == states.REVOKED
-        # Recovered by re-running scan-perform on the existing scan row (so the
-        # scheduled guard cannot no-op it), regardless of the original task name.
-        mock_scan_task.apply_async.assert_called_once()
-        assert mock_scan_task.apply_async.call_args.kwargs["kwargs"]["scan_id"] == str(
-            scan.id
-        )
-
-    def test_reenqueue_skips_when_scan_already_repointed(
-        self, tenants_fixture, providers_fixture
-    ):
-        # The scan already points at a newer task, so a stale orphan must not relink
-        # it or launch a second concurrent run against the same scan row.
-        tenant, provider = tenants_fixture[0], providers_fixture[0]
-        newer_id = str(uuid4())
-        tr = TaskResult.objects.create(
-            task_id=newer_id, status=states.STARTED, task_name="scan-perform"
-        )
-        APITask.objects.create(id=newer_id, tenant_id=tenant.id, task_runner_task=tr)
-        scan = Scan.objects.create(
-            name="scan-orphan",
-            provider=provider,
-            trigger=Scan.TriggerChoices.SCHEDULED,
-            state=StateChoices.EXECUTING,
-            tenant_id=tenant.id,
-            task_id=newer_id,
-            recovery_count=0,
-        )
-
-        with patch("tasks.tasks.perform_scan_task") as mock_scan_task:
-            recovered = _reenqueue_scan(str(uuid4()), scan)
-
-        assert recovered is False
-        mock_scan_task.apply_async.assert_not_called()
-        scan.refresh_from_db()
-        assert scan.recovery_count == 0
-
-
 @pytest.mark.django_db
 class TestOrphanRecoveryHelpers:
     def test_advisory_lock_acquires_and_releases(self):
@@ -370,3 +349,60 @@ class TestOrphanRecoveryHelpers:
         with patch("redis.from_url", return_value=redis_client):
             assert _recovery_attempt_count("probe-task", kwargs_repr, 6) == 1
             assert _recovery_attempt_count("probe-task", kwargs_repr, 6) == 2
+
+
+class TestRecoveryFeatureFlags:
+    def test_all_groups_enabled_by_default(self):
+        tasks = reenqueueable_tasks()
+        assert "scan-summary" in tasks
+        assert {"provider-deletion", "tenant-deletion"} <= tasks
+
+    @override_settings(TASK_RECOVERY_SUMMARIES_ENABLED=False)
+    def test_summaries_group_flag_excludes_summary_tasks(self):
+        tasks = reenqueueable_tasks()
+        assert "scan-summary" not in tasks
+        assert "scan-compliance-overviews" not in tasks
+        assert "provider-deletion" in tasks
+
+    @override_settings(TASK_RECOVERY_DELETIONS_ENABLED=False)
+    def test_deletions_group_flag_excludes_deletion_tasks(self):
+        tasks = reenqueueable_tasks()
+        assert "provider-deletion" not in tasks
+        assert "tenant-deletion" not in tasks
+        assert "scan-summary" in tasks
+
+
+@pytest.mark.django_db
+class TestRecoveryMasterFlag:
+    @override_settings(TASK_RECOVERY_ENABLED=False)
+    def test_master_flag_disables_task_recovery(self):
+        with (
+            patch(
+                "tasks.jobs.orphan_recovery._reconcile_task_results"
+            ) as mock_reconcile,
+            patch(
+                "tasks.jobs.attack_paths.cleanup.cleanup_stale_attack_paths_scans",
+                return_value={},
+            ),
+        ):
+            result = reconcile_orphans(grace_minutes=2, max_attempts=3, dry_run=False)
+
+        mock_reconcile.assert_not_called()
+        assert result["acquired"] is True
+        assert result["enabled"] is False
+
+    @override_settings(TASK_RECOVERY_ENABLED=True)
+    def test_master_flag_enabled_runs_task_recovery(self):
+        with (
+            patch(
+                "tasks.jobs.orphan_recovery._reconcile_task_results",
+                return_value={"recovered": [], "failed": [], "skipped": []},
+            ) as mock_reconcile,
+            patch(
+                "tasks.jobs.attack_paths.cleanup.cleanup_stale_attack_paths_scans",
+                return_value={},
+            ),
+        ):
+            reconcile_orphans(grace_minutes=2, max_attempts=3, dry_run=False)
+
+        mock_reconcile.assert_called_once()

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

@@ -32,15 +32,12 @@ from tasks.utils import CustomEncoder
 from api.db_router import MainRouter
 from api.exceptions import ProviderConnectionError
 from api.models import (
-    AttackSurfaceOverview,
     Finding,
     MuteRule,
     Provider,
     Resource,
     ResourceScanSummary,
     Scan,
-    ScanCategorySummary,
-    ScanGroupSummary,
     ScanSummary,
     StateChoices,
     StatusChoices,
@@ -232,131 +229,6 @@ class TestPerformScan:
         # Assert that failed_findings_count is 0 (finding is PASS and muted)
         assert scan_resource.failed_findings_count == 0
 
-    def test_perform_prowler_scan_idempotent_on_rerun(
-        self,
-        tenants_fixture,
-        scans_fixture,
-        providers_fixture,
-    ):
-        """Re-running a scan for the same scan_id must not duplicate findings."""
-        with (
-            patch("api.db_utils.rls_transaction"),
-            patch(
-                "tasks.jobs.scan.initialize_prowler_provider"
-            ) as mock_initialize_prowler_provider,
-            patch("tasks.jobs.scan.ProwlerScan") as mock_prowler_scan_class,
-            patch(
-                "tasks.jobs.scan.PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE",
-                new_callable=dict,
-            ),
-            patch("api.compliance.PROWLER_CHECKS", new_callable=dict) as mock_checks,
-        ):
-            mock_checks["aws"] = {"check1": {"compliance1"}}
-
-            tenant = tenants_fixture[0]
-            scan = scans_fixture[0]
-            provider = providers_fixture[0]
-            provider.provider = Provider.ProviderChoices.AWS
-            provider.save()
-
-            tenant_id = str(tenant.id)
-            scan_id = str(scan.id)
-            provider_id = str(provider.id)
-
-            stale_resource = Resource.objects.create(
-                tenant_id=tenant.id,
-                provider=provider,
-                uid="stale_resource_uid",
-                name="stale",
-                region="stale-region",
-                service="stale-service",
-                type="stale-type",
-            )
-            ResourceScanSummary.objects.create(
-                tenant_id=tenant.id,
-                scan_id=scan.id,
-                resource_id=stale_resource.id,
-                service="stale-service",
-                region="stale-region",
-                resource_type="stale-type",
-            )
-            ScanCategorySummary.objects.create(
-                tenant_id=tenant.id,
-                scan=scan,
-                category="stale-category",
-                severity=Severity.medium,
-                total_findings=1,
-            )
-            ScanGroupSummary.objects.create(
-                tenant_id=tenant.id,
-                scan=scan,
-                resource_group="stale-group",
-                severity=Severity.medium,
-                total_findings=1,
-            )
-            ScanSummary.objects.create(
-                tenant_id=tenant.id,
-                scan=scan,
-                check_id="stale_check",
-                service="stale-service",
-                severity=Severity.medium,
-                region="stale-region",
-                total=1,
-            )
-            AttackSurfaceOverview.objects.create(
-                tenant_id=tenant.id,
-                scan=scan,
-                attack_surface_type=AttackSurfaceOverview.AttackSurfaceTypeChoices.SECRETS,
-                total_findings=1,
-            )
-
-            finding = MagicMock()
-            finding.uid = "dup_probe_finding"
-            finding.status = StatusChoices.PASS
-            finding.status_extended = "x"
-            finding.severity = Severity.medium
-            finding.check_id = "check1"
-            finding.get_metadata.return_value = {"key": "value"}
-            finding.resource_uid = "resource_uid"
-            finding.resource_name = "resource_name"
-            finding.region = "region"
-            finding.service_name = "service_name"
-            finding.resource_type = "resource_type"
-            finding.resource_tags = {}
-            finding.muted = False
-            finding.raw = {}
-            finding.resource_metadata = {}
-            finding.resource_details = {}
-            finding.partition = "partition"
-            finding.compliance = {}
-
-            mock_scan_instance = MagicMock()
-            mock_scan_instance.scan.return_value = [(100, [finding])]
-            mock_prowler_scan_class.return_value = mock_scan_instance
-
-            mock_provider_instance = MagicMock()
-            mock_provider_instance.get_regions.return_value = ["region"]
-            mock_initialize_prowler_provider.return_value = mock_provider_instance
-
-            # Run the same scan twice (simulating an orphan-recovery re-run).
-            perform_prowler_scan(tenant_id, scan_id, provider_id, ["check1"])
-            perform_prowler_scan(tenant_id, scan_id, provider_id, ["check1"])
-
-        # Neither findings nor resources are duplicated by the re-run: findings are
-        # scope-deleted before re-insert; resources are upserted by (tenant, provider, uid).
-        assert Finding.objects.filter(scan=scan).count() == 1
-        assert Resource.objects.filter(provider=provider).count() == 2
-        assert ResourceScanSummary.objects.filter(scan_id=scan.id).count() == 1
-        assert not ResourceScanSummary.objects.filter(
-            scan_id=scan.id, resource_id=stale_resource.id
-        ).exists()
-        assert not ScanCategorySummary.objects.filter(scan=scan).exists()
-        assert not ScanGroupSummary.objects.filter(scan=scan).exists()
-        assert not ScanSummary.objects.filter(
-            scan=scan, check_id="stale_check"
-        ).exists()
-        assert not AttackSurfaceOverview.objects.filter(scan=scan).exists()
-
     @patch("tasks.jobs.scan.ProwlerScan")
     @patch(
         "tasks.jobs.scan.initialize_prowler_provider",

api/uv.lock

@@ -163,7 +163,7 @@ constraints = [
     { name = "drf-simple-apikey", specifier = "==2.2.1" },
     { name = "drf-spectacular", specifier = "==0.27.2" },
     { name = "drf-spectacular-jsonapi", specifier = "==0.5.1" },
-    { name = "dulwich", specifier = "==0.23.0" },
+    { name = "dulwich", specifier = "==1.2.5" },
     { name = "duo-client", specifier = "==5.5.0" },
     { name = "durationpy", specifier = "==0.10" },
     { name = "email-validator", specifier = "==2.2.0" },
@@ -291,7 +291,7 @@ constraints = [
     { name = "pydantic-core", specifier = "==2.41.5" },
     { name = "pygithub", specifier = "==2.8.0" },
     { name = "pygments", specifier = "==2.20.0" },
-    { name = "pyjwt", specifier = "==2.12.1" },
+    { name = "pyjwt", specifier = "==2.13.0" },
     { name = "pylint", specifier = "==3.2.5" },
     { name = "pymsalruntime", specifier = "==0.18.1" },
     { name = "pynacl", specifier = "==1.6.2" },
@@ -374,8 +374,10 @@ constraints = [
     { name = "zstd", specifier = "==1.5.7.3" },
 ]
 overrides = [
+    { name = "dulwich", specifier = "==1.2.5" },
     { name = "microsoft-kiota-abstractions", specifier = "==1.9.9" },
     { name = "okta", specifier = "==3.4.2" },
+    { name = "pyjwt", extras = ["crypto"], specifier = "==2.13.0" },
 ]
 
 [[package]]
@@ -393,7 +395,7 @@ version = "1.2.7"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "cryptography" },
-    { name = "pyjwt" },
+    { name = "pyjwt", extra = ["crypto"] },
     { name = "python-dateutil" },
     { name = "requests" },
 ]
@@ -1074,7 +1076,7 @@ dependencies = [
     { name = "pkginfo" },
     { name = "psutil", marker = "sys_platform != 'cygwin'" },
     { name = "py-deviceid" },
-    { name = "pyjwt" },
+    { name = "pyjwt", extra = ["crypto"] },
     { name = "pyopenssl" },
     { name = "requests", extra = ["socks"] },
 ]
@@ -2457,7 +2459,7 @@ source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "django" },
     { name = "djangorestframework" },
-    { name = "pyjwt" },
+    { name = "pyjwt", extra = ["crypto"] },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/a8/27/2874a325c11112066139769f7794afae238a07ce6adf96259f08fd37a9d7/djangorestframework_simplejwt-5.5.1.tar.gz", hash = "sha256:e72c5572f51d7803021288e2057afcbd03f17fe11d484096f40a460abc76e87f", size = 101265, upload-time = "2025-07-21T16:52:25.026Z" }
 wheels = [
@@ -2576,24 +2578,27 @@ wheels = [
 
 [[package]]
 name = "dulwich"
-version = "0.23.0"
+version = "1.2.5"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
+    { name = "typing-extensions", marker = "python_full_version < '3.12'" },
     { name = "urllib3" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/4b/ac/ba58cf420640c7bc77ae8e1b31e174d83c9117750c63cf9ea3b5e202e5c4/dulwich-0.23.0.tar.gz", hash = "sha256:0aa6c2489dd5e978b27e9b75983b7331a66c999f0efc54ebe37cab808ed322ae", size = 575116, upload-time = "2025-06-21T17:56:47.494Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/7f/85/ceb8ecff5cdeee4ceeebb86b599476dee559041dacc6c2c50cc0d4711549/dulwich-1.2.5.tar.gz", hash = "sha256:0395b2c8924c3424bafe2d9c1edd5348cc4b21ce9c1d6655bf01f9a5c47164c8", size = 1253230, upload-time = "2026-05-28T22:27:55.17Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/ae/11/f6bbba8583f69cf19ef4bd7f5fde1a6b5ccaf8b6951781cec8db247116f4/dulwich-0.23.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:d68498fdda13ab00791b483daab3bcfe9f9721c037aa458695e6ad81640c57cc", size = 972658, upload-time = "2025-06-21T17:56:13.505Z" },
-    { url = "https://files.pythonhosted.org/packages/2b/9d/2720e0ab58666378a33c752a61543f936cd6b06dfe5d84a2215ddc0914b0/dulwich-0.23.0-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:cb7bb930b12471a1cfcea4b3d25a671dc0ad32573f0ad25684684298959a1527", size = 1049813, upload-time = "2025-06-21T17:56:14.884Z" },
-    { url = "https://files.pythonhosted.org/packages/e5/f3/81d8075141dfcc0a0449c2093596e58d3e11444e3af54e819eca63b84dd0/dulwich-0.23.0-cp311-cp311-manylinux_2_28_x86_64.whl", hash = "sha256:a2abbce32fd2bc7902bcc5f69b10bf22576810de21651baaa864b78fd7aec261", size = 1051639, upload-time = "2025-06-21T17:56:16.437Z" },
-    { url = "https://files.pythonhosted.org/packages/4f/0d/c06ccb227b096aef5906142fe78b5c79f9070a0ea6152fc219941186d540/dulwich-0.23.0-cp311-cp311-win32.whl", hash = "sha256:9e3151f10ce2a9ff91bca64c74345217f53bdd947dc958032343822009832f7a", size = 642918, upload-time = "2025-06-21T17:56:18.373Z" },
-    { url = "https://files.pythonhosted.org/packages/d7/1c/1e99aa34c9aead9e641b2d9934f0a3d00257f75027cf5cdecc8a1a6c18ae/dulwich-0.23.0-cp311-cp311-win_amd64.whl", hash = "sha256:3ae9f1d9dc92d4e9a3f89ba2c55221f7b6442c5dd93b3f6f539a3c9eb3f37bdd", size = 659010, upload-time = "2025-06-21T17:56:19.947Z" },
-    { url = "https://files.pythonhosted.org/packages/4a/d7/1e6fba0235babe912e8467b036062e37d11672cbbeb0d8074f9d4559057b/dulwich-0.23.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:52cdef66a7994d29528ca79ca59452518bbba3fd56a9c61c61f6c467c1c7956e", size = 960292, upload-time = "2025-06-21T17:56:21.308Z" },
-    { url = "https://files.pythonhosted.org/packages/4b/6a/23f0c487ec03f2752600cab4a8e0dedb38186246c475bf3fa90a8db830d5/dulwich-0.23.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:d473888a6ab9ed5d4a4c3f053cbe5b77f72d54b6efdf5688fed76094316e571e", size = 1047892, upload-time = "2025-06-21T17:56:22.989Z" },
-    { url = "https://files.pythonhosted.org/packages/c7/e2/8f3d216be5fd0ee1180d917b59b34b54b9896384cf139f319b5d3a8f16b4/dulwich-0.23.0-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:19fcf20224c641a61c774da92f098fbaae9938c7e17a52841e64092adf7e78f9", size = 1048699, upload-time = "2025-06-21T17:56:24.602Z" },
-    { url = "https://files.pythonhosted.org/packages/8f/c4/18e6223cd4ad1ae9334eb4e6aa5952fd8f5c3d75762918eb90c209fec4ba/dulwich-0.23.0-cp312-cp312-win32.whl", hash = "sha256:7fc8b76b704ef35cd001e993e3aa4e1d666a2064bf467c07c560f12b2959dcaf", size = 641268, upload-time = "2025-06-21T17:56:26.18Z" },
-    { url = "https://files.pythonhosted.org/packages/b8/9c/65bfbbac62d8a2967e13f6a1512371c5eb6b906a61fb6dead992669cad0e/dulwich-0.23.0-cp312-cp312-win_amd64.whl", hash = "sha256:cb0566b888b578325350b4d67c61a0de35d417e9877560e3a6df88cae4576a59", size = 657837, upload-time = "2025-06-21T17:56:27.821Z" },
-    { url = "https://files.pythonhosted.org/packages/35/31/49318ee9db4b402e6d8b9b01bd4cae9298f59e1bb9bd56cf4a94e48fa069/dulwich-0.23.0-py3-none-any.whl", hash = "sha256:d8da6694ca332bb48775e35ee2215aa4673821164a91b83062f699c69f7cd135", size = 313776, upload-time = "2025-06-21T17:56:46.221Z" },
+    { url = "https://files.pythonhosted.org/packages/4a/4a/654ae1671610fdf6b65a64586ad67ddd8550d4d08a632b2a4b9614754b6d/dulwich-1.2.5-cp311-cp311-macosx_10_12_x86_64.whl", hash = "sha256:556593fd11637f80f6018bee1916b1a84f5b420423b470ebb3f1a782ad6ef081", size = 1399277, upload-time = "2026-05-28T22:27:00.801Z" },
+    { url = "https://files.pythonhosted.org/packages/85/d8/06ee3bc8eded4bd7adf8adf0c9ea5f19bf96f7e5e626bfaf7311cde4208a/dulwich-1.2.5-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:a70477c991e96cfe8fdd7c866e7251faf71b38bfeb51d6f27554c9cce1caabf3", size = 1382310, upload-time = "2026-05-28T22:27:02.216Z" },
+    { url = "https://files.pythonhosted.org/packages/07/17/a03adf50b9095f9f5d863393f21d585dea39bdc4fdf60788ff3a9407a512/dulwich-1.2.5-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:9008ef25cabd379cda4fa86000fc38ca14b72afe17db798a8c85c0b2b7ce4d1e", size = 1470993, upload-time = "2026-05-28T22:27:04.075Z" },
+    { url = "https://files.pythonhosted.org/packages/60/58/1dc352d2a5e80befe4338af7208febb44bcfd7496b0dde5ac6dacb07b031/dulwich-1.2.5-cp311-cp311-manylinux_2_28_x86_64.whl", hash = "sha256:a5549f4afc973e0a15ea6b0244d57f848d3f3ee13dac557eb311024aebebf128", size = 1497820, upload-time = "2026-05-28T22:27:05.549Z" },
+    { url = "https://files.pythonhosted.org/packages/c1/a8/e058959a87e7df7753b112ef66a43ccbc57338c1bbdc23a0edf3833396df/dulwich-1.2.5-cp311-cp311-win32.whl", hash = "sha256:5108acead814d1de8b6262d6d8fb90af7e82f5a4d83788b6b48e39d01800a92f", size = 1066549, upload-time = "2026-05-28T22:27:06.832Z" },
+    { url = "https://files.pythonhosted.org/packages/33/91/ff0b444f686718635348986bd73dfce42e947912417893de35de399b878b/dulwich-1.2.5-cp311-cp311-win_amd64.whl", hash = "sha256:5e067b7feceb7034bc99e7c7143a704f1d97d4be7027d9a0aa5a83c0657ff091", size = 1079481, upload-time = "2026-05-28T22:27:08.33Z" },
+    { url = "https://files.pythonhosted.org/packages/19/22/4f75770bbe5521cac61c4820ef46d4fbf8c2175d3519ba3d0378d4ba798e/dulwich-1.2.5-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:701a9ecf7a8a44f5e2459e46befa93530cf36a8b1ae3140aefc007db1d7d0207", size = 1396522, upload-time = "2026-05-28T22:27:09.997Z" },
+    { url = "https://files.pythonhosted.org/packages/e5/b1/c07c347681c0cf6acd4b189bf6e8d6207c71a1347b7a1e865eb40faa46b9/dulwich-1.2.5-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:2f90d68bfa97c4ca71de7507984365aefe27b6d248cb28dc99644d0f3ae8c60b", size = 1334826, upload-time = "2026-05-28T22:27:11.582Z" },
+    { url = "https://files.pythonhosted.org/packages/13/80/6818eb7ce492e18ab2efa92ab901d173b4b0b159e5681c1424f329600c40/dulwich-1.2.5-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:00b54a1d56ddbacdd8eadd6d4787a51b3a05fefa30eadbf9165fd283a00b90ed", size = 1416616, upload-time = "2026-05-28T22:27:13.195Z" },
+    { url = "https://files.pythonhosted.org/packages/14/a7/9790e60d19870f6554f7583722bb324c1355784316f20aeda1c0b5b1491a/dulwich-1.2.5-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:d8f7ea8f47e38e5b0de3fab97e07e9c9161ffddc90b3964512cab2b7749df4e6", size = 1441354, upload-time = "2026-05-28T22:27:14.683Z" },
+    { url = "https://files.pythonhosted.org/packages/91/44/0ea8a69c24aa1254ff5996d682eae2eab287d471b937dcdb26d9ea9720b4/dulwich-1.2.5-cp312-cp312-win32.whl", hash = "sha256:8929134acf4ff967203df7600b38535f9b5b590462067a7e30dbce01acb97af9", size = 1017058, upload-time = "2026-05-28T22:27:16.121Z" },
+    { url = "https://files.pythonhosted.org/packages/bd/9c/2fcddda7faec3bae52db7c64bfcb5dc756f597f33fae90e8d4e4b4d3b39b/dulwich-1.2.5-cp312-cp312-win_amd64.whl", hash = "sha256:9693d2c9e226b2ea855c1dc3a87e2f4d972f7523fc0f7924e5997e9f4c23d97f", size = 1031731, upload-time = "2026-05-28T22:27:17.633Z" },
+    { url = "https://files.pythonhosted.org/packages/07/4b/4a18a59ad230581cd0ef460e96001f90762e566dc2dfdba22aa358eb5a0e/dulwich-1.2.5-py3-none-any.whl", hash = "sha256:1679b376433a0fc7f36586afda1d4ed7427afa7a79d4bf17e5014474eea69fa4", size = 686745, upload-time = "2026-05-28T22:27:53.695Z" },
 ]
 
 [[package]]
@@ -4031,7 +4036,7 @@ dependencies = [
     { name = "pycryptodomex" },
     { name = "pydantic" },
     { name = "pydash" },
-    { name = "pyjwt" },
+    { name = "pyjwt", extra = ["crypto"] },
     { name = "python-dateutil" },
     { name = "pyyaml" },
     { name = "requests" },
@@ -4873,11 +4878,11 @@ wheels = [
 
 [[package]]
 name = "pyjwt"
-version = "2.12.1"
+version = "2.13.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/c2/27/a3b6e5bf6ff856d2509292e95c8f57f0df7017cf5394921fc4e4ef40308a/pyjwt-2.12.1.tar.gz", hash = "sha256:c74a7a2adf861c04d002db713dd85f84beb242228e671280bf709d765b03672b", size = 102564, upload-time = "2026-03-13T19:27:37.25Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/3b/81/58d0ac84e1ef3a3843791d6954d94c0b33d526c75eeb1efbce9d0a4c4077/pyjwt-2.13.0.tar.gz", hash = "sha256:41571c89ca91598c79e8ef18a2d07367d4810fbbd6f637794879baf1b7703423", size = 107515, upload-time = "2026-05-21T19:54:36.618Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/e5/7a/8dd906bd22e79e47397a61742927f6747fe93242ef86645ee9092e610244/pyjwt-2.12.1-py3-none-any.whl", hash = "sha256:28ca37c070cad8ba8cd9790cd940535d40274d22f80ab87f3ac6a713e6e8454c", size = 29726, upload-time = "2026-03-13T19:27:35.677Z" },
+    { url = "https://files.pythonhosted.org/packages/a3/5e/ecf12fdb62546d64385c158514e9b2b671f7832108ef2ecd2020ce0af2d1/pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728", size = 31274, upload-time = "2026-05-21T19:54:35.362Z" },
 ]
 
 [package.optional-dependencies]
@@ -5785,7 +5790,7 @@ source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "cryptography" },
     { name = "httpx" },
-    { name = "pyjwt" },
+    { name = "pyjwt", extra = ["crypto"] },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/3c/2f/99fb8718274116c5c146c745755620fd5c5943f78ca52ca9b17e94348286/workos-6.0.4.tar.gz", hash = "sha256:b0bfe8fd212b8567422c4ea3732eb33608794033eb3a69900c6b04db183c32d6", size = 172217, upload-time = "2026-04-16T03:09:28.583Z" }
 wheels = [

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/developer-guide/ai-skills.mdx

@@ -8,7 +8,77 @@ This guide explains the AI Skills system that provides on-demand context and pat
 **What are AI Skills?** Skills are structured instructions that help AI agents (Claude Code, Cursor, Copilot, etc.) understand Prowler's conventions, patterns, and best practices.
 </Info>
 
-## Architecture Overview
+Skills live in the [`skills/`](https://github.com/prowler-cloud/prowler/tree/master/skills) directory of the Prowler OSS repository. Each skill is a folder containing a `SKILL.md` file with its patterns and metadata.
+
+## Installation
+
+To enable skills for the supported AI coding assistants, run the setup script from the repository root:
+
+```bash
+./skills/setup.sh
+```
+
+The script creates symlinks so each tool finds the skills in its expected location:
+
+| Tool | Created by setup |
+|------|------------------|
+| Claude Code | `.claude/skills/` symlink and `CLAUDE.md` |
+| Gemini CLI | `.gemini/skills/` symlink and `GEMINI.md` |
+| Codex (OpenAI) | `.codex/skills/` symlink (uses `AGENTS.md` natively) |
+| GitHub Copilot | `.github/copilot-instructions.md` symlink to `AGENTS.md` |
+
+After running the setup, restart the AI coding assistant to load the skills.
+
+## Using Skills
+
+AI agents discover skills automatically and load them when a request matches a skill trigger. To load a skill manually during a session, point the agent to the skill's `SKILL.md` file:
+
+```text
+Read skills/{skill-name}/SKILL.md
+```
+
+For the full list of available skills, their triggers, and the Auto-invoke mappings, see the [`skills/README.md`](https://github.com/prowler-cloud/prowler/blob/master/skills/README.md) and [`AGENTS.md`](https://github.com/prowler-cloud/prowler/blob/master/AGENTS.md) in the repository.
+
+## Available Skills
+
+| Type | Skills |
+|------|--------|
+| **Generic** | typescript, react-19, nextjs-16, tailwind-4, pytest, playwright, django-drf, zod-4, zustand-5, ai-sdk-5, vitest, tdd |
+| **Prowler** | prowler, prowler-sdk-check, prowler-api, prowler-ui, prowler-mcp, prowler-provider, prowler-compliance, prowler-compliance-review, prowler-docs, prowler-pr, prowler-ci, prowler-attack-paths-query |
+| **Testing** | prowler-test-sdk, prowler-test-api, prowler-test-ui |
+| **Meta** | skill-creator, skill-sync |
+
+<Note>
+This table is a snapshot. The repository is the source of truth: see [`skills/README.md`](https://github.com/prowler-cloud/prowler/blob/master/skills/README.md) for the current, complete list.
+</Note>
+
+## Skill Structure
+
+Each skill follows the [Agent Skills spec](https://agentskills.io):
+
+```text
+skills/{skill-name}/
+├── SKILL.md          # Patterns, rules, decision trees
+├── assets/           # Code templates, schemas
+└── references/       # Links to local docs (single source of truth)
+```
+
+## Key Design Decisions
+
+1. **Self-contained skills** - Critical patterns inline for fast loading
+2. **Local doc references** - No web URLs, points to `docs/developer-guide/*.mdx`
+3. **Single source of truth** - Skills reference docs, no duplication
+4. **On-demand loading** - AI loads only what's needed for the task
+
+## Creating New Skills
+
+Use the `skill-creator` meta-skill to create new skills that follow the Agent Skills spec. See [`AGENTS.md`](https://github.com/prowler-cloud/prowler/blob/master/AGENTS.md) for the full list of available skills and their triggers.
+
+## How Skills Work
+
+The diagrams below explain the internals of the skill system. They are useful for understanding the design, but are not required to install or use skills.
+
+### Architecture Overview
 
 ```mermaid
 graph LR
@@ -28,7 +98,7 @@ graph LR
     style F fill:#1a4d2e,stroke:#66bb6a,color:#fff
 ```
 
-## How It Works
+### Request Lifecycle
 
 ```mermaid
 sequenceDiagram
@@ -68,7 +138,7 @@ sequenceDiagram
     A->>U: Creates check with correct patterns
 ```
 
-## Before vs After
+### With and Without Skills
 
 ```mermaid
 graph TD
@@ -96,7 +166,7 @@ graph TD
     style AFTER fill:#1a4d1a,stroke:#66bb6a,color:#fff
 ```
 
-## Complete Architecture
+### Full Component Map
 
 ```mermaid
 flowchart TB
@@ -110,7 +180,7 @@ flowchart TB
         subgraph GENERIC["Generic Skills"]
             G1["typescript"]
             G2["react-19"]
-            G3["nextjs-15"]
+            G3["nextjs-16"]
             G4["tailwind-4"]
             G5["pytest"]
             G6["playwright"]
@@ -186,34 +256,3 @@ flowchart TB
     style STRUCTURE fill:#5c3d1a,stroke:#ffb74d,color:#fff
     style DOCS fill:#1a3d4d,stroke:#4dd0e1,color:#fff
 ```
-
-## Skills Included
-
-| Type | Skills |
-|------|--------|
-| **Generic** | typescript, react-19, nextjs-15, tailwind-4, pytest, playwright, django-drf, zod-4, zustand-5, ai-sdk-5 |
-| **Prowler** | prowler, prowler-sdk-check, prowler-api, prowler-ui, prowler-mcp, prowler-provider, prowler-compliance, prowler-compliance-review, prowler-docs, prowler-pr, prowler-ci |
-| **Testing** | prowler-test-sdk, prowler-test-api, prowler-test-ui |
-| **Meta** | skill-creator, skill-sync |
-
-## Skill Structure
-
-Each skill follows the [Agent Skills spec](https://agentskills.io):
-
-```
-skills/{skill-name}/
-├── SKILL.md          # Patterns, rules, decision trees
-├── assets/           # Code templates, schemas
-└── references/       # Links to local docs (single source of truth)
-```
-
-## Key Design Decisions
-
-1. **Self-contained skills** - Critical patterns inline for fast loading
-2. **Local doc references** - No web URLs, points to `docs/developer-guide/*.mdx`
-3. **Single source of truth** - Skills reference docs, no duplication
-4. **On-demand loading** - AI loads only what's needed for the task
-
-## Creating New Skills
-
-Use the `skill-creator` meta-skill to create new skills that follow the Agent Skills spec. See `AGENTS.md` for the full list of available skills and their triggers.

docs/developer-guide/configurable-checks.mdx

@@ -40,9 +40,76 @@ When adding a new configurable check to Prowler, update the following files:
   # aws.awslambda_function_vpc_multi_az
   lambda_min_azs: 2
   ```
+- **Provider Schema:** Add the typed field to the provider's Pydantic schema in `prowler/config/schema/<provider>.py`. This is required: the loader validates user configs against these schemas and the shipped `config.yaml` must round-trip with zero warnings. See [Adding a parameter to the provider schema](#adding-a-parameter-to-the-provider-schema) below.
 - **Test Fixtures:** If tests depend on this configuration, add the variable to `tests/config/fixtures/config.yaml`.
 - **Documentation:** Document the new variable in the list of configurable checks in `docs/tutorials/configuration_file.md`.
 
 For a complete list of checks that already support configuration, see the [Configuration File Tutorial](/user-guide/cli/tutorials/configuration_file).
 
+## Adding a parameter to the provider schema
+
+Every provider has a typed Pydantic schema in `prowler/config/schema/`. When a config is loaded, `validate_provider_config` checks each user-supplied key against the schema, logs a warning, and drops any field that fails validation. The consumer's `.get(key, default)` then falls back to the built-in default.
+
+This catches typos in a value (for example, `0.2` typed as `20`, or `"medium"` for an enum that expects `"MEDIUM"`). It does NOT catch typos in a key name: `disalowed_regions` (one `l` missing) is treated as an unknown key and passes through untouched, because third-party check plugins legitimately rely on unknown keys being preserved. Reviewers should still check that any new key the YAML adds is named exactly the same as the field on the schema.
+
+### Where to add the field
+
+1. Open `prowler/config/schema/<provider>.py` (for example, `aws.py`).
+2. Add a field on the provider's schema class. Always make it `Optional[...] = None` so the absence of the key is valid.
+3. Apply the tightest type the value allows. Examples below.
+
+If you are introducing an entirely new provider rather than a new parameter, also add an entry mapping the provider name to its schema class in `prowler/config/schema/registry.py`. The loader uses that registry to find the schema for the provider it is loading.
+
+### Choosing the right type
+
+| Value kind | Field declaration |
+|---|---|
+| Boolean toggle | `Optional[bool] = None` |
+| Strictly positive integer (days, counts) | `Optional[int] = Field(default=None, gt=0)` |
+| Fraction in 0..1 (threshold) | `Optional[float] = Field(default=None, ge=0.0, le=1.0)` |
+| Closed set of strings | `Optional[Literal["A", "B", "C"]] = None` |
+| Free-form string | `Optional[str] = None` |
+| List of strings or ints | `Optional[list[str]] = None` |
+
+Prefer `Literal[...]` over `str` whenever the value is one of a known set. Prefer `Field(gt=0)` over `int` whenever zero or negative would be nonsensical. The point of the schema is to catch real-world mistakes that previously passed silently.
+
+### Custom validators (only when needed)
+
+If the value has structural rules beyond type and range, add a `field_validator`. Examples already in `aws.py`:
+
+- `_validate_port_range` rejects ports outside `0..65535`.
+- `_validate_account_ids` rejects anything that isn't a 12-digit AWS account ID.
+- `_validate_trusted_ips` rejects entries that aren't a valid IP or CIDR.
+
+Raise `ValueError` from the validator. The framework converts the error into a warning and drops the offending key.
+
+### Example: adding a new parameter
+
+Say a new check needs `max_iam_role_session_hours`, a strictly positive integer that defaults to 12 in code.
+
+1. **Schema** (`prowler/config/schema/aws.py`):
+    ```python
+    # IAM
+    max_iam_role_session_hours: Optional[int] = Field(default=None, gt=0)
+    ```
+2. **Shipped config** (`prowler/config/config.yaml`):
+    ```yaml
+    # aws.iam_role_session_duration_within_limit
+    max_iam_role_session_hours: 12
+    ```
+3. **Consumer** (the check):
+    ```python
+    max_hours = iam_client.audit_config.get("max_iam_role_session_hours", 12)
+    ```
+4. **Tests** in `tests/config/schema/aws_schema_test.py`:
+    - one test for a valid value that round-trips,
+    - one test for an invalid value (zero, negative, wrong type) that is dropped.
+
+### What the loader guarantees
+
+- **Unknown keys pass through.** Third-party check plugins can introduce arbitrary keys without schema edits; they will not be filtered.
+- **Invalid values never crash the run.** They produce a single warning per field and the key is dropped.
+- **Coerced values are normalized.** A YAML-quoted `"180"` for an `int` field arrives downstream as the integer `180`.
+- **The shipped `config.yaml` must round-trip cleanly.** The integration test `test_shipped_default_config_loads_without_warnings` will fail if a key is added to the YAML without a matching schema field, so the two stay in sync.
+
 This approach ensures that checks are easily configurable, making Prowler highly adaptable to different environments and requirements.

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

@@ -35,14 +35,28 @@ The bundled checks require the following read-only scopes:
 - `okta.policies.read`
 - `okta.brands.read`
 - `okta.apps.read`
+- `okta.authenticators.read`
+- `okta.networkZones.read`
+- `okta.apiTokens.read`
+- `okta.roles.read`
+- `okta.groups.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.authenticators.read` | Okta authenticator configuration, including Okta Verify and Smart Card IdP |
+| `okta.networkZones.read` | Network Zone inventory, anonymized-proxy blocklist checks, and API token Network Zone validation |
+| `okta.apiTokens.read` | API token metadata and token network conditions |
+| `okta.roles.read` | Admin role assignments for API token owners (both direct and group-inherited) |
+| `okta.groups.read` | Group memberships of API token owners, used to resolve admin roles inherited via group assignment (e.g. Super Admin granted through the default admin group) |
+| `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 +82,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.
@@ -122,7 +138,7 @@ Okta displays the private key **only once**. If you close the modal without copy
 
 ### 5. Grant the required OAuth scopes
 
-On the app, open the **Okta API Scopes** tab and click **Grant** on every scope Prowler needs. The bundled checks require `okta.policies.read`, `okta.brands.read`, and `okta.apps.read`.
+On the app, open the **Okta API Scopes** tab and click **Grant** on every scope Prowler needs. The bundled checks require `okta.policies.read`, `okta.brands.read`, `okta.apps.read`, `okta.authenticators.read`, `okta.networkZones.read`, `okta.apiTokens.read`, `okta.roles.read`, `okta.groups.read`, `okta.logStreams.read`, and `okta.idps.read`.
 
 ![Okta — grant OAuth scopes](/user-guide/providers/okta/images/grant-permissions.png)
 
@@ -158,8 +174,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.authenticators.read,okta.networkZones.read,okta.apiTokens.read,okta.roles.read,okta.groups.read,okta.logStreams.read,okta.idps.read"
+export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read,okta.authenticators.read,okta.networkZones.read,okta.apiTokens.read,okta.roles.read,okta.groups.read,okta.logStreams.read,okta.idps.read"
 
 uv run python prowler-cli.py okta
 ```
@@ -200,7 +216,7 @@ Prowler validates credentials at startup by listing one sign-on policy. This err
 
 Raised when the credential probe succeeds at the OAuth layer but the request is rejected because the service app lacks the required scope or admin role:
 
-- **`invalid_scope`** — one of the requested scopes (`okta.policies.read`, `okta.brands.read`, or `okta.apps.read`) is not granted on the service app. Grant the missing scope from **Okta API Scopes**.
+- **`invalid_scope`** — one of the requested scopes (`okta.policies.read`, `okta.brands.read`, `okta.apps.read`, `okta.authenticators.read`, `okta.networkZones.read`, `okta.apiTokens.read`, `okta.roles.read`, `okta.groups.read`, `okta.logStreams.read`, and `okta.idps.read`) is not granted on the service app. Grant the missing scope from **Okta API Scopes**.
 - **`Forbidden` / `not authorized`** — no admin role is assigned to the service app. Assign **Read-Only Administrator** (or **Super Administrator** for the first-party application checks) from **Admin roles**.
 
 ### Application-service checks return MANUAL on first-party apps

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

@@ -12,7 +12,7 @@ Set up authentication for Okta with the [Okta Authentication](/user-guide/provid
 
 - An Okta organization. The UI examples below use **Identity Engine** terminology such as **Global Session Policy**; Classic Engine exposes the equivalent sign-on policy concepts under older names.
 - A **Super Administrator** account on that organization for the one-time service-app setup.
-- An **API Services** app integration in the Okta Admin Console with the `okta.policies.read`, `okta.brands.read`, and `okta.apps.read` scopes granted and an admin role assigned. **Read-Only Administrator** covers every `signon` check and runs the per-app network-zone check against the apps the service app can see (under Read-Only Administrator that is typically only the service app's own row — the rest of the org's app inventory stays invisible). **Super Administrator** is required additionally to evaluate the five first-party application checks (Okta Admin Console / Okta Dashboard idle timeout, MFA, phishing-resistant authentication) and to widen the network-zone check to the full app inventory — see [Okta Authentication](/user-guide/providers/okta/authentication#required-admin-role) for the full breakdown.
+- An **API Services** app integration in the Okta Admin Console with the `okta.policies.read`, `okta.brands.read`, `okta.apps.read`, `okta.authenticators.read`, `okta.networkZones.read`, `okta.apiTokens.read`, `okta.roles.read`, `okta.groups.read`, `okta.logStreams.read`, and `okta.idps.read` scopes granted and an admin role assigned. **Read-Only Administrator** covers the Sign-On, Network, API Token, User, System Log, and Identity Provider checks, and runs the per-app application network-zone check against the apps the service app can see (under Read-Only Administrator that is typically only the service app's own row — the rest of the org's app inventory stays invisible). **Super Administrator** is required additionally to evaluate the five first-party application checks (Okta Admin Console / Okta Dashboard idle timeout, MFA, phishing-resistant authentication) and to widen the application network-zone check to the full app inventory — see [Okta Authentication](/user-guide/providers/okta/authentication#required-admin-role) for the full breakdown.
 - Python 3.10+ and Prowler 5.27.0 or later installed locally.
 
 <CardGroup cols={2}>
@@ -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.authenticators.read,okta.networkZones.read,okta.apiTokens.read,okta.roles.read,okta.groups.read,okta.logStreams.read,okta.idps.read"
+export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read,okta.authenticators.read,okta.networkZones.read,okta.apiTokens.read,okta.roles.read,okta.groups.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,16 @@ 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) |
+| **Authenticator**     | Password Policy controls plus Okta Verify FIPS and Smart Card IdP authenticator status                                                                |
+| **Network**           | Network Zone blocklists for anonymized proxy sources                                                                                                  |
+| **API Token**         | API token owner-role validation and Network Zone restrictions                                                                                         |
+| **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,22 +164,29 @@ 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, Authenticator, Network, API Token, User, System Log, and Identity Provider services:
 
 - `okta.policies.read`
 - `okta.brands.read`
 - `okta.apps.read`
+- `okta.authenticators.read`
+- `okta.networkZones.read`
+- `okta.apiTokens.read`
+- `okta.roles.read`
+- `okta.groups.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.
+The service app must have these scopes granted in the **Okta API Scopes** tab. `okta.groups.read` is required so the API token Super Admin check can resolve admin roles inherited via group membership; without it the check falls back to direct-only role assignments and emits a best-effort caveat. 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.
 
 When additional checks are enabled — or when running against a service app that exposes a different scope set — override the default with `OKTA_SCOPES` (comma-separated string for the env var) or `--okta-scopes` (space-separated list for the CLI):
 
 ```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.authenticators.read,okta.networkZones.read,okta.apiTokens.read,okta.roles.read,okta.groups.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.authenticators.read okta.networkZones.read okta.apiTokens.read okta.roles.read okta.groups.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" />
 

osv-scanner.toml

@@ -12,8 +12,9 @@ reason = """
 CVE-2025-45768 is disputed by the pyjwt maintainers. The advisory describes
 weak encryption, but the underlying issue is that callers may pick a short
 HMAC secret — key-length enforcement is the application's responsibility, not
-a defect in the library. We are on pyjwt 2.12.1 (latest at pin time) and
-enforce key strength in our own auth code, so this advisory does not apply.
+a defect in the library. We are on pyjwt 2.13.0 (which now also emits an
+InsecureKeyLengthWarning for short HMAC secrets) and enforce key strength in
+our own auth code, so this advisory does not apply.
 Re-evaluate when a non-disputed advisory or upstream fix lands.
 """
 

prowler/CHANGELOG.md

@@ -8,6 +8,39 @@ 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)
+- Okta authenticator and password policy checks for STIG-aligned hardening requirements [(#11465)](https://github.com/prowler-cloud/prowler/pull/11465)
+- Okta network zone check to detect whether anonymized proxy traffic is blocked [(#11463)](https://github.com/prowler-cloud/prowler/pull/11463)
+- Okta API token checks for super admin ownership and network zone restrictions [(#11464)](https://github.com/prowler-cloud/prowler/pull/11464)
+- Support for external/custom providers, checks, and compliance frameworks without modifying core code [(#10700)](https://github.com/prowler-cloud/prowler/pull/10700)
+- `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)
+- External multi-provider compliance frameworks can be registered via the `prowler.compliance.universal` entry point group [(#11490)](https://github.com/prowler-cloud/prowler/pull/11490)
+- AWS AI Security Framework support in the CLI dashboard [(#11475)](https://github.com/prowler-cloud/prowler/pull/11475)
+- `entra_service_principal_privileged_role_no_owners` check for M365 provider, failing when a service principal with a permanent Tier 0 directory role has owners on the service principal or its parent app registration [(#11070)](https://github.com/prowler-cloud/prowler/issues/11070)
+- `kms_key_rotation_max_90_days` check for GCP provider, verifying KMS customer-managed keys are rotated every 90 days or less in line with the CIS Benchmark [(#11516)](https://github.com/prowler-cloud/prowler/pull/11516)
+- `exchange_mailbox_primary_smtp_uses_custom_domain` check for M365 provider [(#11215)](https://github.com/prowler-cloud/prowler/pull/11215)
+
+### 🐞 Fixed
+
+- `load_and_validate_config_file` now unwraps namespaced config for every built-in and external provider, and no longer leaks the full file as the provider's config when the file is namespaced [(#10700)](https://github.com/prowler-cloud/prowler/pull/10700)
+- `entra_users_mfa_capable` no longer flags pre-provisioned users with future `employeeHireDate`; future-hire date comparisons now tolerate naive datetimes [(#11511)](https://github.com/prowler-cloud/prowler/pull/11511)
+- M365 Admin Center group enumeration now follows Microsoft Graph pagination so group-scoped checks include groups beyond the first page [(#11510)](https://github.com/prowler-cloud/prowler/pull/11510)
+- GCP `kms_key_rotation_enabled` check now only verifies that automatic key rotation is enabled (any interval) instead of enforcing a 90-day period, resolving the mismatch between the check and its documentation; the CIS, Prowler ThreatScore, and CCC requirements that mandate a 90-day maximum were remapped to the new `kms_key_rotation_max_90_days` check [(#11516)](https://github.com/prowler-cloud/prowler/pull/11516)
+
+---
+
+## [5.29.3] (Prowler v5.29.3)
+
+### 🐞 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)
+- GCP `logging_log_metric_filter_and_alert_*` checks now recognize organization-level aggregated sinks with `includeChildren=True`, no longer false-failing projects covered by a central bucket-scoped metric + alert [(#11488)](https://github.com/prowler-cloud/prowler/pull/11488)
+- 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)
+
+### 🔐 Security
+
+- `dulwich` from 0.23.0 to 1.2.5 and `pyjwt` from 2.12.1 to 2.13.0, patching `GHSA-897w-fcg9-f6xj` (arbitrary file write) and `PYSEC-2026-179` (HMAC/JWK key confusion) flagged by osv-scanner [(#11499)](https://github.com/prowler-cloud/prowler/pull/11499)
 
 ---
 
@@ -95,6 +128,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `entra_service_principal_no_secrets_for_permanent_tier0_roles` check for M365 provider [(#10788)](https://github.com/prowler-cloud/prowler/pull/10788)
 - `iam_user_access_not_stale_to_sagemaker` check for AWS provider with configurable `max_unused_sagemaker_access_days` (default 90) [(#11000)](https://github.com/prowler-cloud/prowler/pull/11000)
 - `cloudtrail_bedrock_logging_enabled` check for AWS provider [(#10858)](https://github.com/prowler-cloud/prowler/pull/10858)
+- Per-provider scan configuration schema with bounds validation that drops out-of-range values with a warning on config load [(#11518)](https://github.com/prowler-cloud/prowler/pull/11518)
 - Okta provider with OAuth 2.0 authentication and `signon_global_session_idle_timeout_15min` check [(#11079)](https://github.com/prowler-cloud/prowler/pull/11079)
 - `sagemaker_domain_sso_configured` check for AWS provider [(#11094)](https://github.com/prowler-cloud/prowler/pull/11094)
 - Scaleway provider with `iam_api_keys_no_root_owned` check [(#11166)](https://github.com/prowler-cloud/prowler/pull/11166)

prowler/__main__.py

@@ -10,7 +10,6 @@ from colorama import Fore, Style
 from colorama import init as colorama_init
 
 from prowler.config.config import (
-    EXTERNAL_TOOL_PROVIDERS,
     cloud_api_base_url,
     csv_file_suffix,
     get_available_compliance_frameworks,
@@ -205,9 +204,10 @@ def prowler():
     # We treat the compliance framework as another output format
     if compliance_framework:
         args.output_formats.extend(compliance_framework)
-    # If no input compliance framework, set all, unless a specific service or check is input
-    # Skip for IAC and LLM providers that don't use compliance frameworks
-    elif default_execution and provider not in ["iac", "llm"]:
+    # If no input compliance framework, set all, unless a specific service or check is input.
+    # Skip for tool-wrapper providers (iac, llm, image, and any external plug-in
+    # declaring `is_external_tool_provider = True`) — they don't use compliance frameworks.
+    elif default_execution and not Provider.is_tool_wrapper_provider(provider):
         args.output_formats.extend(get_available_compliance_frameworks(provider))
 
     # Set Logger configuration
@@ -245,7 +245,7 @@ def prowler():
     universal_frameworks = {}
 
     # Skip compliance frameworks for external-tool providers
-    if provider not in EXTERNAL_TOOL_PROVIDERS:
+    if not Provider.is_tool_wrapper_provider(provider):
         bulk_compliance_frameworks = Compliance.get_bulk(provider)
         # Complete checks metadata with the compliance framework specification
         bulk_checks_metadata = update_checks_metadata_with_compliance(
@@ -313,7 +313,7 @@ def prowler():
         sys.exit()
 
     # Skip service and check loading for external-tool providers
-    if provider not in EXTERNAL_TOOL_PROVIDERS:
+    if not Provider.is_tool_wrapper_provider(provider):
         # Import custom checks from folder
         if checks_folder:
             custom_checks = parse_checks_from_folder(global_provider, checks_folder)
@@ -436,6 +436,20 @@ def prowler():
         output_options = ScalewayOutputOptions(
             args, bulk_checks_metadata, global_provider.identity
         )
+    else:
+        # Dynamic fallback: any external/custom provider
+        try:
+            output_options = global_provider.get_output_options(
+                args, bulk_checks_metadata
+            )
+        except NotImplementedError:
+            # No provider-specific OutputOptions: use the generic default so the
+            # run still produces output instead of aborting.
+            from prowler.providers.common.models import default_output_options
+
+            output_options = default_output_options(
+                global_provider, args, bulk_checks_metadata
+            )
 
     # Run the quick inventory for the provider if available
     if hasattr(args, "quick_inventory") and args.quick_inventory:
@@ -445,7 +459,7 @@ def prowler():
     # Execute checks
     findings = []
 
-    if provider in EXTERNAL_TOOL_PROVIDERS:
+    if Provider.is_tool_wrapper_provider(provider):
         # For external-tool providers, run the scan directly
         if provider == "llm":
 
@@ -455,12 +469,19 @@ def prowler():
 
             findings = global_provider.run_scan(streaming_callback=streaming_callback)
         else:
-            # Original behavior for IAC and Image
-            try:
+            if provider == "image":
+                try:
+                    findings = global_provider.run()
+                except ImageBaseException as error:
+                    logger.critical(f"{error}")
+                    sys.exit(1)
+            else:
+                # IAC and external tool-wrapper providers registered via entry
+                # points. Unexpected failures propagate to the outer except
+                # Exception backstop further down in this file — keeping the
+                # branch free of an Image-specific catch that would otherwise
+                # mislead plug-in authors reading this code.
                 findings = global_provider.run()
-            except ImageBaseException as error:
-                logger.critical(f"{error}")
-                sys.exit(1)
             # Note: External tool providers don't support granular progress tracking since
             # they run external tools as a black box and return all findings at once.
             # Progress tracking would just be 0% → 100%.
@@ -1293,6 +1314,30 @@ def prowler():
                 )
                 generated_outputs["compliance"].append(generic_compliance)
                 generic_compliance.batch_write_data_to_file()
+    else:
+        # Dynamic fallback: any external/custom provider
+        try:
+            global_provider.generate_compliance_output(
+                finding_outputs,
+                bulk_compliance_frameworks,
+                input_compliance_frameworks,
+                output_options,
+                generated_outputs,
+            )
+        except NotImplementedError:
+            # Last resort: generic compliance
+            for compliance_name in input_compliance_frameworks:
+                filename = (
+                    f"{output_options.output_directory}/compliance/"
+                    f"{output_options.output_filename}_{compliance_name}.csv"
+                )
+                generic_compliance = GenericCompliance(
+                    findings=finding_outputs,
+                    compliance=bulk_compliance_frameworks[compliance_name],
+                    file_path=filename,
+                )
+                generated_outputs["compliance"].append(generic_compliance)
+                generic_compliance.batch_write_data_to_file()
 
     # AWS Security Hub Integration
     if provider == "aws":

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/compliance/gcp/ccc_gcp.json

@@ -889,7 +889,7 @@
         }
       ],
       "Checks": [
-        "kms_key_rotation_enabled"
+        "kms_key_rotation_max_90_days"
       ]
     },
     {

prowler/compliance/gcp/cis_2.0_gcp.json

@@ -150,7 +150,7 @@
       "Id": "1.10",
       "Description": "Google Cloud Key Management Service stores cryptographic keys in a hierarchical structure designed for useful and elegant access control management.   The format for the rotation schedule depends on the client library that is used. For the gcloud command-line tool, the next rotation time must be in `ISO` or `RFC3339` format, and the rotation period must be in the form `INTEGERUNIT`, where units can be one of seconds (s), minutes (m), hours (h) or days (d).",
       "Checks": [
-        "kms_key_rotation_enabled"
+        "kms_key_rotation_max_90_days"
       ],
       "Attributes": [
         {

prowler/compliance/gcp/cis_3.0_gcp.json

@@ -201,7 +201,7 @@
       "Id": "1.10",
       "Description": "Ensure KMS Encryption Keys Are Rotated Within a Period of 90 Days",
       "Checks": [
-        "kms_key_rotation_enabled"
+        "kms_key_rotation_max_90_days"
       ],
       "Attributes": [
         {

prowler/compliance/gcp/cis_4.0_gcp.json

@@ -201,7 +201,7 @@
       "Id": "1.10",
       "Description": "Ensure KMS Encryption Keys Are Rotated Within a Period of 90 Days",
       "Checks": [
-        "kms_key_rotation_enabled"
+        "kms_key_rotation_max_90_days"
       ],
       "Attributes": [
         {

prowler/compliance/gcp/prowler_threatscore_gcp.json

@@ -117,7 +117,7 @@
       "Id": "1.2.4",
       "Description": "Ensure KMS Encryption Keys Are Rotated Within a Period of 90 Days",
       "Checks": [
-        "kms_key_rotation_enabled"
+        "kms_key_rotation_max_90_days"
       ],
       "Attributes": [
         {

prowler/config/config.py

@@ -1,3 +1,4 @@
+import importlib.metadata
 import os
 import pathlib
 from datetime import datetime, timezone
@@ -85,13 +86,38 @@ class Provider(str, Enum):
 actual_directory = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
 
 
+def _get_ep_compliance_dirs() -> dict:
+    """Discover compliance directories from entry points. Returns {provider: path}."""
+    dirs = {}
+    for ep in importlib.metadata.entry_points(group="prowler.compliance"):
+        try:
+            module = ep.load()
+            if hasattr(module, "__path__"):
+                dirs[ep.name] = module.__path__[0]
+            elif hasattr(module, "__file__"):
+                dirs[ep.name] = os.path.dirname(module.__file__)
+        except Exception as error:
+            logger.warning(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+    return dirs
+
+
 def get_available_compliance_frameworks(provider=None):
     available_compliance_frameworks = []
-    providers = [p.value for p in Provider]
+    # Built-in compliance
+    compliance_base = f"{actual_directory}/../compliance"
     if provider:
         providers = [provider]
-    for current_provider in providers:
-        compliance_dir = f"{actual_directory}/../compliance/{current_provider}"
+    else:
+        # Scan compliance directory for all provider subdirectories
+        providers = []
+        if os.path.isdir(compliance_base):
+            for entry in os.scandir(compliance_base):
+                if entry.is_dir():
+                    providers.append(entry.name)
+    for prov in providers:
+        compliance_dir = f"{compliance_base}/{prov}"
         if not os.path.isdir(compliance_dir):
             continue
         with os.scandir(compliance_dir) as files:
@@ -100,7 +126,8 @@ def get_available_compliance_frameworks(provider=None):
                     available_compliance_frameworks.append(
                         file.name.removesuffix(".json")
                     )
-    # Also scan top-level compliance/ for multi-provider (universal) JSONs.
+    # Built-in multi-provider frameworks at top-level compliance/ directory.
+    # Placed before external entry points so built-ins win on name collisions.
     # When a specific provider was requested, only include the framework if it
     # declares support for that provider; otherwise include all universal frameworks.
     compliance_root = f"{actual_directory}/../compliance"
@@ -117,6 +144,43 @@ def get_available_compliance_frameworks(provider=None):
                             continue
                     if name not in available_compliance_frameworks:
                         available_compliance_frameworks.append(name)
+    # External per-provider compliance via entry points.
+    ep_dirs = _get_ep_compliance_dirs()
+    for prov, path in ep_dirs.items():
+        if provider and prov != provider:
+            continue
+        if os.path.isdir(path):
+            for file in os.scandir(path):
+                if file.is_file() and file.name.endswith(".json"):
+                    name = file.name.removesuffix(".json")
+                    if name not in available_compliance_frameworks:
+                        available_compliance_frameworks.append(name)
+    # External multi-provider frameworks via the dedicated universal group;
+    # filtered by supports_provider when a provider is given.
+    for ep in importlib.metadata.entry_points(group="prowler.compliance.universal"):
+        try:
+            module = ep.load()
+            path = (
+                module.__path__[0]
+                if hasattr(module, "__path__")
+                else os.path.dirname(module.__file__)
+            )
+        except Exception as error:
+            logger.warning(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            continue
+        if not os.path.isdir(path):
+            continue
+        for file in os.scandir(path):
+            if file.is_file() and file.name.endswith(".json"):
+                name = file.name.removesuffix(".json")
+                if provider:
+                    framework = load_compliance_framework_universal(file.path)
+                    if framework is None or not framework.supports_provider(provider):
+                        continue
+                if name not in available_compliance_frameworks:
+                    available_compliance_frameworks.append(name)
     return available_compliance_frameworks
 
 
@@ -224,24 +288,41 @@ def load_and_validate_config_file(provider: str, config_file_path: str) -> dict:
     Returns:
         dict: The configuration dictionary for the specified provider.
     """
+    # Imported lazily to avoid an import cycle: schemas may eventually want to
+    # import from prowler.config.config (e.g. for shared constants).
+    from prowler.config.schema.registry import SCHEMAS
+    from prowler.config.schema.validator import validate_provider_config
+
     try:
         with open(config_file_path, "r", encoding=encoding_format_utf_8) as f:
             config_file = yaml.safe_load(f)
 
-            # Not to introduce a breaking change, allow the old format config file without any provider keys
-            # and a new format with a key for each provider to include their configuration values within.
-            if any(
-                key in config_file
-                for key in ["aws", "gcp", "azure", "kubernetes", "m365"]
+            # Namespaced format: each provider has its own top-level key.
+            # Works for every built-in and every external plugin without a hardcoded list.
+            # Flat legacy format is AWS-only (historical, pre-multicloud). We identify it
+            # by the absence of nested-dict top-level values (namespaced files always
+            # have dict values; the legacy AWS format only has primitives/lists).
+            if (
+                isinstance(config_file, dict)
+                and provider in config_file
+                and isinstance(config_file[provider], dict)
             ):
-                config = config_file.get(provider, {})
+                config = config_file.get(provider, {}) or {}
+            elif (
+                isinstance(config_file, dict)
+                and config_file
+                and provider == "aws"
+                and not any(isinstance(v, dict) for v in config_file.values())
+            ):
+                config = config_file
             else:
-                config = config_file if config_file else {}
-                # Not to break Azure, K8s and GCP does not support or use the old config format
-                if provider in ["azure", "gcp", "kubernetes", "m365"]:
-                    config = {}
+                config = {}
 
-            return config
+            return validate_provider_config(
+                provider=provider,
+                raw=config,
+                schema_cls=SCHEMAS.get(provider),
+            )
 
     except FileNotFoundError as error:
         logger.error(

prowler/config/scan_config_schema.py

@@ -0,0 +1,106 @@
+"""Bridge between the Pydantic-based provider schemas in
+`prowler.config.schema` and the Prowler App backend (Django) + UI.
+
+The SDK runtime is intentionally LENIENT: invalid keys are dropped with a
+warning and downstream checks fall back to their defaults
+(`prowler.config.schema.validator.validate_provider_config`).
+
+The Prowler App, however, needs to surface those errors to the user when
+they save a Scan Config from the UI, and to expose the schema as JSON so
+the UI can validate live with `ajv`. This module provides:
+
+- `validate_scan_config(payload)` — STRICT: returns a list of
+  `{path, message}` errors without silently dropping anything. The DRF
+  serializer (`api/.../v1/serializers.py:validate_scan_config_payload`)
+  turns each entry into a `ValidationError`.
+
+- `SCAN_CONFIG_SCHEMA` — aggregated JSON Schema derived from the Pydantic
+  models via `model_json_schema()`. Served by the `/scan-configs/schema`
+  endpoint and consumed by the UI editor for in-editor live validation.
+"""
+
+from typing import Any
+
+from pydantic import ValidationError
+
+from prowler.config.schema.registry import SCHEMAS
+
+
+def _format_loc(loc: tuple) -> str:
+    """Render a Pydantic error location as `key[idx].nested`."""
+    parts: list[str] = []
+    for piece in loc:
+        if isinstance(piece, int):
+            if parts:
+                parts[-1] = f"{parts[-1]}[{piece}]"
+            else:
+                parts.append(f"[{piece}]")
+        else:
+            parts.append(str(piece))
+    return ".".join(parts) if parts else "<root>"
+
+
+def validate_scan_config(payload: Any) -> list[dict]:
+    """Validate a scan config payload against the registered provider schemas.
+
+    Strict by design: every Pydantic violation surfaces as a `{path, message}`
+    entry so the caller can decide how to present it. Unknown provider
+    sections are accepted (consistent with `additionalProperties: True` at
+    the top level — the SDK simply has no opinion on them).
+    """
+    if not isinstance(payload, dict):
+        return [
+            {
+                "path": "<root>",
+                "message": "Scan config must be a mapping with provider sections.",
+            }
+        ]
+
+    errors: list[dict] = []
+    for provider, section in payload.items():
+        schema_cls = SCHEMAS.get(provider)
+        if schema_cls is None:
+            # Unknown provider type: tolerated. The SDK will simply ignore it.
+            continue
+        if not isinstance(section, dict):
+            errors.append(
+                {
+                    "path": str(provider),
+                    "message": "section must be a mapping.",
+                }
+            )
+            continue
+        try:
+            schema_cls.model_validate(section)
+        except ValidationError as exc:
+            for err in exc.errors():
+                loc = err.get("loc") or ()
+                path = _format_loc((str(provider), *loc))
+                errors.append(
+                    {
+                        "path": path,
+                        "message": err.get("msg", "validation error"),
+                    }
+                )
+    return errors
+
+
+def _build_aggregated_schema() -> dict:
+    """Compose one JSON Schema per provider into a single top-level schema.
+
+    The output mirrors the layout of `prowler/config/config.yaml` (a mapping
+    keyed by provider type) and is what the UI consumes via `ajv`.
+    """
+    properties: dict[str, dict] = {}
+    for provider, schema_cls in SCHEMAS.items():
+        properties[provider] = schema_cls.model_json_schema()
+    return {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        "title": "Prowler Scan Config",
+        "type": "object",
+        "additionalProperties": True,
+        "properties": properties,
+    }
+
+
+SCAN_CONFIG_SCHEMA: dict = _build_aggregated_schema()

prowler/config/schema/aws.py

@@ -0,0 +1,449 @@
+"""AWS provider config schema.
+
+Bounds on every field are intentionally conservative: they are not the
+absolute service maxima but the values that produce a useful security
+check. A user is free to keep the built-in default by omitting the key —
+out-of-range values are dropped with a warning at SDK runtime, and
+rejected at the Prowler App backend.
+
+Whenever an upper bound is uncertain, the cap is set to a value that
+still keeps the check meaningful (e.g. a 10-year window for date-based
+thresholds) and avoids ints that obviously break downstream maths
+(`min_kinesis_stream_retention_hours = 99999`).
+"""
+
+from ipaddress import ip_network
+from typing import Annotated, Literal, Optional
+
+from pydantic import AfterValidator, Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+# ---- Reusable constants -----------------------------------------------------
+
+# CloudWatch Logs only accepts these retention values (in days). Anything else
+# is silently coerced to the next valid value by the API — we reject upfront.
+_CLOUDWATCH_RETENTION_DAYS = (
+    1,
+    3,
+    5,
+    7,
+    14,
+    30,
+    60,
+    90,
+    120,
+    150,
+    180,
+    365,
+    400,
+    545,
+    731,
+    1827,
+    2192,
+    2557,
+    2922,
+    3288,
+    3653,
+)
+
+_VALID_CW_RETENTION_LITERAL = Literal[
+    1,
+    3,
+    5,
+    7,
+    14,
+    30,
+    60,
+    90,
+    120,
+    150,
+    180,
+    365,
+    400,
+    545,
+    731,
+    1827,
+    2192,
+    2557,
+    2922,
+    3288,
+    3653,
+]
+
+
+# ---- Custom validators ------------------------------------------------------
+
+
+def _validate_port_range(v: Optional[list[int]]) -> Optional[list[int]]:
+    if v is None:
+        return v
+    for port in v:
+        if not 1 <= port <= 65535:
+            raise ValueError(f"port {port} is outside the valid range 1..65535")
+    return v
+
+
+def _validate_account_ids(v: Optional[list[str]]) -> Optional[list[str]]:
+    if v is None:
+        return v
+    for account_id in v:
+        if not (account_id.isdigit() and len(account_id) == 12):
+            raise ValueError(
+                f"trusted_account_ids entry {account_id!r} is not a 12-digit AWS account id"
+            )
+    return v
+
+
+def _validate_trusted_ips(v: Optional[list[str]]) -> Optional[list[str]]:
+    if v is None:
+        return v
+    for entry in v:
+        try:
+            ip_network(entry, strict=False)
+        except ValueError as exc:
+            raise ValueError(
+                f"trusted_ips entry {entry!r} is not a valid IP or CIDR ({exc})"
+            ) from exc
+    return v
+
+
+def _validate_semver(v: Optional[str]) -> Optional[str]:
+    """Accept "1.4.0" style strings (used by Fargate platform versions)."""
+    if v is None:
+        return v
+    parts = v.split(".")
+    if len(parts) != 3 or not all(p.isdigit() for p in parts):
+        raise ValueError(f"{v!r} is not a valid semantic version (expected X.Y.Z)")
+    return v
+
+
+def _validate_eks_minor(v: Optional[str]) -> Optional[str]:
+    """Accept "1.28" style strings (EKS minor versions)."""
+    if v is None:
+        return v
+    parts = v.split(".")
+    if len(parts) != 2 or not all(p.isdigit() for p in parts):
+        raise ValueError(f"{v!r} is not a valid EKS version (expected X.Y)")
+    return v
+
+
+# ---- Nested models ----------------------------------------------------------
+
+
+class _DetectSecretsPlugin(ProviderConfigBase):
+    """One entry inside ``detect_secrets_plugins``.
+
+    Only ``name`` is required by the upstream library. ``limit`` is used by
+    the entropy detectors. Any other plugin-specific kwarg is preserved by
+    the ``extra="allow"`` policy inherited from ProviderConfigBase.
+    """
+
+    name: str
+    limit: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        le=10.0,
+        description=(
+            "Entropy threshold for detect-secrets entropy plugins. Range: 0..10 "
+            "(Shannon entropy is bounded by log2(256)=8; >10 is meaningless)."
+        ),
+    )
+
+
+# ---- Main schema ------------------------------------------------------------
+
+
+class AWSProviderConfig(ProviderConfigBase):
+    # --- IAM ---------------------------------------------------------------
+    mute_non_default_regions: Optional[bool] = None
+    disallowed_regions: Optional[list[str]] = None
+    max_unused_access_keys_days: Optional[int] = Field(
+        default=None,
+        ge=30,
+        le=180,
+        description=(
+            "Days an IAM user access key can stay unused before being flagged. "
+            "Range: 30..180 days (CIS AWS 1.13 recommends 45; NIST IA-5 ≤90)."
+        ),
+    )
+    max_console_access_days: Optional[int] = Field(
+        default=None,
+        ge=30,
+        le=180,
+        description=(
+            "Days an IAM console password can stay unused before being flagged. "
+            "Range: 30..180 days (CIS AWS 1.12 recommends 45)."
+        ),
+    )
+    max_unused_sagemaker_access_days: Optional[int] = Field(
+        default=None,
+        ge=7,
+        le=180,
+        description=(
+            "Days a SageMaker user access key can stay unused. Range: 7..180 "
+            "(SageMaker tokens are usually high-privilege over S3/KMS)."
+        ),
+    )
+
+    # --- EC2 ---------------------------------------------------------------
+    shodan_api_key: Optional[str] = Field(
+        default=None,
+        max_length=512,
+        description="API key for Shodan lookups on EC2 public IPs.",
+    )
+    max_security_group_rules: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=1000,
+        description="Max ingress+egress rules per security group. AWS hard limit is 1000.",
+    )
+    max_ec2_instance_age_in_days: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=1095,
+        description=(
+            "Days an EC2 instance can run before being flagged as old. "
+            "Range: 1..1095 (3 years; instances should be refreshed for patching "
+            "per NIST CM-3 — anything older is a security smell)."
+        ),
+    )
+    ec2_allowed_interface_types: Optional[list[str]] = None
+    ec2_allowed_instance_owners: Optional[list[str]] = None
+    ec2_high_risk_ports: Annotated[
+        Optional[list[int]], AfterValidator(_validate_port_range)
+    ] = Field(
+        default=None,
+        description="TCP/UDP ports considered high-risk when reachable from the Internet (1..65535; port 0 is reserved).",
+    )
+
+    # --- ECS ---------------------------------------------------------------
+    fargate_linux_latest_version: Annotated[
+        Optional[str], AfterValidator(_validate_semver)
+    ] = Field(default=None, description="Fargate Linux platform version (X.Y.Z).")
+    fargate_windows_latest_version: Annotated[
+        Optional[str], AfterValidator(_validate_semver)
+    ] = Field(default=None, description="Fargate Windows platform version (X.Y.Z).")
+
+    # --- Cross-account trust ----------------------------------------------
+    trusted_account_ids: Annotated[
+        Optional[list[str]], AfterValidator(_validate_account_ids)
+    ] = Field(
+        default=None,
+        description="Additional 12-digit AWS account IDs trusted by cross-account checks.",
+    )
+    trusted_ips: Annotated[
+        Optional[list[str]], AfterValidator(_validate_trusted_ips)
+    ] = Field(
+        default=None,
+        description="IPv4/IPv6 addresses or CIDR ranges that are NOT considered public.",
+    )
+
+    # --- CloudWatch / CloudFormation --------------------------------------
+    log_group_retention_days: Optional[_VALID_CW_RETENTION_LITERAL] = Field(
+        default=None,
+        description=(
+            "Required CloudWatch Logs retention in days. Must match one of the "
+            f"values accepted by the AWS API: {list(_CLOUDWATCH_RETENTION_DAYS)}."
+        ),
+    )
+    recommended_cdk_bootstrap_version: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=100,
+        description="Min CDK bootstrap version expected on the account.",
+    )
+
+    # --- AppStream --------------------------------------------------------
+    max_idle_disconnect_timeout_in_seconds: Optional[int] = Field(
+        default=None,
+        ge=60,
+        le=1800,
+        description=(
+            "AppStream idle disconnect timeout (seconds). Range: 60..1800 "
+            "(NIST AC-12: sensitive sessions ≤15 min — cap at 30 min)."
+        ),
+    )
+    max_disconnect_timeout_in_seconds: Optional[int] = Field(
+        default=None,
+        ge=60,
+        le=3600,
+        description="AppStream disconnect timeout (seconds). Range: 60..3600.",
+    )
+    max_session_duration_seconds: Optional[int] = Field(
+        default=None,
+        ge=600,
+        le=86400,
+        description=(
+            "AppStream max session duration (seconds). Range: 600..86400 "
+            "(10 min .. 24 h — AWS AppStream hard limit per session)."
+        ),
+    )
+
+    # --- Lambda -----------------------------------------------------------
+    obsolete_lambda_runtimes: Optional[list[str]] = None
+    lambda_min_azs: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=6,
+        description="Min number of AZs a VPC-bound Lambda must span. Range: 1..6.",
+    )
+
+    # --- Organizations ----------------------------------------------------
+    organizations_enabled_regions: Optional[list[str]] = None
+    organizations_trusted_delegated_administrators: Annotated[
+        Optional[list[str]], AfterValidator(_validate_account_ids)
+    ] = None
+    organizations_trusted_ids: Optional[list[str]] = None
+
+    # --- ECR --------------------------------------------------------------
+    ecr_repository_vulnerability_minimum_severity: Optional[
+        Literal["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFORMATIONAL"]
+    ] = Field(
+        default=None,
+        description="Highest severity tolerated for ECR images.",
+    )
+
+    # --- Trusted Advisor --------------------------------------------------
+    verify_premium_support_plans: Optional[bool] = None
+
+    # --- CloudTrail threat detection: privilege escalation ----------------
+    threat_detection_privilege_escalation_threshold: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        le=1.0,
+        description="Fraction of suspicious actions that triggers the priv-esc detection.",
+    )
+    threat_detection_privilege_escalation_minutes: Optional[int] = Field(
+        default=None,
+        ge=5,
+        le=43200,
+        description=(
+            "Lookback window (minutes) for priv-esc detection. Range: 5..43200 "
+            "(under 5 min the signal is dominated by false positives)."
+        ),
+    )
+    threat_detection_privilege_escalation_actions: Optional[list[str]] = None
+
+    # --- CloudTrail threat detection: enumeration -------------------------
+    threat_detection_enumeration_threshold: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        le=1.0,
+        description="Fraction of suspicious actions that triggers the enumeration detection.",
+    )
+    threat_detection_enumeration_minutes: Optional[int] = Field(
+        default=None,
+        ge=5,
+        le=43200,
+        description="Lookback window (minutes) for enumeration detection. Range: 5..43200.",
+    )
+    threat_detection_enumeration_actions: Optional[list[str]] = None
+
+    # --- CloudTrail threat detection: LLM jacking -------------------------
+    threat_detection_llm_jacking_threshold: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        le=1.0,
+        description="Fraction of suspicious actions that triggers the LLM-jacking detection.",
+    )
+    threat_detection_llm_jacking_minutes: Optional[int] = Field(
+        default=None,
+        ge=5,
+        le=43200,
+        description="Lookback window (minutes) for LLM-jacking detection. Range: 5..43200.",
+    )
+    threat_detection_llm_jacking_actions: Optional[list[str]] = None
+
+    # --- RDS --------------------------------------------------------------
+    check_rds_instance_replicas: Optional[bool] = None
+
+    # --- ACM --------------------------------------------------------------
+    days_to_expire_threshold: Optional[int] = Field(
+        default=None,
+        ge=7,
+        le=365,
+        description=(
+            "Days before certificate expiration to flag. Range: 7..365 "
+            "(PCI-DSS 4.2.1.1: alert ≥30 days before expiry; <7 days is too "
+            "tight to actually act on)."
+        ),
+    )
+    insecure_key_algorithms: Optional[list[str]] = None
+
+    # --- EKS --------------------------------------------------------------
+    eks_required_log_types: Optional[
+        list[
+            Literal[
+                "api",
+                "audit",
+                "authenticator",
+                "controllerManager",
+                "scheduler",
+            ]
+        ]
+    ] = Field(
+        default=None,
+        description="EKS control plane log types that must be enabled.",
+    )
+    eks_cluster_oldest_version_supported: Annotated[
+        Optional[str], AfterValidator(_validate_eks_minor)
+    ] = Field(
+        default=None,
+        description='Minimum supported EKS minor version, expected as "X.Y".',
+    )
+
+    # --- CodeBuild --------------------------------------------------------
+    excluded_sensitive_environment_variables: Optional[list[str]] = None
+    codebuild_github_allowed_organizations: Optional[list[str]] = None
+
+    # --- ELB / ELBv2 ------------------------------------------------------
+    elb_min_azs: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=6,
+        description="Min AZs a Classic ELB must span. Range: 1..6.",
+    )
+    elbv2_min_azs: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=6,
+        description="Min AZs an Application/Network LB must span. Range: 1..6.",
+    )
+
+    # --- ElastiCache -----------------------------------------------------
+    minimum_snapshot_retention_period: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=35,
+        description="Days an ElastiCache backup must be retained. Range: 1..35 (service hard limit).",
+    )
+
+    # --- Secrets ---------------------------------------------------------
+    secrets_ignore_patterns: Optional[list[str]] = None
+    max_days_secret_unused: Optional[int] = Field(
+        default=None,
+        ge=7,
+        le=365,
+        description="Days a Secrets Manager secret can stay unused. Range: 7..365.",
+    )
+    max_days_secret_unrotated: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=180,
+        description=(
+            "Days a Secrets Manager secret can go without rotation. Range: 1..180 "
+            "(NIST IA-5: rotate quarterly; CIS recommends ≤90)."
+        ),
+    )
+
+    # --- Kinesis ---------------------------------------------------------
+    min_kinesis_stream_retention_hours: Optional[int] = Field(
+        default=None,
+        ge=24,
+        le=8760,
+        description="Hours of Kinesis stream retention. Range: 24..8760 (1 day .. 1 year).",
+    )
+
+    # --- detect-secrets plugin list -------------------------------------
+    detect_secrets_plugins: Optional[list[_DetectSecretsPlugin]] = None

prowler/config/schema/azure.py

@@ -0,0 +1,91 @@
+"""Azure provider config schema with safety bounds.
+
+Bounds aim for values that produce a meaningful security check; out-of-range
+values are dropped (SDK runtime) or rejected (Prowler App backend).
+"""
+
+from typing import Annotated, Literal, Optional
+
+from pydantic import AfterValidator, Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+
+def _validate_dotted_version(v: Optional[str]) -> Optional[str]:
+    """Accept ``"8.2"``, ``"3.12"``, ``"17"`` style version strings.
+
+    Used by App Service language version fields where the upstream APIs
+    accept either ``MAJOR`` or ``MAJOR.MINOR`` notation.
+    """
+    if v is None:
+        return v
+    parts = v.split(".")
+    if not (1 <= len(parts) <= 2) or not all(p.isdigit() for p in parts):
+        raise ValueError(f"{v!r} is not a valid version (expected 'X' or 'X.Y')")
+    return v
+
+
+class AzureProviderConfig(ProviderConfigBase):
+    # --- Network ---------------------------------------------------------
+    shodan_api_key: Optional[str] = Field(
+        default=None,
+        max_length=512,
+        description="API key for Shodan lookups on Azure public IPs.",
+    )
+
+    # --- Defender --------------------------------------------------------
+    defender_attack_path_minimal_risk_level: Optional[
+        Literal["Low", "Medium", "High", "Critical"]
+    ] = Field(
+        default=None,
+        description="Minimum attack-path risk level worth a notification.",
+    )
+
+    # --- App Service ----------------------------------------------------
+    php_latest_version: Annotated[
+        Optional[str], AfterValidator(_validate_dotted_version)
+    ] = Field(default=None, description='PHP minimum acceptable version, e.g. "8.2".')
+    python_latest_version: Annotated[
+        Optional[str], AfterValidator(_validate_dotted_version)
+    ] = Field(
+        default=None, description='Python minimum acceptable version, e.g. "3.12".'
+    )
+    java_latest_version: Annotated[
+        Optional[str], AfterValidator(_validate_dotted_version)
+    ] = Field(default=None, description='Java minimum acceptable version, e.g. "17".')
+
+    # --- SQL ------------------------------------------------------------
+    recommended_minimal_tls_versions: Optional[list[Literal["1.2", "1.3"]]] = Field(
+        default=None,
+        description="TLS versions accepted on Azure SQL Server.",
+    )
+
+    # --- Virtual Machines -----------------------------------------------
+    desired_vm_sku_sizes: Optional[list[str]] = None
+    vm_backup_min_daily_retention_days: Optional[int] = Field(
+        default=None,
+        ge=7,
+        le=9999,
+        description=(
+            "Min daily backup retention days. Range: 7..9999 "
+            "(Azure Backup hard limit; <7 days defeats DR/ransomware recovery)."
+        ),
+    )
+
+    # --- API Management threat detection (LLM jacking) -----------------
+    apim_threat_detection_llm_jacking_threshold: Optional[float] = Field(
+        default=None,
+        ge=0.0,
+        le=1.0,
+        description="Fraction of suspicious actions that triggers the detection.",
+    )
+    apim_threat_detection_llm_jacking_minutes: Optional[int] = Field(
+        default=None,
+        ge=5,
+        le=43200,
+        description=(
+            "Lookback window (minutes) for LLM-jacking detection. Range: 5..43200 "
+            "(under 5 min the signal is dominated by false positives)."
+        ),
+    )
+    apim_threat_detection_llm_jacking_actions: Optional[list[str]] = None

prowler/config/schema/base.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel, ConfigDict
+
+
+class ProviderConfigBase(BaseModel):
+    """Base for every provider config schema.
+
+    ``extra="allow"`` is REQUIRED for backwards compatibility: third-party
+    check plugins frequently introduce config keys we do not know about,
+    and pre-existing user configs may carry deprecated keys. Validation
+    must never reject these.
+    """
+
+    model_config = ConfigDict(
+        extra="allow",
+        str_strip_whitespace=True,
+        validate_assignment=False,
+    )

prowler/config/schema/cloudflare.py

@@ -0,0 +1,18 @@
+"""Cloudflare provider config schema with safety bounds."""
+
+from typing import Optional
+
+from pydantic import Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+
+class CloudflareProviderConfig(ProviderConfigBase):
+    max_retries: Optional[int] = Field(
+        default=None,
+        ge=0,
+        le=10,
+        description=(
+            "Max retries for Cloudflare API requests. Range: 0..10 (0 disables retries)."
+        ),
+    )

prowler/config/schema/gcp.py

@@ -0,0 +1,45 @@
+"""GCP provider config schema with safety bounds."""
+
+from typing import Optional
+
+from pydantic import Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+
+class GCPProviderConfig(ProviderConfigBase):
+    shodan_api_key: Optional[str] = Field(
+        default=None,
+        max_length=512,
+        description="API key for Shodan lookups on GCP public IPs.",
+    )
+    mig_min_zones: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=5,
+        description="Min zones a Managed Instance Group must span. Range: 1..5.",
+    )
+    max_snapshot_age_days: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=1095,
+        description=(
+            "Days a disk snapshot can age before being flagged. Range: 1..1095 "
+            "(3 years; older snapshots typically miss data-class compliance)."
+        ),
+    )
+    max_unused_account_days: Optional[int] = Field(
+        default=None,
+        ge=30,
+        le=365,
+        description=(
+            "Days a service account or user-managed key can stay unused. "
+            "Range: 30..365."
+        ),
+    )
+    storage_min_retention_days: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=3650,
+        description="Min retention period on Cloud Storage buckets. Range: 1..3650.",
+    )

prowler/config/schema/github.py

@@ -0,0 +1,20 @@
+"""GitHub provider config schema with safety bounds."""
+
+from typing import Optional
+
+from pydantic import Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+
+class GitHubProviderConfig(ProviderConfigBase):
+    inactive_not_archived_days_threshold: Optional[int] = Field(
+        default=None,
+        ge=30,
+        le=3650,
+        description=(
+            "Days a repository can stay inactive without being archived before "
+            "being flagged. Range: 30..3650 (CIS GitHub recommends 180; "
+            "<30 days produces false positives on seasonal projects)."
+        ),
+    )

prowler/config/schema/kubernetes.py

@@ -0,0 +1,45 @@
+"""Kubernetes provider config schema with safety bounds."""
+
+from typing import Optional
+
+from pydantic import Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+
+class KubernetesProviderConfig(ProviderConfigBase):
+    audit_log_maxbackup: Optional[int] = Field(
+        default=None,
+        ge=2,
+        le=1000,
+        description=(
+            "API server audit log file rotations to keep. Range: 2..1000 "
+            "(CIS Kubernetes 1.2.18 recommends ≥10)."
+        ),
+    )
+    audit_log_maxsize: Optional[int] = Field(
+        default=None,
+        ge=10,
+        le=10000,
+        description=(
+            "Max MB per audit log file before rotation. Range: 10..10000 MB "
+            "(CIS Kubernetes 1.2.19 recommends ≥100 MB)."
+        ),
+    )
+    audit_log_maxage: Optional[int] = Field(
+        default=None,
+        ge=7,
+        le=3650,
+        description=(
+            "Days an audit log file is retained. Range: 7..3650 "
+            "(CIS Kubernetes 1.2.17 recommends ≥30 days)."
+        ),
+    )
+    apiserver_strong_ciphers: Optional[list[str]] = Field(
+        default=None,
+        description="Whitelist of strong TLS cipher suites required on the API server.",
+    )
+    kubelet_strong_ciphers: Optional[list[str]] = Field(
+        default=None,
+        description="Whitelist of strong TLS cipher suites required on kubelet.",
+    )

prowler/config/schema/m365.py

@@ -0,0 +1,54 @@
+"""M365 provider config schema with safety bounds."""
+
+from typing import Optional
+
+from pydantic import Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+
+class M365ProviderConfig(ProviderConfigBase):
+    # --- Entra (sign-in policy) ----------------------------------------
+    sign_in_frequency: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=168,
+        description=(
+            "Hours between forced sign-ins for admin users. Range: 1..168 (1 h .. 7 days). "
+            "Microsoft Conditional Access baseline for admin roles is ≤24 h."
+        ),
+    )
+
+    # --- Teams ---------------------------------------------------------
+    allowed_cloud_storage_services: Optional[list[str]] = Field(
+        default=None,
+        description="External cloud storage services allowed in Teams.",
+    )
+
+    # --- Exchange ------------------------------------------------------
+    recommended_mailtips_large_audience_threshold: Optional[int] = Field(
+        default=None,
+        ge=5,
+        le=10000,
+        description=(
+            "Recipient count that should trigger a 'large audience' MailTip. "
+            "Range: 5..10000 (Microsoft default 25)."
+        ),
+    )
+
+    # --- Defender malware policy --------------------------------------
+    default_recommended_extensions: Optional[list[str]] = Field(
+        default=None,
+        description="File extensions blocked by the malware policy.",
+    )
+
+    # --- Mailbox auditing ---------------------------------------------
+    audit_log_age: Optional[int] = Field(
+        default=None,
+        ge=30,
+        le=3650,
+        description=(
+            "Days mailbox audit logs must be retained. Range: 30..3650 "
+            "(M365 E3 default is 90 days; SEC/FINRA require ≥7 years)."
+        ),
+    )

prowler/config/schema/mongodbatlas.py

@@ -0,0 +1,19 @@
+"""MongoDB Atlas provider config schema with safety bounds."""
+
+from typing import Optional
+
+from pydantic import Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+
+class MongoDBAtlasProviderConfig(ProviderConfigBase):
+    max_service_account_secret_validity_hours: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=720,
+        description=(
+            "Max hours a service account secret can stay valid. "
+            "Range: 1..720 (1 h .. 30 days)."
+        ),
+    )

prowler/config/schema/registry.py

@@ -0,0 +1,28 @@
+"""Mapping of provider name to its Pydantic schema class.
+
+Kept in its own module so the validator stays free of provider-schema imports
+and callers pay the import cost only when they actually need the registry.
+"""
+
+from prowler.config.schema.aws import AWSProviderConfig
+from prowler.config.schema.azure import AzureProviderConfig
+from prowler.config.schema.base import ProviderConfigBase
+from prowler.config.schema.cloudflare import CloudflareProviderConfig
+from prowler.config.schema.gcp import GCPProviderConfig
+from prowler.config.schema.github import GitHubProviderConfig
+from prowler.config.schema.kubernetes import KubernetesProviderConfig
+from prowler.config.schema.m365 import M365ProviderConfig
+from prowler.config.schema.mongodbatlas import MongoDBAtlasProviderConfig
+from prowler.config.schema.vercel import VercelProviderConfig
+
+SCHEMAS: dict[str, type[ProviderConfigBase]] = {
+    "aws": AWSProviderConfig,
+    "azure": AzureProviderConfig,
+    "gcp": GCPProviderConfig,
+    "kubernetes": KubernetesProviderConfig,
+    "m365": M365ProviderConfig,
+    "github": GitHubProviderConfig,
+    "mongodbatlas": MongoDBAtlasProviderConfig,
+    "cloudflare": CloudflareProviderConfig,
+    "vercel": VercelProviderConfig,
+}

prowler/config/schema/validator.py

@@ -0,0 +1,61 @@
+from typing import Any
+
+from pydantic import ValidationError
+
+from prowler.config.schema.base import ProviderConfigBase
+from prowler.lib.logger import logger
+
+
+def validate_provider_config(
+    provider: str,
+    raw: Any,
+    schema_cls: type[ProviderConfigBase] | None,
+) -> dict:
+    """Validate a provider's config dict against its Pydantic schema.
+
+    Behavior is intentionally lenient to preserve backwards compatibility:
+
+    - If ``raw`` is not a dict, return an empty dict (mirrors prior loader).
+    - If no schema is registered for ``provider``, return ``raw`` untouched.
+    - On validation errors, log one WARNING per offending field, DROP those
+      keys from the result, and continue. Consumers fall back to their own
+      hard-coded defaults via ``audit_config.get(key, default)``.
+    - Coerced values (e.g. ``"180"`` -> ``180``) replace the user's input
+      so that downstream checks never receive a wrongly-typed value.
+    """
+    if not isinstance(raw, dict):
+        return {}
+
+    if schema_cls is None:
+        return raw
+
+    try:
+        model = schema_cls.model_validate(raw)
+        return model.model_dump(exclude_unset=True)
+    except ValidationError as exc:
+        bad_keys: set[str] = set()
+        for err in exc.errors():
+            loc = err.get("loc") or ()
+            if not loc:
+                continue
+            key = loc[0]
+            if not isinstance(key, str):
+                continue
+            bad_keys.add(key)
+            logger.warning(
+                f"prowler.config[{provider}.{key}] = {raw.get(key)!r} is invalid "
+                f"({err.get('msg', 'validation error')}); the value will be ignored "
+                f"and the built-in default will be used."
+            )
+
+        cleaned = {k: v for k, v in raw.items() if k not in bad_keys}
+        try:
+            model = schema_cls.model_validate(cleaned)
+            return model.model_dump(exclude_unset=True)
+        except ValidationError as exc2:
+            logger.error(
+                f"prowler.config[{provider}] could not be revalidated after dropping "
+                f"invalid keys ({bad_keys}); passing through the cleaned dict as-is. "
+                f"Underlying errors: {exc2.errors()}"
+            )
+            return cleaned

prowler/config/schema/vercel.py

@@ -0,0 +1,61 @@
+"""Vercel provider config schema with safety bounds."""
+
+from typing import Optional
+
+from pydantic import Field
+
+from prowler.config.schema.base import ProviderConfigBase
+
+
+class VercelProviderConfig(ProviderConfigBase):
+    stable_branches: Optional[list[str]] = Field(
+        default=None,
+        description="Branches considered stable for production deployments.",
+    )
+    days_to_expire_threshold: Optional[int] = Field(
+        default=None,
+        ge=7,
+        le=365,
+        description=(
+            "Days before token/certificate expiration to flag. Range: 7..365 "
+            "(PCI-DSS 4.2.1.1: alert ≥30 days before expiry)."
+        ),
+    )
+    stale_token_threshold_days: Optional[int] = Field(
+        default=None,
+        ge=30,
+        le=3650,
+        description=(
+            "Days of inactivity before a token is considered stale. Range: 30..3650 "
+            "(NIST AC-2(3) typical window 30..90 days)."
+        ),
+    )
+    stale_invitation_threshold_days: Optional[int] = Field(
+        default=None,
+        ge=7,
+        le=365,
+        description=(
+            "Days a pending invitation can stay open. Range: 7..365 "
+            "(OWASP ASVS 2.7.1 recommends short-lived invitations)."
+        ),
+    )
+    max_owner_percentage: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=50,
+        description=(
+            "Max percentage of team members that can have the OWNER role. "
+            "Range: 1..50 (PoLP — having >50% of a team as OWNER defeats RBAC; "
+            "industry guidance recommends ≤25%)."
+        ),
+    )
+    max_owners: Optional[int] = Field(
+        default=None,
+        ge=1,
+        le=1000,
+        description="Absolute max owners (overrides percentage for large teams). Range: 1..1000.",
+    )
+    secret_suffixes: Optional[list[str]] = Field(
+        default=None,
+        description="Suffixes that mark a project env var as secret-like.",
+    )

prowler/lib/check/check.py

@@ -1,4 +1,6 @@
 import importlib
+import importlib.metadata
+import importlib.util
 import json
 import os
 import re
@@ -19,6 +21,7 @@ from prowler.lib.check.utils import recover_checks_from_provider
 from prowler.lib.logger import logger
 from prowler.lib.outputs.outputs import report
 from prowler.lib.utils.utils import open_file, parse_json_file, print_boxes
+from prowler.providers.common.builtin import is_builtin_provider
 from prowler.providers.common.models import Audit_Metadata
 
 
@@ -385,6 +388,45 @@ def import_check(check_path: str) -> ModuleType:
     return lib
 
 
+def _resolve_check_module(
+    provider_type: str, service: str, check_name: str
+) -> ModuleType:
+    """Resolve and import a check module.
+
+    Built-in wins on CheckID collision. Plug-ins are first-class extenders
+    (they can add new checks under new CheckIDs) but cannot override
+    existing built-ins — a security tool prefers fail-loud predictability
+    over silent overrides. CheckMetadata.get_bulk() applies the same
+    precedence on the metadata side (first-write-wins) and emits a warning
+    when a plug-in tries to override, so the user knows their plug-in
+    duplicate is being ignored and can rename it.
+
+    Gates the built-in branch on `is_builtin_provider(provider_type)` —
+    calling `find_spec` on `prowler.providers.{provider_type}.services...`
+    directly would propagate `ModuleNotFoundError` for external providers
+    (their parent package `prowler.providers.{provider_type}` does not
+    exist) instead of returning None. The leaf helper encapsulates the
+    safe lookup, so external providers go straight to entry points. For
+    built-ins we still use `find_spec` to distinguish "check doesn't
+    exist" from "check exists but failed to import" (broken transitive
+    dep, etc.).
+    """
+    # Built-in first — built-in wins on CheckID collision
+    if is_builtin_provider(provider_type):
+        builtin_path = f"prowler.providers.{provider_type}.services.{service}.{check_name}.{check_name}"
+        if importlib.util.find_spec(builtin_path) is not None:
+            return import_check(builtin_path)
+
+    # Entry point lookup — only consulted when the built-in truly doesn't exist
+    for ep in importlib.metadata.entry_points(group=f"prowler.checks.{provider_type}"):
+        if ep.name == check_name:
+            return importlib.import_module(ep.value)
+
+    raise ModuleNotFoundError(
+        f"Check '{check_name}' not found for provider '{provider_type}'"
+    )
+
+
 def run_fixer(check_findings: list) -> int:
     """
     Run the fixer for the check if it exists and there are any FAIL findings
@@ -525,9 +567,10 @@ def execute_checks(
             service = check_name.split("_")[0]
             try:
                 try:
-                    # Import check module
-                    check_module_path = f"prowler.providers.{global_provider.type}.services.{service}.{check_name}.{check_name}"
-                    lib = import_check(check_module_path)
+                    # Import check module (built-in or entry point)
+                    lib = _resolve_check_module(
+                        global_provider.type, service, check_name
+                    )
                     # Recover functions from check
                     check_to_execute = getattr(lib, check_name)
                     check = check_to_execute()
@@ -605,9 +648,10 @@ def execute_checks(
                 )
                 try:
                     try:
-                        # Import check module
-                        check_module_path = f"prowler.providers.{global_provider.type}.services.{service}.{check_name}.{check_name}"
-                        lib = import_check(check_module_path)
+                        # Import check module (built-in or entry point)
+                        lib = _resolve_check_module(
+                            global_provider.type, service, check_name
+                        )
                         # Recover functions from check
                         check_to_execute = getattr(lib, check_name)
                         check = check_to_execute()
@@ -753,6 +797,10 @@ def execute(
                 is_finding_muted_args["org_domain"] = (
                     global_provider.identity.org_domain
                 )
+            elif not is_builtin_provider(global_provider.type):
+                # External/custom provider — delegate identity args
+                is_finding_muted_args = global_provider.get_mutelist_finding_args()
+
             for finding in check_findings:
                 if global_provider.type == "cloudflare":
                     is_finding_muted_args["account_id"] = finding.account_id

prowler/lib/check/checks_loader.py

@@ -2,10 +2,10 @@ import sys
 
 from colorama import Fore, Style
 
-from prowler.config.config import EXTERNAL_TOOL_PROVIDERS
 from prowler.lib.check.check import parse_checks_from_file
 from prowler.lib.check.compliance_models import Compliance
 from prowler.lib.check.models import CheckMetadata, Severity
+from prowler.lib.check.tool_wrapper import is_tool_wrapper_provider
 from prowler.lib.logger import logger
 
 
@@ -26,8 +26,13 @@ def load_checks_to_execute(
 ) -> set:
     """Generate the list of checks to execute based on the cloud provider and the input arguments given"""
     try:
-        # Bypass check loading for providers that use external tools directly
-        if provider in EXTERNAL_TOOL_PROVIDERS:
+        # Bypass check loading for tool-wrapper providers — they delegate
+        # scanning to an external tool and have no checks to recover.
+        # Single source of truth across __main__, the CheckMetadata validators,
+        # check discovery and this loader, covering both built-in tool wrappers
+        # (iac/llm/image) and external plug-ins that declare
+        # `is_external_tool_provider = True` via the contract.
+        if is_tool_wrapper_provider(provider):
             return set()
 
         # Local subsets

prowler/lib/check/compliance_models.py

@@ -1,3 +1,4 @@
+import importlib.metadata
 import json
 import os
 import sys
@@ -434,26 +435,63 @@ class Compliance(BaseModel):
         """Bulk load all compliance frameworks specification into a dict"""
         try:
             bulk_compliance_frameworks = {}
+            # Built-in compliance from prowler/compliance/{provider}/
             available_compliance_framework_modules = list_compliance_modules()
             for compliance_framework in available_compliance_framework_modules:
-                if provider in compliance_framework.name:
+                # Match the provider segment exactly, not as a substring, so
+                # e.g. `cloud` does not capture `cloudflare`.
+                if compliance_framework.name.split(".")[-1] == provider:
                     compliance_specification_dir_path = (
                         f"{compliance_framework.module_finder.path}/{provider}"
                     )
-                    # for compliance_framework in available_compliance_framework_modules:
                     for filename in os.listdir(compliance_specification_dir_path):
                         file_path = os.path.join(
                             compliance_specification_dir_path, filename
                         )
-                        # Check if it is a file and ti size is greater than 0
                         if os.path.isfile(file_path) and os.stat(file_path).st_size > 0:
-                            # Open Compliance file in JSON
-                            # cis_v1.4_aws.json --> cis_v1.4_aws
                             compliance_framework_name = filename.split(".json")[0]
-                            # Store the compliance info
                             bulk_compliance_frameworks[compliance_framework_name] = (
                                 load_compliance_framework(file_path)
                             )
+
+            # External compliance via entry points
+            for ep in importlib.metadata.entry_points(group="prowler.compliance"):
+                if ep.name == provider:
+                    try:
+                        module = ep.load()
+                        compliance_dir = (
+                            module.__path__[0]
+                            if hasattr(module, "__path__")
+                            else os.path.dirname(module.__file__)
+                        )
+                        for filename in os.listdir(compliance_dir):
+                            if filename.endswith(".json"):
+                                file_path = os.path.join(compliance_dir, filename)
+                                if (
+                                    os.path.isfile(file_path)
+                                    and os.stat(file_path).st_size > 0
+                                ):
+                                    compliance_framework_name = filename.split(".json")[
+                                        0
+                                    ]
+                                    if (
+                                        compliance_framework_name
+                                        not in bulk_compliance_frameworks
+                                    ):
+                                        # External JSON: tolerate non-legacy
+                                        # schemas (skip + warn) instead of aborting.
+                                        framework = load_compliance_framework(
+                                            file_path, fatal=False
+                                        )
+                                        if framework is not None:
+                                            bulk_compliance_frameworks[
+                                                compliance_framework_name
+                                            ] = framework
+                    except Exception as error:
+                        logger.warning(
+                            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                        )
+
         except Exception as e:
             logger.error(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}] -- {e}")
 
@@ -462,18 +500,26 @@ class Compliance(BaseModel):
 
 # Testing Pending
 def load_compliance_framework(
-    compliance_specification_file: str,
-) -> Compliance:
-    """load_compliance_framework loads and parse a Compliance Framework Specification"""
+    compliance_specification_file: str, fatal: bool = True
+) -> Optional[Compliance]:
+    """load_compliance_framework loads and parse a Compliance Framework Specification.
+
+    With ``fatal=True`` (built-in JSONs) an invalid file aborts the run; with
+    ``fatal=False`` (external JSONs) it is skipped with a warning and ``None``
+    is returned.
+    """
     try:
-        compliance_framework = Compliance.parse_file(compliance_specification_file)
+        return Compliance.parse_file(compliance_specification_file)
     except ValidationError as error:
-        logger.critical(
-            f"Compliance Framework Specification from {compliance_specification_file} is not valid: {error}"
+        if fatal:
+            logger.critical(
+                f"Compliance Framework Specification from {compliance_specification_file} is not valid: {error}"
+            )
+            sys.exit(1)
+        logger.warning(
+            f"Skipping invalid compliance framework {compliance_specification_file}: {error}"
         )
-        sys.exit(1)
-    else:
-        return compliance_framework
+        return None
 
 
 # ─── Universal Compliance Schema Models (Phase 1-3) ─────────────────────────
@@ -950,6 +996,25 @@ def get_bulk_compliance_frameworks_universal(provider: str) -> dict:
         if compliance_root and os.path.isdir(compliance_root):
             _load_jsons_from_dir(compliance_root, provider, bulk)
 
+        # External multi-provider frameworks via the dedicated universal entry
+        # point group, kept separate from the per-provider `prowler.compliance`
+        # group so the legacy loader never parses a universal JSON. Built-ins
+        # (already in bulk) win on a name collision.
+        for ep in importlib.metadata.entry_points(group="prowler.compliance.universal"):
+            try:
+                module = ep.load()
+                ep_dir = (
+                    module.__path__[0]
+                    if hasattr(module, "__path__")
+                    else os.path.dirname(module.__file__)
+                )
+                if os.path.isdir(ep_dir):
+                    _load_jsons_from_dir(ep_dir, provider, bulk)
+            except Exception as error:
+                logger.warning(
+                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+
     except Exception as e:
         logger.error(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}] -- {e}")
     return bulk

prowler/lib/check/models.py

@@ -11,10 +11,10 @@ from typing import Any, Dict, Optional, Set
 from pydantic.v1 import BaseModel, Field, ValidationError, validator
 from pydantic.v1.error_wrappers import ErrorWrapper
 
-from prowler.config.config import EXTERNAL_TOOL_PROVIDERS, Provider
 from prowler.lib.check.compliance_models import Compliance
 from prowler.lib.check.utils import recover_checks_from_provider
 from prowler.lib.logger import logger
+from prowler.providers.common.provider import Provider as ProviderABC
 
 # Valid ResourceGroup values as defined in the RFC
 VALID_RESOURCE_GROUPS = frozenset(
@@ -259,7 +259,7 @@ class CheckMetadata(BaseModel):
             )
         if (
             value_lower not in VALID_CATEGORIES
-            and values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS
+            and not ProviderABC.is_tool_wrapper_provider(values.get("Provider"))
         ):
             raise ValueError(
                 f"Invalid category: '{value_lower}'. Must be one of: {', '.join(sorted(VALID_CATEGORIES))}."
@@ -288,7 +288,9 @@ class CheckMetadata(BaseModel):
             raise ValueError("ServiceName must be a non-empty string")
 
         check_id = values.get("CheckID")
-        if check_id and values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
+        if check_id and not ProviderABC.is_tool_wrapper_provider(
+            values.get("Provider")
+        ):
             service_from_check_id = check_id.split("_")[0]
             if service_name != service_from_check_id:
                 raise ValueError(
@@ -304,7 +306,9 @@ class CheckMetadata(BaseModel):
         if not check_id:
             raise ValueError("CheckID must be a non-empty string")
 
-        if check_id and values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
+        if check_id and not ProviderABC.is_tool_wrapper_provider(
+            values.get("Provider")
+        ):
             if "-" in check_id:
                 raise ValueError(
                     f"CheckID {check_id} contains a hyphen, which is not allowed"
@@ -313,8 +317,9 @@ class CheckMetadata(BaseModel):
         return check_id
 
     @validator("CheckTitle", pre=True, always=True)
+    @classmethod
     def validate_check_title(cls, check_title, values):  # noqa: F841
-        if values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
+        if not ProviderABC.is_tool_wrapper_provider(values.get("Provider")):
             if len(check_title) > 150:
                 raise ValueError(
                     f"CheckTitle must not exceed 150 characters, got {len(check_title)} characters"
@@ -326,14 +331,18 @@ class CheckMetadata(BaseModel):
         return check_title
 
     @validator("RelatedUrl", pre=True, always=True)
+    @classmethod
     def validate_related_url(cls, related_url, values):  # noqa: F841
-        if related_url and values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
+        if related_url and not ProviderABC.is_tool_wrapper_provider(
+            values.get("Provider")
+        ):
             raise ValueError("RelatedUrl must be empty. This field is deprecated.")
         return related_url
 
     @validator("Remediation")
+    @classmethod
     def validate_recommendation_url(cls, remediation, values):  # noqa: F841
-        if values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
+        if not ProviderABC.is_tool_wrapper_provider(values.get("Provider")):
             url = remediation.Recommendation.Url
             if url and not url.startswith("https://hub.prowler.com/"):
                 raise ValueError(
@@ -346,7 +355,7 @@ class CheckMetadata(BaseModel):
         provider = values.get("Provider", "").lower()
 
         # Non-AWS providers must have an empty CheckType list
-        if provider != "aws" and provider not in EXTERNAL_TOOL_PROVIDERS:
+        if provider != "aws" and not ProviderABC.is_tool_wrapper_provider(provider):
             if check_type:
                 raise ValueError(
                     f"CheckType must be empty for non-AWS providers. Got {check_type} for provider '{provider}'."
@@ -371,8 +380,9 @@ class CheckMetadata(BaseModel):
         return check_type
 
     @validator("Description", pre=True, always=True)
+    @classmethod
     def validate_description(cls, description, values):  # noqa: F841
-        if values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
+        if not ProviderABC.is_tool_wrapper_provider(values.get("Provider")):
             if len(description) > 400:
                 raise ValueError(
                     f"Description must not exceed 400 characters, got {len(description)} characters"
@@ -380,8 +390,9 @@ class CheckMetadata(BaseModel):
         return description
 
     @validator("Risk", pre=True, always=True)
+    @classmethod
     def validate_risk(cls, risk, values):  # noqa: F841
-        if values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
+        if not ProviderABC.is_tool_wrapper_provider(values.get("Provider")):
             if len(risk) > 400:
                 raise ValueError(
                     f"Risk must not exceed 400 characters, got {len(risk)} characters"
@@ -433,6 +444,20 @@ class CheckMetadata(BaseModel):
             metadata_file = f"{check_path}/{check_name}.metadata.json"
             # Load metadata
             check_metadata = load_check_metadata(metadata_file)
+            # Built-in wins on CheckID collision. Plug-in entry points are
+            # appended after built-ins by `recover_checks_from_provider`, so
+            # a duplicate CheckID here means an entry-point check is trying
+            # to override a built-in. Ignore the override (the built-in
+            # metadata stays) and surface it via a warning — matching the
+            # precedence enforced by `_resolve_check_module`.
+            if check_metadata.CheckID in bulk_check_metadata:
+                logger.warning(
+                    f"Plug-in check metadata '{check_metadata.CheckID}' "
+                    f"(loaded from '{metadata_file}') is being IGNORED — "
+                    f"a built-in with the same CheckID exists. To use your "
+                    f"plug-in, register it under a different CheckID."
+                )
+                continue
             bulk_check_metadata[check_metadata.CheckID] = check_metadata
 
         return bulk_check_metadata
@@ -470,7 +495,7 @@ class CheckMetadata(BaseModel):
         # If the bulk checks metadata is not provided, get it
         if not bulk_checks_metadata:
             bulk_checks_metadata = {}
-            available_providers = [p.value for p in Provider]
+            available_providers = ProviderABC.get_available_providers()
             for provider_name in available_providers:
                 bulk_checks_metadata.update(CheckMetadata.get_bulk(provider_name))
         if provider:
@@ -495,7 +520,7 @@ class CheckMetadata(BaseModel):
             # Loaded here, as it is not always needed
             if not bulk_compliance_frameworks:
                 bulk_compliance_frameworks = {}
-                available_providers = [p.value for p in Provider]
+                available_providers = ProviderABC.get_available_providers()
                 for provider in available_providers:
                     bulk_compliance_frameworks = Compliance.get_bulk(provider=provider)
             checks_from_compliance_framework = (

prowler/lib/check/tool_wrapper.py

@@ -0,0 +1,62 @@
+"""Standalone helper for tool-wrapper provider detection.
+
+A provider is a "tool wrapper" if it delegates scanning to an external tool
+(Trivy, promptfoo, etc.) instead of running checks/services through the
+standard Prowler engine. This module is the single source of truth for that
+classification across the codebase.
+
+Kept as a leaf module with no Prowler imports beyond the leaf
+`external_tool_providers` so it can be referenced from `prowler.lib.check.*`
+and `prowler.providers.common.provider` without forming an import cycle.
+"""
+
+import importlib.metadata
+
+from prowler.lib.check.external_tool_providers import EXTERNAL_TOOL_PROVIDERS
+from prowler.providers.common.builtin import is_builtin_provider
+
+# Module-level cache for entry-point classes consulted by this helper.
+# Independent of `Provider._ep_providers` to keep this module leaf — the cost
+# of a duplicate cache entry is negligible (one class object per external
+# provider, loaded lazily on first lookup).
+_ep_class_cache: dict = {}
+
+
+def _load_ep_class(provider: str):
+    """Return the entry-point provider class for `provider`, or None.
+
+    Caches the result in `_ep_class_cache`. Errors during entry-point loading
+    are swallowed (returning None) so a broken plug-in never crashes the
+    is-tool-wrapper check; it just falls through to "not a tool wrapper".
+    """
+    if provider in _ep_class_cache:
+        return _ep_class_cache[provider]
+    for ep in importlib.metadata.entry_points(group="prowler.providers"):
+        if ep.name == provider:
+            try:
+                cls = ep.load()
+            except Exception:
+                cls = None
+            _ep_class_cache[provider] = cls
+            return cls
+    _ep_class_cache[provider] = None
+    return None
+
+
+def is_tool_wrapper_provider(provider: str) -> bool:
+    """Return True if the provider delegates scanning to an external tool.
+
+    Combines the built-in `EXTERNAL_TOOL_PROVIDERS` frozenset (fast path for
+    iac/llm/image) with the `is_external_tool_provider` class attribute of
+    external plug-ins registered via entry points. This is the single source
+    of truth consulted by `__main__`, the `CheckMetadata` validators, the
+    check-loading utilities, and the checks loader.
+    """
+    if provider in EXTERNAL_TOOL_PROVIDERS:
+        return True
+    # Built-in wins: short-circuit before ep.load() so a same-name plug-in
+    # cannot flip a built-in onto the tool-wrapper path or run its code.
+    if is_builtin_provider(provider):
+        return False
+    cls = _load_ep_class(provider)
+    return bool(cls and getattr(cls, "is_external_tool_provider", False))

prowler/lib/check/utils.py

@@ -1,9 +1,43 @@
 import importlib
+import importlib.metadata
+import importlib.util
+import os
 import sys
 from pkgutil import walk_packages
 
-from prowler.lib.check.external_tool_providers import EXTERNAL_TOOL_PROVIDERS
+from prowler.lib.check.tool_wrapper import is_tool_wrapper_provider
 from prowler.lib.logger import logger
+from prowler.providers.common.builtin import is_builtin_provider
+
+
+def _recover_ep_checks(provider: str, service: str = None) -> list[tuple]:
+    """Discover external checks registered via entry points for a provider.
+
+    External plugins follow the same layout as built-ins:
+    `{plugin_root}.services.{service}.{check}.{check}`
+
+    When `service` is provided, only entry points whose dotted path contains
+    `.services.{service}.` are included — mirroring how built-in discovery
+    filters by the `prowler.providers.{provider}.services.{service}` package.
+
+    Uses find_spec to locate the check module without importing it,
+    avoiding service client initialization at discovery time.
+    """
+    checks = []
+    for ep in importlib.metadata.entry_points(group=f"prowler.checks.{provider}"):
+        try:
+            if service and f".services.{service}." not in ep.value:
+                continue
+
+            spec = importlib.util.find_spec(ep.value)
+            if spec and spec.origin:
+                check_path = os.path.dirname(spec.origin)
+                checks.append((ep.name, check_path))
+        except Exception as error:
+            logger.warning(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+    return checks
 
 
 def recover_checks_from_provider(
@@ -15,29 +49,55 @@ def recover_checks_from_provider(
     Returns a list of tuples with the following format (check_name, check_path)
     """
     try:
-        # Bypass check loading for providers that use external tools directly
-        if provider in EXTERNAL_TOOL_PROVIDERS:
+        # Bypass check loading for tool-wrapper providers — they delegate
+        # scanning to an external tool and have no checks to recover.
+        # Single source of truth: combines the EXTERNAL_TOOL_PROVIDERS
+        # frozenset (built-ins) with the per-provider `is_external_tool_provider`
+        # class attribute (so external plug-ins opt in via the contract).
+        if is_tool_wrapper_provider(provider):
             return []
 
         checks = []
-        modules = list_modules(provider, service)
-        for module_name in modules:
-            # Format: "prowler.providers.{provider}.services.{service}.{check_name}.{check_name}"
-            check_module_name = module_name.name
-            # We need to exclude common shared libraries in services
-            if (
-                check_module_name.count(".") == 6
-                and ".lib." not in check_module_name
-                and (not check_module_name.endswith("_fixer") or include_fixers)
-            ):
-                check_path = module_name.module_finder.path
-                # Check name is the last part of the check_module_name
-                check_name = check_module_name.split(".")[-1]
-                check_info = (check_name, check_path)
-                checks.append(check_info)
-    except ModuleNotFoundError:
-        logger.critical(f"Service {service} was not found for the {provider} provider.")
-        sys.exit(1)
+        # Built-in checks from prowler.providers.{provider}.services. Gate
+        # the built-in branch on `is_builtin_provider(provider)` — calling
+        # `find_spec` directly on `prowler.providers.{provider}.services`
+        # would propagate `ModuleNotFoundError` when the parent package
+        # `prowler.providers.{provider}` does not exist (i.e. the provider
+        # is external), instead of returning None. The leaf helper
+        # encapsulates the safe lookup, so we only run the built-in
+        # discovery when the provider actually ships with the SDK; for
+        # external providers we go straight to entry points.
+        if is_builtin_provider(provider):
+            modules = list_modules(provider, service)
+            for module_name in modules:
+                # Format: "prowler.providers.{provider}.services.{service}.{check_name}.{check_name}"
+                check_module_name = module_name.name
+                # We need to exclude common shared libraries in services
+                if (
+                    check_module_name.count(".") == 6
+                    and ".lib." not in check_module_name
+                    and (not check_module_name.endswith("_fixer") or include_fixers)
+                ):
+                    check_path = module_name.module_finder.path
+                    check_name = check_module_name.split(".")[-1]
+                    check_info = (check_name, check_path)
+                    checks.append(check_info)
+
+        # External checks registered via entry points — always consulted, with
+        # optional service filter. Previously gated by `if not service:`, which
+        # prevented external providers from being usable with --service.
+        checks.extend(_recover_ep_checks(provider, service))
+
+        # A service was requested but nothing matched in either built-ins or
+        # entry points — surface this as a clear error instead of silently
+        # returning an empty list.
+        if service and not checks:
+            logger.critical(
+                f"Service '{service}' was not found for the '{provider}' provider "
+                f"(neither as a built-in nor via external entry points)."
+            )
+            sys.exit(1)
+
     except Exception as e:
         logger.critical(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}")
         sys.exit(1)
@@ -64,8 +124,9 @@ def recover_checks_from_service(service_list: list, provider: str) -> set:
     Returns a set of checks from the given services
     """
     try:
-        # Bypass check loading for providers that use external tools directly
-        if provider in EXTERNAL_TOOL_PROVIDERS:
+        # Bypass check loading for tool-wrapper providers — symmetric with
+        # `recover_checks_from_provider` above, using the same source of truth.
+        if is_tool_wrapper_provider(provider):
             return set()
 
         checks = set()

prowler/lib/cli/parser.py

@@ -20,19 +20,61 @@ from prowler.providers.common.arguments import (
     validate_provider_arguments,
     validate_sarif_usage,
 )
+from prowler.providers.common.provider import Provider
 
 
 class ProwlerArgumentParser:
     # Set the default parser
     def __init__(self):
+        # Discover any providers not in the hardcoded list below
+        # TODO - First step to support current providers and the new external provider implementation
+        known_providers = {
+            "aws",
+            "azure",
+            "gcp",
+            "kubernetes",
+            "m365",
+            "github",
+            "googleworkspace",
+            "cloudflare",
+            "oraclecloud",
+            "openstack",
+            "alibabacloud",
+            "iac",
+            "llm",
+            "image",
+            "nhn",
+            "mongodbatlas",
+            "vercel",
+            "okta",
+            "scaleway",
+            "stackit",
+        }
+        all_providers = set(Provider.get_available_providers())
+        new_providers = sorted(all_providers - known_providers)
+
+        # Build extra strings for dynamically discovered providers
+        extra_providers_csv = ""
+        extra_providers_text = ""
+        if new_providers:
+            providers_help = Provider.get_providers_help_text()
+            extra_providers_csv = "," + ",".join(new_providers)
+            extra_lines = []
+            for name in new_providers:
+                help_text = providers_help.get(name, "")
+                if help_text:
+                    extra_lines.append(f"    {name:<20}{help_text}")
+            if extra_lines:
+                extra_providers_text = "\n" + "\n".join(extra_lines)
+
         # CLI Arguments
         self.parser = argparse.ArgumentParser(
             prog="prowler",
             formatter_class=RawTextHelpFormatter,
-            usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,googleworkspace,okta,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,scaleway,stackit,vercel,dashboard,iac,image,llm} ...",
-            epilog="""
+            usage=f"prowler [-h] [--version] {{aws,azure,gcp,kubernetes,m365,github,googleworkspace,okta,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,scaleway,stackit,vercel,dashboard,iac,image,llm{extra_providers_csv}}} ...",
+            epilog=f"""
 Available Cloud Providers:
-  {aws,azure,gcp,kubernetes,m365,github,googleworkspace,okta,iac,llm,image,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,scaleway,stackit,vercel}
+  {{aws,azure,gcp,kubernetes,m365,github,googleworkspace,okta,iac,llm,image,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,scaleway,stackit,vercel{extra_providers_csv}}}
     aws                 AWS Provider
     azure               Azure Provider
     gcp                 GCP Provider
@@ -52,13 +94,14 @@ Available Cloud Providers:
     nhn                 NHN Provider (Unofficial)
     mongodbatlas        MongoDB Atlas Provider
     scaleway            Scaleway Provider
-    vercel              Vercel Provider
+    vercel              Vercel Provider{extra_providers_text}
+
 
 Available components:
     dashboard           Local dashboard
 
 To see the different available options on a specific component, run:
-    prowler {provider|dashboard} -h|--help
+    prowler {{provider|dashboard}} -h|--help
 
 Detailed documentation at https://docs.prowler.com
 """,
@@ -117,8 +160,10 @@ Detailed documentation at https://docs.prowler.com
             and (sys.argv[1] not in ("-v", "--version"))
         ):
             # Since the provider is always the second argument, we are checking if
-            # a flag, starting by "-", is supplied
-            if "-" in sys.argv[1]:
+            # a flag is supplied. Use startswith("-") instead of "in" to avoid
+            # matching external provider names that contain hyphens
+            # (e.g. "local-acme-snowflake").
+            if sys.argv[1].startswith("-"):
                 sys.argv = self.__set_default_provider__(sys.argv)
 
             # Provider aliases mapping

prowler/lib/outputs/compliance/compliance.py

@@ -253,14 +253,32 @@ def display_compliance_table(
                 compliance_overview,
             )
         else:
-            get_generic_compliance_table(
-                findings,
-                bulk_checks_metadata,
-                compliance_framework,
-                output_filename,
-                output_directory,
-                compliance_overview,
-            )
+            # Try provider-specific table first, fall back to generic
+            from prowler.providers.common.provider import Provider
+
+            provider = Provider.get_global_provider()
+            handled = False
+            if provider is not None:
+                try:
+                    handled = provider.display_compliance_table(
+                        findings,
+                        bulk_checks_metadata,
+                        compliance_framework,
+                        output_filename,
+                        output_directory,
+                        compliance_overview,
+                    )
+                except NotImplementedError:
+                    handled = False
+            if not handled:
+                get_generic_compliance_table(
+                    findings,
+                    bulk_checks_metadata,
+                    compliance_framework,
+                    output_filename,
+                    output_directory,
+                    compliance_overview,
+                )
     except Exception as error:
         logger.critical(
             f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"

prowler/lib/outputs/compliance/generic/generic.py

@@ -34,60 +34,48 @@ class GenericCompliance(ComplianceOutput):
         Returns:
             - None
         """
+
+        def compliance_row(requirement, attribute, finding=None):
+            # Read attribute fields defensively: GenericCompliance is the
+            # last-resort renderer for any framework, and provider-specific
+            # schemas (e.g. CIS, ENS, ISO27001) do not declare the universal
+            # Section/SubSection/SubGroup/Service/Type/Comment fields.
+            return GenericComplianceModel(
+                Provider=(finding.provider if finding else compliance.Provider.lower()),
+                Description=compliance.Description,
+                AccountId=finding.account_uid if finding else "",
+                Region=finding.region if finding else "",
+                AssessmentDate=str(timestamp),
+                Requirements_Id=requirement.Id,
+                Requirements_Description=requirement.Description,
+                Requirements_Attributes_Section=getattr(attribute, "Section", None),
+                Requirements_Attributes_SubSection=getattr(
+                    attribute, "SubSection", None
+                ),
+                Requirements_Attributes_SubGroup=getattr(attribute, "SubGroup", None),
+                Requirements_Attributes_Service=getattr(attribute, "Service", None),
+                Requirements_Attributes_Type=getattr(attribute, "Type", None),
+                Requirements_Attributes_Comment=getattr(attribute, "Comment", None),
+                Status=finding.status if finding else "MANUAL",
+                StatusExtended=(finding.status_extended if finding else "Manual check"),
+                ResourceId=finding.resource_uid if finding else "manual_check",
+                ResourceName=finding.resource_name if finding else "Manual check",
+                CheckId=finding.check_id if finding else "manual",
+                Muted=finding.muted if finding else False,
+                Framework=compliance.Framework,
+                Name=compliance.Name,
+            )
+
         for finding in findings:
             for requirement in compliance.Requirements:
                 # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
                 if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
-                        compliance_row = GenericComplianceModel(
-                            Provider=finding.provider,
-                            Description=compliance.Description,
-                            AccountId=finding.account_uid,
-                            Region=finding.region,
-                            AssessmentDate=str(timestamp),
-                            Requirements_Id=requirement.Id,
-                            Requirements_Description=requirement.Description,
-                            Requirements_Attributes_Section=attribute.Section,
-                            Requirements_Attributes_SubSection=attribute.SubSection,
-                            Requirements_Attributes_SubGroup=attribute.SubGroup,
-                            Requirements_Attributes_Service=attribute.Service,
-                            Requirements_Attributes_Type=attribute.Type,
-                            Requirements_Attributes_Comment=attribute.Comment,
-                            Status=finding.status,
-                            StatusExtended=finding.status_extended,
-                            ResourceId=finding.resource_uid,
-                            ResourceName=finding.resource_name,
-                            CheckId=finding.check_id,
-                            Muted=finding.muted,
-                            Framework=compliance.Framework,
-                            Name=compliance.Name,
+                        self._data.append(
+                            compliance_row(requirement, attribute, finding)
                         )
-                        self._data.append(compliance_row)
         # Add manual requirements to the compliance output
         for requirement in compliance.Requirements:
             if not requirement.Checks:
                 for attribute in requirement.Attributes:
-                    compliance_row = GenericComplianceModel(
-                        Provider=compliance.Provider.lower(),
-                        Description=compliance.Description,
-                        AccountId="",
-                        Region="",
-                        AssessmentDate=str(timestamp),
-                        Requirements_Id=requirement.Id,
-                        Requirements_Description=requirement.Description,
-                        Requirements_Attributes_Section=attribute.Section,
-                        Requirements_Attributes_SubSection=attribute.SubSection,
-                        Requirements_Attributes_SubGroup=attribute.SubGroup,
-                        Requirements_Attributes_Service=attribute.Service,
-                        Requirements_Attributes_Type=attribute.Type,
-                        Requirements_Attributes_Comment=attribute.Comment,
-                        Status="MANUAL",
-                        StatusExtended="Manual check",
-                        ResourceId="manual_check",
-                        ResourceName="Manual check",
-                        CheckId="manual",
-                        Muted=False,
-                        Framework=compliance.Framework,
-                        Name=compliance.Name,
-                    )
-                    self._data.append(compliance_row)
+                    self._data.append(compliance_row(requirement, attribute))

prowler/lib/outputs/finding.py

@@ -517,6 +517,11 @@ class Finding(BaseModel):
                     check_output, "fixed_version", ""
                 )
 
+            else:
+                # Dynamic fallback: any external/custom provider
+                provider_data = provider.get_finding_output_data(check_output)
+                output_data.update(provider_data)
+
             # check_output Unique ID
             # TODO: move this to a function
             # TODO: in Azure, GCP and K8s there are findings without resource_name

prowler/lib/outputs/html/html.py

@@ -1608,11 +1608,13 @@ class HTML(Output):
             # Azure_provider --> azure
             # Kubernetes_provider --> kubernetes
 
-            # Dynamically get the Provider quick inventory handler
-            provider_html_assessment_summary_function = (
-                f"get_{provider.type}_assessment_summary"
-            )
-            return getattr(HTML, provider_html_assessment_summary_function)(provider)
+            # Try static method first, fall back to provider method
+            method_name = f"get_{provider.type}_assessment_summary"
+            if hasattr(HTML, method_name):
+                return getattr(HTML, method_name)(provider)
+            else:
+                # Dynamic fallback: any external/custom provider
+                return provider.get_html_assessment_summary()
         except Exception as error:
             logger.error(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"

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/lib/outputs/outputs.py

@@ -7,45 +7,52 @@ from prowler.lib.outputs.common import Status
 from prowler.lib.outputs.finding import Finding
 
 
-def stdout_report(finding, color, verbose, status, fix):
+def stdout_report(finding, color, verbose, status, fix, provider=None):
     if finding.check_metadata.Provider == "aws":
         details = finding.region
-    if finding.check_metadata.Provider == "azure":
+    elif finding.check_metadata.Provider == "azure":
         details = finding.location
-    if finding.check_metadata.Provider == "gcp":
+    elif finding.check_metadata.Provider == "gcp":
         details = finding.location.lower()
-    if finding.check_metadata.Provider == "kubernetes":
+    elif finding.check_metadata.Provider == "kubernetes":
         details = finding.namespace.lower()
-    if finding.check_metadata.Provider == "github":
+    elif finding.check_metadata.Provider == "github":
         details = finding.owner
-    if finding.check_metadata.Provider == "m365":
+    elif finding.check_metadata.Provider == "m365":
         details = finding.location
-    if finding.check_metadata.Provider == "mongodbatlas":
+    elif finding.check_metadata.Provider == "mongodbatlas":
         details = finding.location
-    if finding.check_metadata.Provider == "nhn":
+    elif finding.check_metadata.Provider == "nhn":
         details = finding.location
-    if finding.check_metadata.Provider == "stackit":
+    elif finding.check_metadata.Provider == "stackit":
         details = finding.location
-    if finding.check_metadata.Provider == "llm":
+    elif finding.check_metadata.Provider == "llm":
         details = finding.check_metadata.CheckID
-    if finding.check_metadata.Provider == "iac":
+    elif finding.check_metadata.Provider == "iac":
         details = finding.check_metadata.CheckID
-    if finding.check_metadata.Provider == "oraclecloud":
+    elif finding.check_metadata.Provider == "oraclecloud":
         details = finding.region
-    if finding.check_metadata.Provider == "alibabacloud":
+    elif finding.check_metadata.Provider == "alibabacloud":
         details = finding.region
-    if finding.check_metadata.Provider == "openstack":
+    elif finding.check_metadata.Provider == "openstack":
         details = finding.region
-    if finding.check_metadata.Provider == "cloudflare":
+    elif finding.check_metadata.Provider == "cloudflare":
         details = finding.zone_name
-    if finding.check_metadata.Provider == "googleworkspace":
+    elif finding.check_metadata.Provider == "googleworkspace":
         details = finding.location
-    if finding.check_metadata.Provider == "vercel":
+    elif finding.check_metadata.Provider == "vercel":
         details = finding.region
-    if finding.check_metadata.Provider == "okta":
+    elif finding.check_metadata.Provider == "okta":
         details = finding.region
-    if finding.check_metadata.Provider == "scaleway":
+    elif finding.check_metadata.Provider == "scaleway":
         details = finding.region
+    else:
+        # Dynamic fallback: any external/custom provider
+        if provider is None:
+            from prowler.providers.common.provider import Provider
+
+            provider = Provider.get_global_provider()
+        details = provider.get_stdout_detail(finding)
 
     if (verbose or fix) and (not status or finding.status in status):
         if finding.muted:
@@ -65,12 +72,15 @@ def report(check_findings, provider, output_options):
         if hasattr(output_options, "verbose"):
             verbose = output_options.verbose
         if check_findings:
-            # TO-DO Generic Function
             if provider.type == "aws":
                 check_findings.sort(key=lambda x: x.region)
-
-            if provider.type == "azure":
+            elif provider.type == "azure":
                 check_findings.sort(key=lambda x: x.subscription)
+            else:
+                # Dynamic fallback: any external/custom provider
+                sort_key = provider.get_finding_sort_key()
+                if sort_key and isinstance(sort_key, str):
+                    check_findings.sort(key=lambda x: getattr(x, sort_key, ""))
 
             for finding in check_findings:
                 # Print findings by stdout
@@ -81,12 +91,16 @@ def report(check_findings, provider, output_options):
                 if hasattr(output_options, "fixer"):
                     fixer = output_options.fixer
                 color = set_report_color(finding.status, finding.muted)
+                # Pass the local `provider` through so the dynamic else inside
+                # `stdout_report` does not have to consult the global singleton
+                # — defeating the whole purpose of the new parameter.
                 stdout_report(
                     finding,
                     color,
                     verbose,
                     status,
                     fixer,
+                    provider=provider,
                 )
 
         else:  # No service resources in the whole account

prowler/lib/outputs/summary_table.py

@@ -121,6 +121,9 @@ def display_summary_table(
         elif provider.type == "scaleway":
             entity_type = "Organization"
             audited_entities = provider.identity.organization_id
+        else:
+            # Dynamic fallback: any external/custom provider
+            entity_type, audited_entities = provider.get_summary_entity()
 
         # Check if there are findings and that they are not all MANUAL
         if findings and not all(finding.status == "MANUAL" for finding in findings):

prowler/lib/scan/scan.py

@@ -4,8 +4,8 @@ from types import SimpleNamespace
 from typing import Generator
 
 from prowler.lib.check.check import (
+    _resolve_check_module,
     execute,
-    import_check,
     list_services,
     update_audit_metadata,
 )
@@ -426,9 +426,14 @@ class Scan:
                     # Recover service from check name
                     service = get_service_name_from_check_name(check_name)
                     try:
-                        # Import check module
-                        check_module_path = f"prowler.providers.{self._provider.type}.services.{service}.{check_name}.{check_name}"
-                        lib = import_check(check_module_path)
+                        # Import check module (built-in or entry point) —
+                        # delegates to `_resolve_check_module` so external
+                        # providers registered via entry points are resolved
+                        # correctly (their checks do not live under
+                        # `prowler.providers.{type}.services...`).
+                        lib = _resolve_check_module(
+                            self._provider.type, service, check_name
+                        )
                         # Recover functions from check
                         check_to_execute = getattr(lib, check_name)
                         check = check_to_execute()

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/common/arguments.py

@@ -16,18 +16,41 @@ def init_providers_parser(self):
     # We need to call the arguments parser for each provider
     providers = Provider.get_available_providers()
     for provider in providers:
-        try:
-            getattr(
-                import_module(
-                    f"{providers_path}.{provider}.{provider_arguments_lib_path}"
-                ),
-                init_provider_arguments_function,
-            )(self)
-        except Exception as error:
-            logger.critical(
-                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-            )
-            sys.exit(1)
+        # Discriminate built-in vs external upfront via find_spec, so an
+        # ImportError from a transitive dependency missing inside a built-in
+        # arguments module surfaces clearly instead of being silently
+        # re-routed to the entry-point path (which only has external providers).
+        if Provider.is_builtin(provider):
+            try:
+                getattr(
+                    import_module(
+                        f"{providers_path}.{provider}.{provider_arguments_lib_path}"
+                    ),
+                    init_provider_arguments_function,
+                )(self)
+            except ImportError as e:
+                logger.critical(
+                    f"Failed to load arguments for built-in provider '{provider}'. "
+                    f"Missing dependency: {e}. "
+                    f"Ensure all required dependencies are installed."
+                )
+                logger.debug("Full traceback:", exc_info=True)
+                sys.exit(1)
+            except Exception as error:
+                logger.critical(
+                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+                sys.exit(1)
+        else:
+            # External provider — init_parser classmethod via entry point
+            cls = Provider._load_ep_provider(provider)
+            if cls and hasattr(cls, "init_parser"):
+                try:
+                    cls.init_parser(self)
+                except Exception as error:
+                    logger.warning(
+                        f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                    )
 
 
 def validate_provider_arguments(arguments: Namespace) -> tuple[bool, str]:

prowler/providers/common/builtin.py

@@ -0,0 +1,29 @@
+"""Leaf helper for built-in provider detection.
+
+Lives in its own module — with no imports back into `prowler.lib.check` — so
+that callers in `prowler.lib.check.*` can ask "is this provider built-in?"
+without creating an import cycle through `prowler.providers.common.provider`
+(which transitively imports `prowler.config.config` and from there
+`prowler.lib.check.compliance_models` / `prowler.lib.check.external_tool_providers`).
+
+Same rationale as `prowler.lib.check.tool_wrapper`: extracting the predicate
+to a leaf module is the canonical way to break the cycle in this codebase.
+"""
+
+import importlib.util
+
+
+def is_builtin_provider(provider: str) -> bool:
+    """Return True if the provider's own package ships with the SDK.
+
+    Wraps `importlib.util.find_spec` in `try/except (ImportError, ValueError)`
+    because `find_spec` propagates `ModuleNotFoundError` when a parent package
+    in the dotted path does not exist (instead of returning `None`). The
+    try/except is what makes the call safe for external providers, whose
+    package does not live under `prowler.providers.{provider}`.
+    """
+    try:
+        spec = importlib.util.find_spec(f"prowler.providers.{provider}")
+        return spec is not None
+    except (ImportError, ValueError):
+        return False

prowler/providers/common/models.py

@@ -4,6 +4,7 @@ from os.path import isdir
 
 from pydantic.v1 import BaseModel
 
+from prowler.config.config import output_file_timestamp
 from prowler.providers.common.provider import Provider
 
 
@@ -69,3 +70,15 @@ class Connection:
 
     is_connected: bool = False
     error: Exception = None
+
+
+def default_output_options(provider, arguments, bulk_checks_metadata):
+    """Generic OutputOptions fallback for external providers that do not
+    implement get_output_options, so the run still produces output instead of
+    aborting. Honors arguments.output_filename and otherwise derives a name
+    from the provider type."""
+    output_options = ProviderOutputOptions(arguments, bulk_checks_metadata)
+    output_options.output_filename = getattr(arguments, "output_filename", None) or (
+        f"prowler-output-{provider.type}-{output_file_timestamp}"
+    )
+    return output_options

prowler/providers/common/provider.py

@@ -1,4 +1,6 @@
 import importlib
+import importlib.metadata
+import importlib.util
 import os
 import pkgutil
 import sys
@@ -136,6 +138,106 @@ class Provider(ABC):
         """
         return set()
 
+    # --- Dynamic provider contract methods (not @abstractmethod for incremental migration) ---
+
+    _cli_help_text: str = ""
+
+    @classmethod
+    def from_cli_args(cls, arguments: Namespace, fixer_config: dict) -> "Provider":
+        """Instantiate the provider from CLI arguments and return the instance.
+
+        The caller wires the returned instance into the global provider slot
+        via Provider.set_global_provider(). Implementations that already call
+        set_global_provider(self) from __init__ are also supported — the call
+        site tolerates a None return in that case.
+        """
+        raise NotImplementedError(f"{cls.__name__} has not implemented from_cli_args()")
+
+    def get_output_options(self, arguments, _bulk_checks_metadata):
+        """Create the provider-specific OutputOptions."""
+        raise NotImplementedError(
+            f"{self.__class__.__name__} has not implemented get_output_options()"
+        )
+
+    def get_stdout_detail(self, _finding) -> str:
+        """Return the detail string for stdout reporting (region, location, etc.)."""
+        raise NotImplementedError(
+            f"{self.__class__.__name__} has not implemented get_stdout_detail()"
+        )
+
+    def get_finding_sort_key(self) -> Optional[str]:
+        """Return the attribute name to sort findings by, or None for no sorting."""
+        return None
+
+    def get_summary_entity(self) -> tuple:
+        """Return (entity_type, audited_entities) for the summary table."""
+        return (self.type, getattr(self.identity, "account_id", ""))
+
+    def get_finding_output_data(self, _check_output) -> dict:
+        """Return provider-specific fields for Finding.generate_output()."""
+        raise NotImplementedError(
+            f"{self.__class__.__name__} has not implemented get_finding_output_data()"
+        )
+
+    def get_html_assessment_summary(self) -> str:
+        """Return the HTML assessment summary card for this provider."""
+        raise NotImplementedError(
+            f"{self.__class__.__name__} has not implemented get_html_assessment_summary()"
+        )
+
+    def generate_compliance_output(
+        self,
+        _findings,
+        _bulk_compliance_frameworks,
+        _input_compliance_frameworks,
+        _output_options,
+        _generated_outputs,
+    ) -> None:
+        """Generate compliance CSV output for this provider's frameworks."""
+        raise NotImplementedError(
+            f"{self.__class__.__name__} has not implemented generate_compliance_output()"
+        )
+
+    def get_mutelist_finding_args(self) -> dict:
+        """Return extra kwargs for mutelist.is_finding_muted() besides 'finding'.
+
+        External providers must return a dict with the identity key their
+        Mutelist subclass expects, e.g. ``{"account_id": self.identity.account_id}``.
+        The ``finding`` kwarg is added automatically by the caller.
+        """
+        raise NotImplementedError(
+            f"{self.__class__.__name__} has not implemented get_mutelist_finding_args()"
+        )
+
+    def display_compliance_table(
+        self,
+        _findings: list,
+        _bulk_checks_metadata: dict,
+        _compliance_framework: str,
+        _output_filename: str,
+        _output_directory: str,
+        _compliance_overview: bool,
+    ) -> bool:
+        """Render a custom compliance table in the terminal.
+
+        External providers can override this to display a detailed
+        compliance table (e.g., per-section breakdown). Return True
+        if the table was rendered, False to fall back to the generic table.
+        """
+        raise NotImplementedError(
+            f"{self.__class__.__name__} has not implemented display_compliance_table()"
+        )
+
+    # Class-level flag: True for providers that delegate scanning to an external
+    # tool (e.g. Trivy, promptfoo) and bypass standard check/service loading and
+    # metadata validation. Subclasses override as `is_external_tool_provider = True`.
+    # Kept as a class attribute (not a property) so it can be read from the class
+    # without instantiation — the metadata validators in lib.check.models need to
+    # decide whether to relax validation before any provider instance exists.
+    is_external_tool_provider: bool = False
+
+    # --- End dynamic provider contract methods ---
+
     @staticmethod
     def get_excluded_regions_from_env() -> set:
         """Parse the PROWLER_AWS_DISALLOWED_REGIONS environment variable.
@@ -159,20 +261,74 @@ class Provider(ABC):
     @staticmethod
     def init_global_provider(arguments: Namespace) -> None:
         try:
-            provider_class_path = (
-                f"{providers_path}.{arguments.provider}.{arguments.provider}_provider"
-            )
-            provider_class_name = f"{arguments.provider.capitalize()}Provider"
-            provider_class = getattr(
-                import_module(provider_class_path), provider_class_name
+            # Discriminate built-in vs external upfront via find_spec, so an
+            # ImportError from a transitive dependency missing inside a
+            # built-in's own import chain surfaces clearly instead of being
+            # silently re-routed to the entry-point path.
+            provider_class = None
+            if Provider.is_builtin(arguments.provider):
+                # Built-in wins on provider-name collision. Plug-ins are
+                # first-class extenders (they can register new provider
+                # names) but cannot override existing built-ins — a security
+                # tool prefers fail-loud predictability over silent
+                # overrides. Surface the override so the user knows their
+                # plug-in is being ignored and can rename it.
+                # Match by name only — never ep.load() a shadowing plug-in.
+                if any(
+                    ep.name == arguments.provider
+                    for ep in importlib.metadata.entry_points(group="prowler.providers")
+                ):
+                    logger.warning(
+                        f"Plug-in provider '{arguments.provider}' registered "
+                        f"via entry points is being IGNORED — a built-in with "
+                        f"the same name exists. To use your plug-in, register "
+                        f"it under a different name."
+                    )
+                provider_class_path = f"{providers_path}.{arguments.provider}.{arguments.provider}_provider"
+                provider_class_name = f"{arguments.provider.capitalize()}Provider"
+                try:
+                    provider_class = getattr(
+                        import_module(provider_class_path), provider_class_name
+                    )
+                except ImportError as e:
+                    logger.critical(
+                        f"Failed to load built-in provider '{arguments.provider}'. "
+                        f"Missing dependency: {e}. "
+                        f"Ensure all required dependencies are installed."
+                    )
+                    logger.debug("Full traceback:", exc_info=True)
+                    sys.exit(1)
+                except AttributeError:
+                    # Module exists but doesn't define the expected class —
+                    # treat as external and try entry points.
+                    provider_class = Provider._load_ep_provider(arguments.provider)
+            else:
+                provider_class = Provider._load_ep_provider(arguments.provider)
+
+            if provider_class is None:
+                raise ImportError(
+                    f"Provider '{arguments.provider}' not found as built-in or entry point"
+                )
+
+            # Kept for downstream forks that may extend the dispatch below
+            # with their own custom built-in branches and reference this name.
+            # The upstream chain dispatches by `arguments.provider` directly.
+            provider_class_name = (
+                f"{arguments.provider.capitalize()}Provider"  # noqa: F841
             )
 
             fixer_config = load_and_validate_config_file(
                 arguments.provider, arguments.fixer_config
             )
 
+            # Dispatch by exact provider name (equality, not substring) so
+            # external plug-ins whose names contain a built-in substring
+            # (e.g. `awsx`, `azure_gov`, `iac_v2`) cannot be silently routed
+            # to the wrong built-in branch. Anything that doesn't match a
+            # built-in falls through to the dynamic else and uses the
+            # contract's `from_cli_args`.
             if not isinstance(Provider._global, provider_class):
-                if "aws" in provider_class_name.lower():
+                if arguments.provider == "aws":
                     excluded_regions = (
                         set(arguments.excluded_region)
                         if getattr(arguments, "excluded_region", None)
@@ -196,7 +352,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "azure" in provider_class_name.lower():
+                elif arguments.provider == "azure":
                     provider_class(
                         az_cli_auth=arguments.az_cli_auth,
                         sp_env_auth=arguments.sp_env_auth,
@@ -209,7 +365,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "gcp" in provider_class_name.lower():
+                elif arguments.provider == "gcp":
                     provider_class(
                         retries_max_attempts=arguments.gcp_retries_max_attempts,
                         organization_id=arguments.organization_id,
@@ -223,7 +379,7 @@ class Provider(ABC):
                         fixer_config=fixer_config,
                         skip_api_check=arguments.skip_api_check,
                     )
-                elif "kubernetes" in provider_class_name.lower():
+                elif arguments.provider == "kubernetes":
                     provider_class(
                         kubeconfig_file=arguments.kubeconfig_file,
                         context=arguments.context,
@@ -233,7 +389,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "m365" in provider_class_name.lower():
+                elif arguments.provider == "m365":
                     provider_class(
                         region=arguments.region,
                         config_path=arguments.config_file,
@@ -247,7 +403,7 @@ class Provider(ABC):
                         init_modules=arguments.init_modules,
                         fixer_config=fixer_config,
                     )
-                elif "nhn" in provider_class_name.lower():
+                elif arguments.provider == "nhn":
                     provider_class(
                         username=arguments.nhn_username,
                         password=arguments.nhn_password,
@@ -256,7 +412,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "stackit" in provider_class_name.lower():
+                elif arguments.provider == "stackit":
                     provider_class(
                         project_id=arguments.stackit_project_id,
                         service_account_key_path=getattr(
@@ -275,7 +431,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "github" in provider_class_name.lower():
+                elif arguments.provider == "github":
                     orgs = []
                     repos = []
 
@@ -307,13 +463,13 @@ class Provider(ABC):
                         exclude_workflows=getattr(arguments, "exclude_workflows", []),
                         fixer_config=fixer_config,
                     )
-                elif "googleworkspace" in provider_class_name.lower():
+                elif arguments.provider == "googleworkspace":
                     provider_class(
                         config_path=arguments.config_file,
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "cloudflare" in provider_class_name.lower():
+                elif arguments.provider == "cloudflare":
                     provider_class(
                         filter_zones=arguments.region,
                         filter_accounts=arguments.account_id,
@@ -321,7 +477,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "iac" in provider_class_name.lower():
+                elif arguments.provider == "iac":
                     provider_class(
                         scan_path=arguments.scan_path,
                         scan_repository_url=arguments.scan_repository_url,
@@ -334,13 +490,13 @@ class Provider(ABC):
                         oauth_app_token=arguments.oauth_app_token,
                         provider_uid=arguments.provider_uid,
                     )
-                elif "llm" in provider_class_name.lower():
+                elif arguments.provider == "llm":
                     provider_class(
                         max_concurrency=arguments.max_concurrency,
                         config_path=arguments.config_file,
                         fixer_config=fixer_config,
                     )
-                elif "image" in provider_class_name.lower():
+                elif arguments.provider == "image":
                     provider_class(
                         images=arguments.images,
                         image_list_file=arguments.image_list_file,
@@ -358,7 +514,7 @@ class Provider(ABC):
                         registry_insecure=arguments.registry_insecure,
                         registry_list_images=arguments.registry_list_images,
                     )
-                elif "mongodbatlas" in provider_class_name.lower():
+                elif arguments.provider == "mongodbatlas":
                     provider_class(
                         atlas_public_key=arguments.atlas_public_key,
                         atlas_private_key=arguments.atlas_private_key,
@@ -367,7 +523,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "oraclecloud" in provider_class_name.lower():
+                elif arguments.provider == "oraclecloud":
                     provider_class(
                         oci_config_file=arguments.oci_config_file,
                         profile=arguments.profile,
@@ -378,7 +534,7 @@ class Provider(ABC):
                         fixer_config=fixer_config,
                         use_instance_principal=arguments.use_instance_principal,
                     )
-                elif "openstack" in provider_class_name.lower():
+                elif arguments.provider == "openstack":
                     provider_class(
                         clouds_yaml_file=getattr(arguments, "clouds_yaml_file", None),
                         clouds_yaml_content=getattr(
@@ -403,7 +559,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "alibabacloud" in provider_class_name.lower():
+                elif arguments.provider == "alibabacloud":
                     provider_class(
                         role_arn=arguments.role_arn,
                         role_session_name=arguments.role_session_name,
@@ -415,14 +571,14 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "vercel" in provider_class_name.lower():
+                elif arguments.provider == "vercel":
                     provider_class(
                         projects=getattr(arguments, "project", None),
                         config_path=arguments.config_file,
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "okta" in provider_class_name.lower():
+                elif arguments.provider == "okta":
                     provider_class(
                         okta_org_domain=getattr(arguments, "okta_org_domain", ""),
                         okta_client_id=getattr(arguments, "okta_client_id", ""),
@@ -435,7 +591,7 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
-                elif "scaleway" in provider_class_name.lower():
+                elif arguments.provider == "scaleway":
                     # Credentials are read from the SCW_ACCESS_KEY /
                     # SCW_SECRET_KEY env vars by the provider itself; there
                     # are no credential CLI flags to avoid leaking secrets.
@@ -447,6 +603,18 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         fixer_config=fixer_config,
                     )
+                else:
+                    # Dynamic fallback: any external/custom provider.
+                    # Honor the from_cli_args type hint (-> Provider): if the
+                    # implementation returns an instance, wire it as the global
+                    # provider here. Implementations that call
+                    # set_global_provider(self) from __init__ return None and
+                    # remain supported (the condition below is a no-op for them).
+                    provider_instance = provider_class.from_cli_args(
+                        arguments, fixer_config
+                    )
+                    if provider_instance is not None:
+                        Provider.set_global_provider(provider_instance)
 
         except TypeError as error:
             logger.critical(
@@ -459,17 +627,102 @@ class Provider(ABC):
             )
             sys.exit(1)
 
+    # Cache for entry-point provider classes {name: class}
+    _ep_providers: dict = {}
+
     @staticmethod
     def get_available_providers() -> list[str]:
         """get_available_providers returns a list of the available providers"""
-        providers = []
-        # Dynamically import the package based on its string path
+        providers = set()
+        # Built-in providers from local package
         prowler_providers = importlib.import_module(providers_path)
-        # Iterate over all modules found in the prowler_providers package
         for _, provider, ispkg in pkgutil.iter_modules(prowler_providers.__path__):
             if provider != "common" and ispkg:
-                providers.append(provider)
-        return providers
+                providers.add(provider)
+        # External providers registered via entry points
+        for ep in importlib.metadata.entry_points(group="prowler.providers"):
+            providers.add(ep.name)
+        return sorted(providers)
+
+    @staticmethod
+    def is_tool_wrapper_provider(provider: str) -> bool:
+        """Return True if the provider delegates scanning to an external tool.
+
+        Delegates to `prowler.lib.check.tool_wrapper.is_tool_wrapper_provider`,
+        the leaf module that holds the actual logic. Kept on `Provider` as a
+        convenience entry point for callers that already import `Provider`.
+        """
+        from prowler.lib.check.tool_wrapper import is_tool_wrapper_provider as _impl
+
+        return _impl(provider)
+
+    @staticmethod
+    def is_builtin(provider: str) -> bool:
+        """Return True if the provider's own package is importable as a built-in.
+
+        Delegates to `prowler.providers.common.builtin.is_builtin_provider`,
+        the leaf module that holds the actual check. Kept on `Provider` as a
+        convenience entry point for callers that already import `Provider`.
+        Call sites in `prowler.lib.check.*` should import from the leaf
+        directly to avoid the import cycle through this module.
+        """
+        from prowler.providers.common.builtin import is_builtin_provider as _impl
+
+        return _impl(provider)
+
+    @staticmethod
+    def _load_ep_provider(name: str):
+        """Load an external provider class from entry points, with cache.
+
+        Caches both hits and misses so repeated lookups for unknown names do
+        not re-iterate entry_points(). Symmetric with
+        tool_wrapper._ep_class_cache.
+        """
+        if name in Provider._ep_providers:
+            return Provider._ep_providers[name]
+        for ep in importlib.metadata.entry_points(group="prowler.providers"):
+            if ep.name == name:
+                try:
+                    cls = ep.load()
+                    Provider._ep_providers[name] = cls
+                    return cls
+                except Exception as error:
+                    logger.warning(
+                        f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                    )
+        Provider._ep_providers[name] = None
+        return None
+
+    @staticmethod
+    def get_providers_help_text() -> dict:
+        """Returns a dict of {provider_name: cli_help_text} for all available providers."""
+        help_text = {}
+        for name in Provider.get_available_providers():
+            try:
+                # Try built-in first
+                module_path = f"{providers_path}.{name}.{name}_provider"
+                module = import_module(module_path)
+                cls = None
+                for attr_name in dir(module):
+                    attr = getattr(module, attr_name)
+                    if (
+                        isinstance(attr, type)
+                        and issubclass(attr, Provider)
+                        and attr is not Provider
+                    ):
+                        cls = attr
+                        break
+                help_text[name] = getattr(cls, "_cli_help_text", "") if cls else ""
+            except ImportError:
+                # External provider — load via entry point
+                cls = Provider._load_ep_provider(name)
+                help_text[name] = getattr(cls, "_cli_help_text", "") if cls else ""
+            except Exception as error:
+                logger.warning(
+                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+                help_text[name] = ""
+        return help_text
 
     @staticmethod
     def update_provider_config(audit_config: dict, variable: str, value: str):

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/kms/kms_key_rotation_enabled/kms_key_rotation_enabled.metadata.json

@@ -1,14 +1,14 @@
 {
   "Provider": "gcp",
   "CheckID": "kms_key_rotation_enabled",
-  "CheckTitle": "KMS key is rotated at least annually",
+  "CheckTitle": "KMS key has automatic rotation enabled",
   "CheckType": [],
   "ServiceName": "kms",
   "SubServiceName": "",
   "ResourceIdTemplate": "",
   "Severity": "low",
   "ResourceType": "cloudkms.googleapis.com/CryptoKey",
-  "Description": "Google Cloud KMS customer-managed keys have **automatic rotation** enabled or a rotation interval  `365` days.\n\nThe evaluation reviews each key's rotation settings to confirm periodic creation of new key versions.",
+  "Description": "Google Cloud KMS customer-managed keys have **automatic rotation** enabled, regardless of the rotation interval.\n\nThe evaluation reviews each key's rotation settings to confirm that a rotation period is configured so new key versions are created periodically.",
   "Risk": "Without timely rotation, a stolen key can decrypt an expanding volume of data, eroding **confidentiality**. Prolonged key lifetimes widen windows for misuse, impact **integrity** of protected workloads, and make emergency rollover harder, risking **availability** disruptions.",
   "RelatedUrl": "",
   "AdditionalURLs": [
@@ -17,13 +17,13 @@
   ],
   "Remediation": {
     "Code": {
-      "CLI": "gcloud kms keys update <KEY_NAME> --keyring=<KEY_RING> --location=<LOCATION> --rotation-period=365d --next-rotation-time=<RFC3339_TIME>",
+      "CLI": "gcloud kms keys update <KEY_NAME> --keyring=<KEY_RING> --location=<LOCATION> --rotation-period=<PERIOD> --next-rotation-time=<RFC3339_TIME>",
       "NativeIaC": "",
-      "Other": "1. In Google Cloud Console, go to Security > Key Management > Key rings\n2. Open the key ring and select the key\n3. Click Edit rotation schedule (or Set rotation schedule)\n4. Set Rotation period to 365 days or less\n5. Set Next rotation date/time\n6. Click Save",
-      "Terraform": "```hcl\nresource \"google_kms_crypto_key\" \"<example_resource_name>\" {\n  name     = \"<example_resource_name>\"\n  key_ring = \"<example_resource_id>\"\n  purpose  = \"ENCRYPT_DECRYPT\"\n\n  rotation_period = \"31536000s\" # Critical: sets automatic rotation to 365 days (<= 365 ensures PASS)\n}\n```"
+      "Other": "1. In Google Cloud Console, go to Security > Key Management > Key rings\n2. Open the key ring and select the key\n3. Click Edit rotation schedule (or Set rotation schedule)\n4. Set a Rotation period\n5. Set Next rotation date/time\n6. Click Save",
+      "Terraform": "```hcl\nresource \"google_kms_crypto_key\" \"<example_resource_name>\" {\n  name     = \"<example_resource_name>\"\n  key_ring = \"<example_resource_id>\"\n  purpose  = \"ENCRYPT_DECRYPT\"\n\n  rotation_period = \"7776000s\" # Critical: enables automatic rotation (any period ensures PASS)\n}\n```"
     },
     "Recommendation": {
-      "Text": "Enable **auto-rotation** for customer-managed keys with an interval  `365` days.\n\nAdopt a **key lifecycle** policy: enforce **least privilege** on key usage, apply **separation of duties** between key admins and users, monitor key access, and rehearse emergency rotation to minimize blast radius.",
+      "Text": "Enable **auto-rotation** for customer-managed keys by configuring a rotation period.\n\nAdopt a **key lifecycle** policy: enforce **least privilege** on key usage, apply **separation of duties** between key admins and users, monitor key access, and rehearse emergency rotation to minimize blast radius.",
       "Url": "https://hub.prowler.com/check/kms_key_rotation_enabled"
     }
   },
@@ -31,6 +31,8 @@
     "encryption"
   ],
   "DependsOn": [],
-  "RelatedTo": [],
+  "RelatedTo": [
+    "kms_key_rotation_max_90_days"
+  ],
   "Notes": ""
 }

prowler/providers/gcp/services/kms/kms_key_rotation_enabled/kms_key_rotation_enabled.py

@@ -1,5 +1,3 @@
-import datetime
-
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.kms.kms_client import kms_client
 
@@ -9,36 +7,16 @@ class kms_key_rotation_enabled(Check):
         findings = []
         for key in kms_client.crypto_keys:
             report = Check_Report_GCP(metadata=self.metadata(), resource=key)
-            now = datetime.datetime.now()
-            condition_next_rotation_time = False
-            if key.next_rotation_time:
-                try:
-                    next_rotation_time = datetime.datetime.strptime(
-                        key.next_rotation_time, "%Y-%m-%dT%H:%M:%S.%fZ"
-                    )
-                except ValueError:
-                    next_rotation_time = datetime.datetime.strptime(
-                        key.next_rotation_time, "%Y-%m-%dT%H:%M:%SZ"
-                    )
-                condition_next_rotation_time = (
-                    abs((next_rotation_time - now).days) <= 90
-                )
-            condition_rotation_period = False
             if key.rotation_period:
-                condition_rotation_period = (
-                    int(key.rotation_period[:-1]) // (24 * 3600) <= 90
-                )
-            if condition_rotation_period and condition_next_rotation_time:
                 report.status = "PASS"
-                report.status_extended = f"Key {key.name} is rotated every 90 days or less and the next rotation time is in less than 90 days."
+                report.status_extended = (
+                    f"Key {key.name} has automatic rotation enabled."
+                )
             else:
                 report.status = "FAIL"
-                if condition_rotation_period:
-                    report.status_extended = f"Key {key.name} is rotated every 90 days or less but the next rotation time is in more than 90 days."
-                elif condition_next_rotation_time:
-                    report.status_extended = f"Key {key.name} is not rotated every 90 days or less but the next rotation time is in less than 90 days."
-                else:
-                    report.status_extended = f"Key {key.name} is not rotated every 90 days or less and the next rotation time is in more than 90 days."
+                report.status_extended = (
+                    f"Key {key.name} does not have automatic rotation enabled."
+                )
             findings.append(report)
 
         return findings

prowler/providers/gcp/services/kms/kms_key_rotation_max_90_days/kms_key_rotation_max_90_days.metadata.json

@@ -0,0 +1,38 @@
+{
+  "Provider": "gcp",
+  "CheckID": "kms_key_rotation_max_90_days",
+  "CheckTitle": "KMS key is rotated every 90 days or less",
+  "CheckType": [],
+  "ServiceName": "kms",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "low",
+  "ResourceType": "cloudkms.googleapis.com/CryptoKey",
+  "Description": "Google Cloud KMS customer-managed keys are rotated with an interval of `90` days or less, in line with the CIS Benchmark.\n\nThe evaluation reviews each key's rotation settings to confirm that both the rotation period and the next rotation time stay within 90 days.",
+  "Risk": "Without timely rotation, a stolen key can decrypt an expanding volume of data, eroding **confidentiality**. Prolonged key lifetimes widen windows for misuse, impact **integrity** of protected workloads, and make emergency rollover harder, risking **availability** disruptions.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://www.trendmicro.com/trendaivisiononecloudriskmanagement/knowledge-base/gcp/CloudKMS/rotate-kms-encryption-keys.html",
+    "https://cloud.google.com/iam/docs/manage-access-service-accounts"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "gcloud kms keys update <KEY_NAME> --keyring=<KEY_RING> --location=<LOCATION> --rotation-period=90d --next-rotation-time=<RFC3339_TIME>",
+      "NativeIaC": "",
+      "Other": "1. In Google Cloud Console, go to Security > Key Management > Key rings\n2. Open the key ring and select the key\n3. Click Edit rotation schedule (or Set rotation schedule)\n4. Set Rotation period to 90 days or less\n5. Set Next rotation date/time\n6. Click Save",
+      "Terraform": "```hcl\nresource \"google_kms_crypto_key\" \"<example_resource_name>\" {\n  name     = \"<example_resource_name>\"\n  key_ring = \"<example_resource_id>\"\n  purpose  = \"ENCRYPT_DECRYPT\"\n\n  rotation_period = \"7776000s\" # Critical: sets automatic rotation to 90 days (<= 90 ensures PASS)\n}\n```"
+    },
+    "Recommendation": {
+      "Text": "Enable **auto-rotation** for customer-managed keys with an interval of `90` days or less.\n\nAdopt a **key lifecycle** policy: enforce **least privilege** on key usage, apply **separation of duties** between key admins and users, monitor key access, and rehearse emergency rotation to minimize blast radius.",
+      "Url": "https://hub.prowler.com/check/kms_key_rotation_max_90_days"
+    }
+  },
+  "Categories": [
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [
+    "kms_key_rotation_enabled"
+  ],
+  "Notes": ""
+}

prowler/providers/gcp/services/kms/kms_key_rotation_max_90_days/kms_key_rotation_max_90_days.py

@@ -0,0 +1,44 @@
+import datetime
+
+from prowler.lib.check.models import Check, Check_Report_GCP
+from prowler.providers.gcp.services.kms.kms_client import kms_client
+
+
+class kms_key_rotation_max_90_days(Check):
+    def execute(self) -> Check_Report_GCP:
+        findings = []
+        for key in kms_client.crypto_keys:
+            report = Check_Report_GCP(metadata=self.metadata(), resource=key)
+            now = datetime.datetime.now()
+            condition_next_rotation_time = False
+            if key.next_rotation_time:
+                try:
+                    next_rotation_time = datetime.datetime.strptime(
+                        key.next_rotation_time, "%Y-%m-%dT%H:%M:%S.%fZ"
+                    )
+                except ValueError:
+                    next_rotation_time = datetime.datetime.strptime(
+                        key.next_rotation_time, "%Y-%m-%dT%H:%M:%SZ"
+                    )
+                condition_next_rotation_time = (
+                    abs((next_rotation_time - now).days) <= 90
+                )
+            condition_rotation_period = False
+            if key.rotation_period:
+                condition_rotation_period = (
+                    int(key.rotation_period[:-1]) // (24 * 3600) <= 90
+                )
+            if condition_rotation_period and condition_next_rotation_time:
+                report.status = "PASS"
+                report.status_extended = f"Key {key.name} is rotated every 90 days or less and the next rotation time is in less than 90 days."
+            else:
+                report.status = "FAIL"
+                if condition_rotation_period:
+                    report.status_extended = f"Key {key.name} is rotated every 90 days or less but the next rotation time is in more than 90 days."
+                elif condition_next_rotation_time:
+                    report.status_extended = f"Key {key.name} is not rotated every 90 days or less but the next rotation time is in less than 90 days."
+                else:
+                    report.status_extended = f"Key {key.name} is not rotated every 90 days or less and the next rotation time is in more than 90 days."
+            findings.append(report)
+
+        return findings

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.logging.logging_client import logging_client
+from prowler.providers.gcp.services.logging.logging_service import (
+    get_projects_covered_by_aggregated_metric,
+)
 from prowler.providers.gcp.services.monitoring.monitoring_client import (
     monitoring_client,
 )
@@ -10,12 +13,10 @@ class logging_log_metric_filter_and_alert_for_audit_configuration_changes_enable
 ):
     def execute(self) -> Check_Report_GCP:
         findings = []
+        metric_filter = 'protoPayload.methodName="SetIamPolicy" AND protoPayload.serviceData.policyDelta.auditConfigDeltas:*'
         projects_with_metric = set()
         for metric in logging_client.metrics:
-            if (
-                'protoPayload.methodName="SetIamPolicy" AND protoPayload.serviceData.policyDelta.auditConfigDeltas:*'
-                in metric.filter
-            ):
+            if metric_filter in metric.filter:
                 report = Check_Report_GCP(
                     metadata=self.metadata(),
                     resource=metric,
@@ -33,6 +34,11 @@ class logging_log_metric_filter_and_alert_for_audit_configuration_changes_enable
                             break
                 findings.append(report)
 
+        # Credit projects whose logs are centrally monitored via an org-level
+        # aggregated sink to a bucket-scoped metric + alert (instead of failing them).
+        centrally_covered = get_projects_covered_by_aggregated_metric(
+            logging_client, monitoring_client, metric_filter
+        )
         for project in logging_client.project_ids:
             if project not in projects_with_metric:
                 report = Check_Report_GCP(
@@ -46,8 +52,12 @@ class logging_log_metric_filter_and_alert_for_audit_configuration_changes_enable
                         else "GCP Project"
                     ),
                 )
-                report.status = "FAIL"
-                report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
+                if project in centrally_covered:
+                    report.status = "PASS"
+                    report.status_extended = f"Log metric filter {centrally_covered[project]} found with an alert, covering project {project} via an organization-level aggregated sink."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
                 findings.append(report)
 
         return findings

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.logging.logging_client import logging_client
+from prowler.providers.gcp.services.logging.logging_service import (
+    get_projects_covered_by_aggregated_metric,
+)
 from prowler.providers.gcp.services.monitoring.monitoring_client import (
     monitoring_client,
 )
@@ -8,12 +11,10 @@ from prowler.providers.gcp.services.monitoring.monitoring_client import (
 class logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled(Check):
     def execute(self) -> Check_Report_GCP:
         findings = []
+        metric_filter = 'resource.type="gcs_bucket" AND protoPayload.methodName="storage.setIamPermissions"'
         projects_with_metric = set()
         for metric in logging_client.metrics:
-            if (
-                'resource.type="gcs_bucket" AND protoPayload.methodName="storage.setIamPermissions"'
-                in metric.filter
-            ):
+            if metric_filter in metric.filter:
                 metric_name = getattr(metric, "name", None) or "unknown"
                 report = Check_Report_GCP(
                     metadata=self.metadata(),
@@ -36,6 +37,9 @@ class logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled(
                             break
                 findings.append(report)
 
+        centrally_covered = get_projects_covered_by_aggregated_metric(
+            logging_client, monitoring_client, metric_filter
+        )
         for project in logging_client.project_ids:
             if project not in projects_with_metric:
                 project_obj = logging_client.projects.get(project)
@@ -46,8 +50,12 @@ class logging_log_metric_filter_and_alert_for_bucket_permission_changes_enabled(
                     location=logging_client.region,
                     resource_name=(getattr(project_obj, "name", None) or "GCP Project"),
                 )
-                report.status = "FAIL"
-                report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
+                if project in centrally_covered:
+                    report.status = "PASS"
+                    report.status_extended = f"Log metric filter {centrally_covered[project]} found with an alert, covering project {project} via an organization-level aggregated sink."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
                 findings.append(report)
 
         return findings

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.logging.logging_client import logging_client
+from prowler.providers.gcp.services.logging.logging_service import (
+    get_projects_covered_by_aggregated_metric,
+)
 from prowler.providers.gcp.services.monitoring.monitoring_client import (
     monitoring_client,
 )
@@ -10,9 +13,10 @@ class logging_log_metric_filter_and_alert_for_compute_configuration_changes_enab
 ):
     def execute(self) -> Check_Report_GCP:
         findings = []
+        metric_filter = 'protoPayload.serviceName="compute.googleapis.com"'
         projects_with_metric = set()
         for metric in logging_client.metrics:
-            if 'protoPayload.serviceName="compute.googleapis.com"' in metric.filter:
+            if metric_filter in metric.filter:
                 report = Check_Report_GCP(
                     metadata=self.metadata(),
                     resource=metric,
@@ -30,6 +34,9 @@ class logging_log_metric_filter_and_alert_for_compute_configuration_changes_enab
                             break
                 findings.append(report)
 
+        centrally_covered = get_projects_covered_by_aggregated_metric(
+            logging_client, monitoring_client, metric_filter
+        )
         for project in logging_client.project_ids:
             if project not in projects_with_metric:
                 report = Check_Report_GCP(
@@ -43,8 +50,12 @@ class logging_log_metric_filter_and_alert_for_compute_configuration_changes_enab
                         else "GCP Project"
                     ),
                 )
-                report.status = "FAIL"
-                report.status_extended = f"There are no log metric filters or alerts associated for Compute Engine configuration changes in project {project}."
+                if project in centrally_covered:
+                    report.status = "PASS"
+                    report.status_extended = f"Log metric filter {centrally_covered[project]} found with an alert, covering project {project} via an organization-level aggregated sink."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"There are no log metric filters or alerts associated for Compute Engine configuration changes in project {project}."
                 findings.append(report)
 
         return findings

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.logging.logging_client import logging_client
+from prowler.providers.gcp.services.logging.logging_service import (
+    get_projects_covered_by_aggregated_metric,
+)
 from prowler.providers.gcp.services.monitoring.monitoring_client import (
     monitoring_client,
 )
@@ -8,12 +11,10 @@ from prowler.providers.gcp.services.monitoring.monitoring_client import (
 class logging_log_metric_filter_and_alert_for_custom_role_changes_enabled(Check):
     def execute(self) -> Check_Report_GCP:
         findings = []
+        metric_filter = 'resource.type="iam_role" AND (protoPayload.methodName="google.iam.admin.v1.CreateRole" OR protoPayload.methodName="google.iam.admin.v1.DeleteRole" OR protoPayload.methodName="google.iam.admin.v1.UpdateRole")'
         projects_with_metric = set()
         for metric in logging_client.metrics:
-            if (
-                'resource.type="iam_role" AND (protoPayload.methodName="google.iam.admin.v1.CreateRole" OR protoPayload.methodName="google.iam.admin.v1.DeleteRole" OR protoPayload.methodName="google.iam.admin.v1.UpdateRole")'
-                in metric.filter
-            ):
+            if metric_filter in metric.filter:
                 report = Check_Report_GCP(
                     metadata=self.metadata(),
                     resource=metric,
@@ -31,6 +32,9 @@ class logging_log_metric_filter_and_alert_for_custom_role_changes_enabled(Check)
                             break
                 findings.append(report)
 
+        centrally_covered = get_projects_covered_by_aggregated_metric(
+            logging_client, monitoring_client, metric_filter
+        )
         for project in logging_client.project_ids:
             if project not in projects_with_metric:
                 report = Check_Report_GCP(
@@ -44,8 +48,12 @@ class logging_log_metric_filter_and_alert_for_custom_role_changes_enabled(Check)
                         else "GCP Project"
                     ),
                 )
-                report.status = "FAIL"
-                report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
+                if project in centrally_covered:
+                    report.status = "PASS"
+                    report.status_extended = f"Log metric filter {centrally_covered[project]} found with an alert, covering project {project} via an organization-level aggregated sink."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
                 findings.append(report)
 
         return findings

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.logging.logging_client import logging_client
+from prowler.providers.gcp.services.logging.logging_service import (
+    get_projects_covered_by_aggregated_metric,
+)
 from prowler.providers.gcp.services.monitoring.monitoring_client import (
     monitoring_client,
 )
@@ -8,12 +11,10 @@ from prowler.providers.gcp.services.monitoring.monitoring_client import (
 class logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled(Check):
     def execute(self) -> Check_Report_GCP:
         findings = []
+        metric_filter = '(protoPayload.serviceName="cloudresourcemanager.googleapis.com") AND (ProjectOwnership OR projectOwnerInvitee) OR (protoPayload.serviceData.policyDelta.bindingDeltas.action="REMOVE" AND protoPayload.serviceData.policyDelta.bindingDeltas.role="roles/owner") OR (protoPayload.serviceData.policyDelta.bindingDeltas.action="ADD" AND protoPayload.serviceData.policyDelta.bindingDeltas.role="roles/owner")'
         projects_with_metric = set()
         for metric in logging_client.metrics:
-            if (
-                '(protoPayload.serviceName="cloudresourcemanager.googleapis.com") AND (ProjectOwnership OR projectOwnerInvitee) OR (protoPayload.serviceData.policyDelta.bindingDeltas.action="REMOVE" AND protoPayload.serviceData.policyDelta.bindingDeltas.role="roles/owner") OR (protoPayload.serviceData.policyDelta.bindingDeltas.action="ADD" AND protoPayload.serviceData.policyDelta.bindingDeltas.role="roles/owner")'
-                in metric.filter
-            ):
+            if metric_filter in metric.filter:
                 metric_name = getattr(metric, "name", None) or "unknown"
                 report = Check_Report_GCP(
                     metadata=self.metadata(),
@@ -36,6 +37,9 @@ class logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled(
                             break
                 findings.append(report)
 
+        centrally_covered = get_projects_covered_by_aggregated_metric(
+            logging_client, monitoring_client, metric_filter
+        )
         for project in logging_client.project_ids:
             if project not in projects_with_metric:
                 project_obj = logging_client.projects.get(project)
@@ -47,8 +51,12 @@ class logging_log_metric_filter_and_alert_for_project_ownership_changes_enabled(
                     location=logging_client.region,
                     resource_name=(getattr(project_obj, "name", None) or "GCP Project"),
                 )
-                report.status = "FAIL"
-                report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
+                if project in centrally_covered:
+                    report.status = "PASS"
+                    report.status_extended = f"Log metric filter {centrally_covered[project]} found with an alert, covering project {project} via an organization-level aggregated sink."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
                 findings.append(report)
 
         return findings

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.logging.logging_client import logging_client
+from prowler.providers.gcp.services.logging.logging_service import (
+    get_projects_covered_by_aggregated_metric,
+)
 from prowler.providers.gcp.services.monitoring.monitoring_client import (
     monitoring_client,
 )
@@ -10,9 +13,10 @@ class logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes
 ):
     def execute(self) -> Check_Report_GCP:
         findings = []
+        metric_filter = 'protoPayload.methodName="cloudsql.instances.update"'
         projects_with_metric = set()
         for metric in logging_client.metrics:
-            if 'protoPayload.methodName="cloudsql.instances.update"' in metric.filter:
+            if metric_filter in metric.filter:
                 report = Check_Report_GCP(
                     metadata=self.metadata(),
                     resource=metric,
@@ -30,6 +34,9 @@ class logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes
                             break
                 findings.append(report)
 
+        centrally_covered = get_projects_covered_by_aggregated_metric(
+            logging_client, monitoring_client, metric_filter
+        )
         for project in logging_client.project_ids:
             if project not in projects_with_metric:
                 report = Check_Report_GCP(
@@ -43,8 +50,12 @@ class logging_log_metric_filter_and_alert_for_sql_instance_configuration_changes
                         else "GCP Project"
                     ),
                 )
-                report.status = "FAIL"
-                report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
+                if project in centrally_covered:
+                    report.status = "PASS"
+                    report.status_extended = f"Log metric filter {centrally_covered[project]} found with an alert, covering project {project} via an organization-level aggregated sink."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
                 findings.append(report)
 
         return findings

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.logging.logging_client import logging_client
+from prowler.providers.gcp.services.logging.logging_service import (
+    get_projects_covered_by_aggregated_metric,
+)
 from prowler.providers.gcp.services.monitoring.monitoring_client import (
     monitoring_client,
 )
@@ -8,12 +11,10 @@ from prowler.providers.gcp.services.monitoring.monitoring_client import (
 class logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled(Check):
     def execute(self) -> Check_Report_GCP:
         findings = []
+        metric_filter = 'resource.type="gce_firewall_rule" AND (protoPayload.methodName:"compute.firewalls.patch" OR protoPayload.methodName:"compute.firewalls.insert" OR protoPayload.methodName:"compute.firewalls.delete")'
         projects_with_metric = set()
         for metric in logging_client.metrics:
-            if (
-                'resource.type="gce_firewall_rule" AND (protoPayload.methodName:"compute.firewalls.patch" OR protoPayload.methodName:"compute.firewalls.insert" OR protoPayload.methodName:"compute.firewalls.delete")'
-                in metric.filter
-            ):
+            if metric_filter in metric.filter:
                 report = Check_Report_GCP(
                     metadata=self.metadata(),
                     resource=metric,
@@ -31,6 +32,9 @@ class logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled(
                             break
                 findings.append(report)
 
+        centrally_covered = get_projects_covered_by_aggregated_metric(
+            logging_client, monitoring_client, metric_filter
+        )
         for project in logging_client.project_ids:
             if project not in projects_with_metric:
                 report = Check_Report_GCP(
@@ -44,8 +48,12 @@ class logging_log_metric_filter_and_alert_for_vpc_firewall_rule_changes_enabled(
                         else "GCP Project"
                     ),
                 )
-                report.status = "FAIL"
-                report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
+                if project in centrally_covered:
+                    report.status = "PASS"
+                    report.status_extended = f"Log metric filter {centrally_covered[project]} found with an alert, covering project {project} via an organization-level aggregated sink."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
                 findings.append(report)
 
         return findings

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_GCP
 from prowler.providers.gcp.services.logging.logging_client import logging_client
+from prowler.providers.gcp.services.logging.logging_service import (
+    get_projects_covered_by_aggregated_metric,
+)
 from prowler.providers.gcp.services.monitoring.monitoring_client import (
     monitoring_client,
 )
@@ -8,12 +11,10 @@ from prowler.providers.gcp.services.monitoring.monitoring_client import (
 class logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled(Check):
     def execute(self) -> Check_Report_GCP:
         findings = []
+        metric_filter = 'resource.type="gce_network" AND (protoPayload.methodName:"compute.networks.insert" OR protoPayload.methodName:"compute.networks.patch" OR protoPayload.methodName:"compute.networks.delete" OR protoPayload.methodName:"compute.networks.removePeering" OR protoPayload.methodName:"compute.networks.addPeering")'
         projects_with_metric = set()
         for metric in logging_client.metrics:
-            if (
-                'resource.type="gce_network" AND (protoPayload.methodName:"compute.networks.insert" OR protoPayload.methodName:"compute.networks.patch" OR protoPayload.methodName:"compute.networks.delete" OR protoPayload.methodName:"compute.networks.removePeering" OR protoPayload.methodName:"compute.networks.addPeering")'
-                in metric.filter
-            ):
+            if metric_filter in metric.filter:
                 report = Check_Report_GCP(
                     metadata=self.metadata(),
                     resource=metric,
@@ -31,6 +32,9 @@ class logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled(Check)
                             break
                 findings.append(report)
 
+        centrally_covered = get_projects_covered_by_aggregated_metric(
+            logging_client, monitoring_client, metric_filter
+        )
         for project in logging_client.project_ids:
             if project not in projects_with_metric:
                 report = Check_Report_GCP(
@@ -44,8 +48,12 @@ class logging_log_metric_filter_and_alert_for_vpc_network_changes_enabled(Check)
                         else "GCP Project"
                     ),
                 )
-                report.status = "FAIL"
-                report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
+                if project in centrally_covered:
+                    report.status = "PASS"
+                    report.status_extended = f"Log metric filter {centrally_covered[project]} found with an alert, covering project {project} via an organization-level aggregated sink."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"There are no log metric filters or alerts associated in project {project}."
                 findings.append(report)
 
         return findings