diff --git a/.github/workflows/sdk-tests.yml b/.github/workflows/sdk-tests.yml
index 0646e5f8b59..5009d379e14 100644
--- a/.github/workflows/sdk-tests.yml
+++ b/.github/workflows/sdk-tests.yml
@@ -516,6 +516,32 @@ jobs:
flags: prowler-py${{ matrix.python-version }}-vercel
files: ./vercel_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/**
+ ./poetry.lock
+
+ - name: Run External Provider tests
+ if: steps.changed-external.outputs.any_changed == 'true'
+ run: poetry 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'
diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md
index 3945d1621ab..5c059287dd6 100644
--- a/prowler/CHANGELOG.md
+++ b/prowler/CHANGELOG.md
@@ -6,6 +6,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
### 🚀 Added
+- Support for external/custom providers, checks, and compliance frameworks without modifying core code [(#10700)](https://github.com/prowler-cloud/prowler/pull/10700)
- `bedrock_guardrails_configured` check for AWS provider [(#10844)](https://github.com/prowler-cloud/prowler/pull/10844)
- Universal compliance pipeline integrated into the CLI: `--list-compliance` and `--list-compliance-requirements` show universal frameworks, and CSV plus OCSF outputs are generated for any framework declaring a `TableConfig` [(#10301)](https://github.com/prowler-cloud/prowler/pull/10301)
- ASD Essential Eight Maturity Model compliance framework for AWS (Maturity Level One, Nov 2023) [(#10808)](https://github.com/prowler-cloud/prowler/pull/10808)
@@ -70,6 +71,10 @@ All notable changes to the **Prowler SDK** are documented in this file.
- Google Workspace check reports now store the actual domain or account resource subject instead of `provider.identity` [(#10901)](https://github.com/prowler-cloud/prowler/pull/10901)
- `entra_users_mfa_capable` evaluating disabled guest accounts; CIS 5.2.3.4 only targets enabled member users [(#10785)](https://github.com/prowler-cloud/prowler/pull/10785)
+### 🐞 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)
+
---
## [5.24.3] (Prowler v5.24.3)
diff --git a/prowler/__main__.py b/prowler/__main__.py
index e10c9c745b2..8fc7b5f6d24 100644
--- a/prowler/__main__.py
+++ b/prowler/__main__.py
@@ -10,7 +10,6 @@
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,
@@ -207,9 +206,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
@@ -247,7 +247,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(
@@ -315,7 +315,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)
@@ -426,6 +426,9 @@ def prowler():
output_options = VercelOutputOptions(
args, bulk_checks_metadata, global_provider.identity
)
+ else:
+ # Dynamic fallback: any external/custom provider
+ output_options = global_provider.get_output_options(args, bulk_checks_metadata)
# Run the quick inventory for the provider if available
if hasattr(args, "quick_inventory") and args.quick_inventory:
@@ -435,7 +438,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":
@@ -445,12 +448,19 @@ def streaming_callback(findings_batch):
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%.
@@ -1343,6 +1353,30 @@ def streaming_callback(findings_batch):
)
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":
diff --git a/prowler/config/config.py b/prowler/config/config.py
index aa81e811366..d651211801c 100644
--- a/prowler/config/config.py
+++ b/prowler/config/config.py
@@ -1,3 +1,4 @@
+import importlib.metadata
import os
import pathlib
from datetime import datetime, timezone
@@ -82,13 +83,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:
@@ -97,7 +123,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"
@@ -114,6 +141,18 @@ def get_available_compliance_frameworks(provider=None):
continue
if name not in available_compliance_frameworks:
available_compliance_frameworks.append(name)
+ # External compliance via entry points.
+ # Multi-provider support for external plug-ins is tracked in PROWLER-1444.
+ 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)
return available_compliance_frameworks
@@ -225,18 +264,26 @@ def load_and_validate_config_file(provider: str, config_file_path: str) -> dict:
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, {}) 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.get(provider, {})
+ 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
diff --git a/prowler/lib/check/check.py b/prowler/lib/check/check.py
index b15cf8bfbe1..90956c79efc 100644
--- a/prowler/lib/check/check.py
+++ b/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.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()
@@ -745,6 +789,9 @@ def execute(
is_finding_muted_args["tenancy_id"] = (
global_provider.identity.tenancy_id
)
+ else:
+ # 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
diff --git a/prowler/lib/check/checks_loader.py b/prowler/lib/check/checks_loader.py
index 9ef672df6b3..77840084ffd 100644
--- a/prowler/lib/check/checks_loader.py
+++ b/prowler/lib/check/checks_loader.py
@@ -2,10 +2,10 @@
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
diff --git a/prowler/lib/check/compliance_models.py b/prowler/lib/check/compliance_models.py
index 226af6ec36e..ef11986662d 100644
--- a/prowler/lib/check/compliance_models.py
+++ b/prowler/lib/check/compliance_models.py
@@ -1,3 +1,4 @@
+import importlib.metadata
import json
import os
import sys
@@ -434,26 +435,55 @@ def get_bulk(provider: str) -> dict:
"""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:
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
+ ):
+ bulk_compliance_frameworks[
+ compliance_framework_name
+ ] = load_compliance_framework(file_path)
+ 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}")
diff --git a/prowler/lib/check/models.py b/prowler/lib/check/models.py
index f5f0ff87e81..5eb8ae50da1 100644
--- a/prowler/lib/check/models.py
+++ b/prowler/lib/check/models.py
@@ -11,10 +11,10 @@
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 @@ def valid_category(cls, value, values): # noqa: F841
)
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 @@ def validate_service_name(cls, service_name, values): # noqa: F841
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 @@ def valid_check_id(cls, check_id, values): # noqa: F841
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 @@ def valid_check_id(cls, check_id, values): # noqa: F841
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 @@ def validate_check_title(cls, check_title, values): # noqa: F841
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 @@ def validate_check_type(cls, check_type, values): # noqa: F841
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 @@ def validate_check_type(cls, check_type, values): # noqa: F841
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 @@ def validate_description(cls, description, values): # noqa: F841
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 @@ def get_bulk(provider: str) -> dict[str, "CheckMetadata"]:
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 @@ def list(
# 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 @@ def list(
# 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 = (
diff --git a/prowler/lib/check/tool_wrapper.py b/prowler/lib/check/tool_wrapper.py
new file mode 100644
index 00000000000..f00afd35d60
--- /dev/null
+++ b/prowler/lib/check/tool_wrapper.py
@@ -0,0 +1,57 @@
+"""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
+
+# 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
+ cls = _load_ep_class(provider)
+ return bool(cls and getattr(cls, "is_external_tool_provider", False))
diff --git a/prowler/lib/check/utils.py b/prowler/lib/check/utils.py
index 0e4807078f6..9c9a9c05238 100644
--- a/prowler/lib/check/utils.py
+++ b/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()
diff --git a/prowler/lib/cli/parser.py b/prowler/lib/cli/parser.py
index 24aa02083db..fc75e9618d1 100644
--- a/prowler/lib/cli/parser.py
+++ b/prowler/lib/cli/parser.py
@@ -20,19 +20,58 @@
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",
+ }
+ 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,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel,dashboard,iac,image,llm} ...",
- epilog="""
+ usage=f"prowler [-h] [--version] {{aws,azure,gcp,kubernetes,m365,github,googleworkspace,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel,dashboard,iac,image,llm{extra_providers_csv}}} ...",
+ epilog=f"""
Available Cloud Providers:
- {aws,azure,gcp,kubernetes,m365,github,googleworkspace,iac,llm,image,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel}
+ {{aws,azure,gcp,kubernetes,m365,github,googleworkspace,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel,dashboard,iac,image,llm{extra_providers_csv}}}
aws AWS Provider
azure Azure Provider
gcp GCP Provider
@@ -49,13 +88,13 @@ def __init__(self):
image Container Image Provider
nhn NHN Provider (Unofficial)
mongodbatlas MongoDB Atlas 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
""",
@@ -114,8 +153,10 @@ def parse(self, args=None) -> argparse.Namespace:
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
diff --git a/prowler/lib/outputs/compliance/compliance.py b/prowler/lib/outputs/compliance/compliance.py
index 7fe23305508..b3885b57c97 100644
--- a/prowler/lib/outputs/compliance/compliance.py
+++ b/prowler/lib/outputs/compliance/compliance.py
@@ -243,14 +243,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}"
diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py
index 95a634c85ff..e119d5ce153 100644
--- a/prowler/lib/outputs/finding.py
+++ b/prowler/lib/outputs/finding.py
@@ -474,6 +474,11 @@ def generate_output(
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
diff --git a/prowler/lib/outputs/html/html.py b/prowler/lib/outputs/html/html.py
index d547ea3b950..1cd097815bd 100644
--- a/prowler/lib/outputs/html/html.py
+++ b/prowler/lib/outputs/html/html.py
@@ -1417,11 +1417,13 @@ def get_assessment_summary(provider: Provider) -> str:
# 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}"
diff --git a/prowler/lib/outputs/outputs.py b/prowler/lib/outputs/outputs.py
index 1e2f6f2058d..7bb9a396da8 100644
--- a/prowler/lib/outputs/outputs.py
+++ b/prowler/lib/outputs/outputs.py
@@ -7,39 +7,46 @@
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 == "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
+ 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:
@@ -59,12 +66,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
@@ -75,12 +85,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
diff --git a/prowler/lib/outputs/summary_table.py b/prowler/lib/outputs/summary_table.py
index 1f30063c8f1..06d520ab327 100644
--- a/prowler/lib/outputs/summary_table.py
+++ b/prowler/lib/outputs/summary_table.py
@@ -108,6 +108,9 @@ def display_summary_table(
)
else:
audited_entities = provider.identity.username or "Personal Account"
+ 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):
diff --git a/prowler/lib/scan/scan.py b/prowler/lib/scan/scan.py
index 2ce7263e2be..4bef660d33c 100644
--- a/prowler/lib/scan/scan.py
+++ b/prowler/lib/scan/scan.py
@@ -4,8 +4,8 @@
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 @@ def 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()
diff --git a/prowler/providers/common/arguments.py b/prowler/providers/common/arguments.py
index 9745bab2cac..9e23274a903 100644
--- a/prowler/providers/common/arguments.py
+++ b/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]:
diff --git a/prowler/providers/common/builtin.py b/prowler/providers/common/builtin.py
new file mode 100644
index 00000000000..d60b5483d94
--- /dev/null
+++ b/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
diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py
index b0486d94103..cea9f091121 100644
--- a/prowler/providers/common/provider.py
+++ b/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,108 @@ def get_checks_to_execute_by_audit_resources(self) -> set:
"""
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."""
+ raise NotImplementedError(
+ f"{self.__class__.__name__} has not implemented get_summary_entity()"
+ )
+
+ 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 +263,70 @@ def set_global_provider(global_provider: "Provider") -> None:
@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.
+ if Provider._load_ep_provider(arguments.provider) is not None:
+ 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 +350,7 @@ def init_global_provider(arguments: Namespace) -> None:
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 +363,7 @@ def init_global_provider(arguments: Namespace) -> None:
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 +377,7 @@ def init_global_provider(arguments: Namespace) -> None:
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 +387,7 @@ def init_global_provider(arguments: Namespace) -> None:
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 +401,7 @@ def init_global_provider(arguments: Namespace) -> None:
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 +410,7 @@ def init_global_provider(arguments: Namespace) -> None:
mutelist_path=arguments.mutelist_file,
fixer_config=fixer_config,
)
- elif "github" in provider_class_name.lower():
+ elif arguments.provider == "github":
orgs = []
repos = []
@@ -288,13 +442,13 @@ def init_global_provider(arguments: Namespace) -> None:
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,
@@ -302,7 +456,7 @@ def init_global_provider(arguments: Namespace) -> None:
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,
@@ -315,13 +469,13 @@ def init_global_provider(arguments: Namespace) -> None:
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,
@@ -339,7 +493,7 @@ def init_global_provider(arguments: Namespace) -> None:
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,
@@ -348,7 +502,7 @@ def init_global_provider(arguments: Namespace) -> None:
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,
@@ -359,7 +513,7 @@ def init_global_provider(arguments: Namespace) -> None:
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(
@@ -384,7 +538,7 @@ def init_global_provider(arguments: Namespace) -> None:
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,
@@ -396,13 +550,25 @@ def init_global_provider(arguments: Namespace) -> None:
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,
)
+ 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(
@@ -415,17 +581,102 @@ def init_global_provider(arguments: Namespace) -> None:
)
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):
diff --git a/tests/config/config_test.py b/tests/config/config_test.py
index f7349d20045..0fe00d792f0 100644
--- a/tests/config/config_test.py
+++ b/tests/config/config_test.py
@@ -17,7 +17,7 @@
MOCK_PROWLER_MASTER_VERSION = "3.4.0"
-def mock_prowler_get_latest_release(_, **kwargs):
+def mock_prowler_get_latest_release(_, **_kwargs):
"""Mock requests.get() to get the Prowler latest release"""
response = Response()
response._content = b'[{"name":"3.3.0"}]'
@@ -463,6 +463,32 @@ def test_get_available_compliance_frameworks_does_not_mutate_provider_param(self
all_frameworks = get_available_compliance_frameworks()
assert "csa_ccm_4.0" in all_frameworks
+ @mock.patch("prowler.config.config._get_ep_compliance_dirs")
+ def test_get_available_compliance_frameworks_dedupes_ep_collisions_with_builtins(
+ self, mock_dirs
+ ):
+ """Entry-point compliance frameworks that collide with a built-in
+ name must appear only once in the available frameworks list.
+ Built-in wins silently — same policy as the universal frameworks
+ loop and as Compliance.get_bulk."""
+ import json
+ import tempfile
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ # cis_2.0_aws ships as a built-in under prowler/compliance/aws/
+ json_path = os.path.join(tmpdir, "cis_2.0_aws.json")
+ with open(json_path, "w") as f:
+ json.dump({"Framework": "CIS", "Provider": "aws"}, f)
+
+ mock_dirs.return_value = {"aws": tmpdir}
+
+ frameworks = get_available_compliance_frameworks("aws")
+
+ assert frameworks.count("cis_2.0_aws") == 1, (
+ f"Expected cis_2.0_aws to appear exactly once, got "
+ f"{frameworks.count('cis_2.0_aws')} occurrences in: {frameworks}"
+ )
+
def test_load_and_validate_config_file_aws(self):
path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
config_test_file = f"{path}/fixtures/config.yaml"
@@ -500,6 +526,32 @@ def test_load_and_validate_config_file_old_format(self):
assert load_and_validate_config_file("azure", config_test_file) == {}
assert load_and_validate_config_file("kubernetes", config_test_file) == {}
+ def test_load_and_validate_config_file_namespaced_non_listed_provider(self):
+ path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
+ config_test_file = f"{path}/fixtures/config_namespaced_external.yaml"
+ # github is a built-in not in the legacy hardcoded list; namespaced format must unwrap it.
+ assert load_and_validate_config_file("github", config_test_file) == {
+ "token": "abc",
+ "org": "prowler-cloud",
+ }
+
+ def test_load_and_validate_config_file_namespaced_external_provider(self):
+ path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
+ config_test_file = f"{path}/fixtures/config_namespaced_external.yaml"
+ # External plug-in provider: namespaced format must unwrap its block.
+ assert load_and_validate_config_file("custom_plugin", config_test_file) == {
+ "setting": "value",
+ "nested": {"key": 42},
+ }
+
+ def test_load_and_validate_config_file_namespaced_missing_provider(self):
+ path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
+ config_test_file = f"{path}/fixtures/config_namespaced_external.yaml"
+ # Provider with no section in a namespaced file must return empty config,
+ # not the full file (prevents cross-provider config leakage).
+ assert load_and_validate_config_file("aws", config_test_file) == {}
+ assert load_and_validate_config_file("gcp", config_test_file) == {}
+
def test_load_and_validate_config_file_invalid_config_file_path(self, caplog):
provider = "aws"
config_file_path = "invalid/path/to/fixer_config.yaml"
diff --git a/tests/config/fixtures/config_namespaced_external.yaml b/tests/config/fixtures/config_namespaced_external.yaml
new file mode 100644
index 00000000000..ec9f75c698a
--- /dev/null
+++ b/tests/config/fixtures/config_namespaced_external.yaml
@@ -0,0 +1,8 @@
+# Namespaced config covering a non-listed built-in (github) and an external plugin.
+github:
+ token: abc
+ org: prowler-cloud
+custom_plugin:
+ setting: value
+ nested:
+ key: 42
diff --git a/tests/lib/check/models_test.py b/tests/lib/check/models_test.py
index 71fd1f718b5..f17e359441c 100644
--- a/tests/lib/check/models_test.py
+++ b/tests/lib/check/models_test.py
@@ -95,6 +95,38 @@ def test_get_bulk(self, mock_recover_checks, mock_load_metadata):
"/path/to/accessanalyzer_enabled/accessanalyzer_enabled.metadata.json"
)
+ @mock.patch("prowler.lib.check.models.logger")
+ @mock.patch("prowler.lib.check.models.load_check_metadata")
+ @mock.patch("prowler.lib.check.models.recover_checks_from_provider")
+ def test_get_bulk_builtin_wins_on_check_id_collision(
+ self, mock_recover_checks, mock_load_metadata, mock_logger
+ ):
+ """Regression guard: when an entry-point plug-in re-registers a
+ built-in CheckID, the BUILT-IN metadata wins (first-write-wins) and
+ the plug-in is IGNORED. The override is surfaced via a warning so
+ the user knows their plug-in duplicate is being skipped and can
+ rename it. Matches the precedence in `_resolve_check_module`. See
+ PR #10700 review (HugoPBrito)."""
+ # Built-in first, plug-in last (matches recover_checks_from_provider order)
+ mock_recover_checks.return_value = [
+ ("accessanalyzer_enabled", "/builtin/accessanalyzer_enabled"),
+ ("accessanalyzer_enabled", "/plugin/accessanalyzer_enabled"),
+ ]
+
+ builtin_metadata = mock.MagicMock(CheckID="accessanalyzer_enabled")
+ plugin_metadata = mock.MagicMock(CheckID="accessanalyzer_enabled")
+ mock_load_metadata.side_effect = [builtin_metadata, plugin_metadata]
+
+ result = CheckMetadata.get_bulk(provider="aws")
+
+ # Built-in wins (first-write-wins on CheckID), plug-in is ignored
+ assert result["accessanalyzer_enabled"] is builtin_metadata
+ # Override is surfaced via warning naming the plug-in metadata file
+ mock_logger.warning.assert_called_once()
+ warning_msg = mock_logger.warning.call_args.args[0]
+ assert "accessanalyzer_enabled" in warning_msg
+ assert "/plugin/accessanalyzer_enabled" in warning_msg
+
@mock.patch("prowler.lib.check.models.load_check_metadata")
@mock.patch("prowler.lib.check.models.recover_checks_from_provider")
def test_list(self, mock_recover_checks, mock_load_metadata):
diff --git a/tests/lib/check/tool_wrapper_test.py b/tests/lib/check/tool_wrapper_test.py
new file mode 100644
index 00000000000..ed3b897d8db
--- /dev/null
+++ b/tests/lib/check/tool_wrapper_test.py
@@ -0,0 +1,110 @@
+"""Unit tests for prowler.lib.check.tool_wrapper.
+
+Covers the leaf helper directly (Provider.is_tool_wrapper_provider delegates
+to it). Tests the frozenset fast path, the entry-point fallback for external
+plug-ins, the broken-plug-in path, the no-match path, and the module-level
+cache.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _clear_ep_class_cache():
+ """Reset the leaf module's cache between tests so they stay independent."""
+ from prowler.lib.check import tool_wrapper
+
+ tool_wrapper._ep_class_cache.clear()
+ yield
+ tool_wrapper._ep_class_cache.clear()
+
+
+def _make_entry_point(name, cls):
+ """Create a mock entry point whose `load()` returns `cls`."""
+ ep = MagicMock()
+ ep.name = name
+ ep.load.return_value = cls
+ return ep
+
+
+class TestIsToolWrapperProvider:
+ """is_tool_wrapper_provider: frozenset + entry-point fallback."""
+
+ @pytest.mark.parametrize("name", ["iac", "llm", "image"])
+ def test_returns_true_for_builtin_tool_wrappers(self, name):
+ from prowler.lib.check.tool_wrapper import is_tool_wrapper_provider
+
+ assert is_tool_wrapper_provider(name) is True
+
+ @pytest.mark.parametrize("name", ["aws", "azure", "gcp", "github", "kubernetes"])
+ def test_returns_false_for_regular_builtins(self, name):
+ from prowler.lib.check.tool_wrapper import is_tool_wrapper_provider
+
+ assert is_tool_wrapper_provider(name) is False
+
+ @patch("prowler.lib.check.tool_wrapper.importlib.metadata.entry_points")
+ def test_returns_true_for_external_plugin_with_flag(self, mock_eps):
+ from prowler.lib.check.tool_wrapper import is_tool_wrapper_provider
+
+ cls = MagicMock(is_external_tool_provider=True)
+ mock_eps.return_value = [_make_entry_point("custom_wrapper", cls)]
+
+ assert is_tool_wrapper_provider("custom_wrapper") is True
+
+ @patch("prowler.lib.check.tool_wrapper.importlib.metadata.entry_points")
+ def test_returns_false_for_external_plugin_without_flag(self, mock_eps):
+ from prowler.lib.check.tool_wrapper import is_tool_wrapper_provider
+
+ cls = MagicMock(is_external_tool_provider=False)
+ mock_eps.return_value = [_make_entry_point("vanilla_external", cls)]
+
+ assert is_tool_wrapper_provider("vanilla_external") is False
+
+ @patch("prowler.lib.check.tool_wrapper.importlib.metadata.entry_points")
+ def test_returns_false_for_unknown_provider(self, mock_eps):
+ from prowler.lib.check.tool_wrapper import is_tool_wrapper_provider
+
+ mock_eps.return_value = []
+
+ assert is_tool_wrapper_provider("does-not-exist") is False
+
+
+class TestLoadEpClass:
+ """_load_ep_class: cache, broken plug-ins, no-match."""
+
+ @patch("prowler.lib.check.tool_wrapper.importlib.metadata.entry_points")
+ def test_caches_result_across_calls(self, mock_eps):
+ from prowler.lib.check.tool_wrapper import _load_ep_class
+
+ cls = MagicMock(is_external_tool_provider=True)
+ mock_eps.return_value = [_make_entry_point("cached_one", cls)]
+
+ first = _load_ep_class("cached_one")
+ second = _load_ep_class("cached_one")
+
+ assert first is cls
+ assert second is cls
+ # entry_points consulted only on the first call
+ assert mock_eps.call_count == 1
+
+ @patch("prowler.lib.check.tool_wrapper.importlib.metadata.entry_points")
+ def test_returns_none_for_broken_plugin(self, mock_eps):
+ from prowler.lib.check.tool_wrapper import _load_ep_class
+
+ broken_ep = MagicMock()
+ broken_ep.name = "broken"
+ broken_ep.load.side_effect = ImportError("plug-in is broken")
+ mock_eps.return_value = [broken_ep]
+
+ assert _load_ep_class("broken") is None
+
+ @patch("prowler.lib.check.tool_wrapper.importlib.metadata.entry_points")
+ def test_returns_none_when_no_entry_point_matches(self, mock_eps):
+ from prowler.lib.check.tool_wrapper import _load_ep_class
+
+ cls = MagicMock()
+ mock_eps.return_value = [_make_entry_point("other_provider", cls)]
+
+ assert _load_ep_class("missing_provider") is None
diff --git a/tests/lib/scan/scan_test.py b/tests/lib/scan/scan_test.py
index b0fe82b7b20..8037668b105 100644
--- a/tests/lib/scan/scan_test.py
+++ b/tests/lib/scan/scan_test.py
@@ -51,7 +51,7 @@ def mock_provider():
def mock_execute():
with mock.patch("prowler.lib.scan.scan.execute", autospec=True) as mock_exec:
findings = [finding]
- mock_exec.side_effect = lambda *args, **kwargs: findings
+ mock_exec.side_effect = lambda *_args, **_kwargs: findings
yield mock_exec
@@ -264,10 +264,10 @@ def test_init_with_no_checks(
@patch("prowler.lib.scan.scan.update_checks_metadata_with_compliance")
@patch("prowler.lib.scan.scan.Compliance.get_bulk")
@patch("prowler.lib.scan.scan.CheckMetadata.get_bulk")
- @patch("prowler.lib.scan.scan.import_check")
+ @patch("prowler.lib.scan.scan._resolve_check_module")
def test_scan(
self,
- mock_import_check,
+ mock_resolve_check_module,
mock_get_bulk,
mock_compliance_get_bulk,
mock_update_checks_metadata,
@@ -285,7 +285,7 @@ def test_scan(
mock_check_instance.CheckTitle = "Check if IAM Access Analyzer is enabled"
mock_check_instance.Categories = []
- mock_import_check.return_value = MagicMock(
+ mock_resolve_check_module.return_value = MagicMock(
accessanalyzer_enabled=mock_check_class
)
diff --git a/tests/providers/external/__init__.py b/tests/providers/external/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/tests/providers/external/test_dynamic_provider_loading.py b/tests/providers/external/test_dynamic_provider_loading.py
new file mode 100644
index 00000000000..9f45648c94a
--- /dev/null
+++ b/tests/providers/external/test_dynamic_provider_loading.py
@@ -0,0 +1,1909 @@
+"""
+Tests for dynamic provider loading via entry points.
+
+Covers: provider discovery, check discovery, check execution,
+CLI argument registration, compliance frameworks, parser integration,
+and all dispatch fallbacks for external providers.
+"""
+
+from argparse import Namespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from prowler.providers.common.provider import Provider
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_entry_point(name, value, group):
+ """Create a mock entry point."""
+ ep = MagicMock()
+ ep.name = name
+ ep.value = value
+ ep.group = group
+ return ep
+
+
+class FakeExternalProvider(Provider):
+ """Minimal Provider subclass for testing the dynamic contract."""
+
+ _type = "fakeexternal"
+ _cli_help_text = "Fake External Provider"
+
+ def __init__(self):
+ Provider.set_global_provider(self)
+
+ @property
+ def type(self):
+ return self._type
+
+ @property
+ def identity(self):
+ return MagicMock(host_id="fake-host-1")
+
+ @property
+ def session(self):
+ return MagicMock()
+
+ @property
+ def audit_config(self):
+ return {}
+
+ def setup_session(self):
+ return MagicMock()
+
+ def print_credentials(self):
+ pass
+
+ @classmethod
+ def from_cli_args(cls, _arguments, _fixer_config):
+ cls()
+
+ def get_output_options(self, _arguments, _bulk_checks_metadata):
+ return MagicMock(output_directory="/tmp", output_filename="fake")
+
+ def get_stdout_detail(self, finding):
+ return "fake-detail"
+
+ def get_finding_sort_key(self):
+ return "region"
+
+ def get_summary_entity(self):
+ return ("Fake Host", "fake-host-1")
+
+ def get_finding_output_data(self, check_output):
+ return {
+ "auth_method": "fake",
+ "account_uid": "fake-account",
+ "account_name": "fake",
+ "resource_name": "fake-resource",
+ "resource_uid": "fake-uid",
+ "region": "local",
+ }
+
+ def get_mutelist_finding_args(self):
+ return {"host_id": self.identity.host_id}
+
+ def display_compliance_table(
+ self,
+ findings,
+ _bulk_checks_metadata,
+ _compliance_framework,
+ _output_filename,
+ output_directory, # referenced via name elsewhere in tests
+ _compliance_overview,
+ ):
+ return True
+
+ def get_html_assessment_summary(self):
+ return "Fake Assessment
"
+
+ def generate_compliance_output(
+ self,
+ findings,
+ _bulk_compliance_frameworks,
+ _input_compliance_frameworks,
+ output_options,
+ generated_outputs,
+ ):
+ generated_outputs["compliance"].append("fake-compliance-output")
+
+ @classmethod
+ def init_parser(cls, parser_instance):
+ pass
+
+
+class FakeToolWrapperProvider(Provider):
+ """External provider that declares itself a tool wrapper."""
+
+ _type = "faketoolwrapper"
+ is_external_tool_provider = True
+
+ @property
+ def type(self):
+ return self._type
+
+ @property
+ def identity(self):
+ return MagicMock()
+
+ @property
+ def session(self):
+ return MagicMock()
+
+ @property
+ def audit_config(self):
+ return {}
+
+ def setup_session(self):
+ return MagicMock()
+
+ def print_credentials(self):
+ pass
+
+
+class FakePureContractProvider(Provider):
+ """External provider that honors the from_cli_args type hint literally:
+ returns an instance without calling Provider.set_global_provider() from
+ __init__. Used to verify the call site wires the returned instance."""
+
+ _type = "fakepure"
+
+ @property
+ def type(self):
+ return self._type
+
+ @property
+ def identity(self):
+ return MagicMock(host_id="fake-pure-1")
+
+ @property
+ def session(self):
+ return MagicMock()
+
+ @property
+ def audit_config(self):
+ return {}
+
+ def setup_session(self):
+ return MagicMock()
+
+ def print_credentials(self):
+ pass
+
+ @classmethod
+ def from_cli_args(cls, _arguments, _fixer_config):
+ # Literal contract: return the instance, no side-effect in __init__.
+ return cls()
+
+
+class FakeProviderNoHelpText(Provider):
+ """Provider without _cli_help_text."""
+
+ _type = "nohelptext"
+
+ @property
+ def type(self):
+ return self._type
+
+ @property
+ def identity(self):
+ return MagicMock()
+
+ @property
+ def session(self):
+ return MagicMock()
+
+ @property
+ def audit_config(self):
+ return {}
+
+ def setup_session(self):
+ return MagicMock()
+
+ def print_credentials(self):
+ pass
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _clear_ep_cache():
+ """Clear the entry point provider cache before each test."""
+ Provider._ep_providers = {}
+ yield
+ Provider._ep_providers = {}
+
+
+@pytest.fixture
+def fake_provider():
+ """Create and register a FakeExternalProvider."""
+ p = FakeExternalProvider()
+ yield p
+ Provider._global = None
+
+
+# ===========================================================================
+# 1. Provider Discovery & Loading
+# ===========================================================================
+
+
+class TestProviderDiscovery:
+ """Tests 1-7: get_available_providers, _load_ep_provider, get_providers_help_text."""
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_get_available_providers_merges_builtin_and_entrypoint(self, mock_ep):
+ """Test 1: get_available_providers returns both built-in and entry point providers."""
+ mock_ep.return_value = [
+ _make_entry_point("fakeexternal", "pkg.provider:Cls", "prowler.providers"),
+ ]
+
+ providers = Provider.get_available_providers()
+
+ # Built-in providers from actual prowler package
+ assert "aws" in providers
+ # External provider from entry point
+ assert "fakeexternal" in providers
+ assert "common" not in providers
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_get_available_providers_deduplicates(self, mock_ep):
+ """Test 2: Same provider name in built-in and entry point appears once."""
+ # "aws" exists as built-in AND as entry point
+ mock_ep.return_value = [
+ _make_entry_point("aws", "pkg.provider:Cls", "prowler.providers"),
+ ]
+
+ providers = Provider.get_available_providers()
+
+ assert providers.count("aws") == 1
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_load_ep_provider_loads_class(self, mock_ep):
+ """Test 3: _load_ep_provider loads the class from entry point."""
+ mock_ep.return_value = [
+ _make_entry_point(
+ "fakeexternal", "pkg:FakeExternalProvider", "prowler.providers"
+ ),
+ ]
+ mock_ep.return_value[0].load.return_value = FakeExternalProvider
+
+ cls = Provider._load_ep_provider("fakeexternal")
+
+ assert cls is FakeExternalProvider
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_load_ep_provider_returns_none_for_unknown(self, mock_ep):
+ """Test 4: _load_ep_provider returns None for unknown provider."""
+ mock_ep.return_value = []
+
+ cls = Provider._load_ep_provider("nonexistent")
+
+ assert cls is None
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_load_ep_provider_caches_result(self, mock_ep):
+ """Test 5: _load_ep_provider caches the loaded class."""
+ mock_ep.return_value = [
+ _make_entry_point("fakeexternal", "pkg:Cls", "prowler.providers"),
+ ]
+ mock_ep.return_value[0].load.return_value = FakeExternalProvider
+
+ cls1 = Provider._load_ep_provider("fakeexternal")
+ cls2 = Provider._load_ep_provider("fakeexternal")
+
+ assert cls1 is cls2
+ # load() should only be called once due to caching
+ mock_ep.return_value[0].load.assert_called_once()
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_load_ep_provider_caches_misses(self, mock_ep):
+ """A miss (unknown provider) must also be cached so repeated lookups
+ do not re-iterate entry_points(). Aligns with tool_wrapper._ep_class_cache,
+ which already caches None on miss."""
+ mock_ep.return_value = []
+
+ first = Provider._load_ep_provider("nonexistent")
+ second = Provider._load_ep_provider("nonexistent")
+
+ assert first is None
+ assert second is None
+ # entry_points() should only be called once across the two lookups
+ mock_ep.assert_called_once()
+
+ @patch("prowler.providers.common.provider.Provider._load_ep_provider")
+ @patch("prowler.providers.common.provider.Provider.get_available_providers")
+ def test_get_providers_help_text_reads_cli_help_text(
+ self, mock_providers, mock_load
+ ):
+ """Test 6: get_providers_help_text reads _cli_help_text from entry point provider."""
+ mock_providers.return_value = ["fakeexternal"]
+ mock_load.return_value = FakeExternalProvider
+
+ help_text = Provider.get_providers_help_text()
+
+ assert help_text["fakeexternal"] == "Fake External Provider"
+
+ @patch("prowler.providers.common.provider.Provider._load_ep_provider")
+ @patch("prowler.providers.common.provider.Provider.get_available_providers")
+ def test_get_providers_help_text_empty_without_cli_help_text(
+ self, mock_providers, mock_load
+ ):
+ """Test 7: get_providers_help_text returns empty string without _cli_help_text."""
+ mock_providers.return_value = ["nohelptext"]
+ mock_load.return_value = FakeProviderNoHelpText
+
+ help_text = Provider.get_providers_help_text()
+
+ assert help_text["nohelptext"] == ""
+
+
+class TestIsToolWrapperProvider:
+ """Tests for Provider.is_tool_wrapper_provider — the helper that combines the
+ built-in EXTERNAL_TOOL_PROVIDERS frozenset with the is_external_tool_provider
+ class attribute of entry-point providers."""
+
+ def test_returns_true_for_builtin_tool_wrappers(self):
+ # iac/llm/image are in the EXTERNAL_TOOL_PROVIDERS frozenset (fast path)
+ assert Provider.is_tool_wrapper_provider("iac") is True
+ assert Provider.is_tool_wrapper_provider("llm") is True
+ assert Provider.is_tool_wrapper_provider("image") is True
+
+ def test_returns_false_for_regular_builtin_providers(self):
+ # Regular built-ins must not be classified as tool wrappers
+ assert Provider.is_tool_wrapper_provider("aws") is False
+ assert Provider.is_tool_wrapper_provider("gcp") is False
+ assert Provider.is_tool_wrapper_provider("github") is False
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_returns_true_for_external_provider_declaring_flag(self, mock_ep):
+ # External plugin explicitly declares is_external_tool_provider = True
+ mock_ep.return_value = [
+ _make_entry_point("faketoolwrapper", "pkg:Cls", "prowler.providers"),
+ ]
+ mock_ep.return_value[0].load.return_value = FakeToolWrapperProvider
+
+ assert Provider.is_tool_wrapper_provider("faketoolwrapper") is True
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_returns_false_for_external_provider_without_flag(self, mock_ep):
+ # External plugin without the flag (default False) is treated as regular
+ mock_ep.return_value = [
+ _make_entry_point("fakeexternal", "pkg:Cls", "prowler.providers"),
+ ]
+ mock_ep.return_value[0].load.return_value = FakeExternalProvider
+
+ assert Provider.is_tool_wrapper_provider("fakeexternal") is False
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_returns_false_for_unknown_provider(self, mock_ep):
+ mock_ep.return_value = []
+
+ assert Provider.is_tool_wrapper_provider("does-not-exist") is False
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_returns_false_for_none_provider(self, mock_ep):
+ # Pydantic validators may pass None when values.get("Provider") is missing
+ mock_ep.return_value = []
+
+ assert Provider.is_tool_wrapper_provider(None) is False
+
+
+class TestIsBuiltinProvider:
+ """Tests for Provider.is_builtin — the helper that discriminates built-in
+ providers from external ones before attempting the import, so transitive
+ dependency failures in built-ins don't get silently re-routed to entry points."""
+
+ def test_returns_true_for_builtin_provider(self):
+ assert Provider.is_builtin("aws") is True
+ assert Provider.is_builtin("github") is True
+
+ def test_returns_false_for_unknown_provider(self):
+ assert Provider.is_builtin("nonexistent_xyz") is False
+
+ @patch("prowler.providers.common.provider.importlib.util.find_spec")
+ def test_returns_false_when_find_spec_raises(self, mock_find_spec):
+ # Certain namespace package edge cases raise ValueError/ImportError —
+ # helper should swallow and return False rather than propagate.
+ mock_find_spec.side_effect = ValueError("namespace package edge case")
+
+ assert Provider.is_builtin("some_provider") is False
+
+
+class TestInitProvidersParserBuiltinDependencyFailure:
+ """Tests the critical behavior fix: when a built-in provider's arguments
+ module exists but its imports fail (e.g. boto3 not installed), we must
+ fail loudly with a clear message — not silently fall through to entry
+ points as if the provider were external."""
+
+ @patch("prowler.providers.common.arguments.Provider.is_builtin")
+ @patch("prowler.providers.common.arguments.import_module")
+ def test_builtin_with_missing_transitive_dep_fails_loudly(
+ self, mock_import, mock_is_builtin
+ ):
+ from prowler.providers.common.arguments import init_providers_parser
+
+ mock_is_builtin.return_value = True
+ mock_import.side_effect = ImportError("No module named 'boto3'")
+
+ parser = MagicMock()
+ parser._providers = ["aws"]
+
+ with (
+ patch(
+ "prowler.providers.common.arguments.Provider.get_available_providers",
+ return_value=["aws"],
+ ),
+ pytest.raises(SystemExit),
+ ):
+ init_providers_parser(parser)
+
+ @patch("prowler.providers.common.arguments.Provider.is_builtin")
+ @patch("prowler.providers.common.arguments.Provider._load_ep_provider")
+ def test_external_provider_does_not_touch_builtin_path(
+ self, mock_load_ep, mock_is_builtin
+ ):
+ from prowler.providers.common.arguments import init_providers_parser
+
+ mock_is_builtin.return_value = False
+ ext_cls = MagicMock()
+ ext_cls.init_parser = MagicMock()
+ mock_load_ep.return_value = ext_cls
+
+ parser = MagicMock()
+
+ with patch(
+ "prowler.providers.common.arguments.Provider.get_available_providers",
+ return_value=["fakeexternal"],
+ ):
+ init_providers_parser(parser)
+
+ ext_cls.init_parser.assert_called_once_with(parser)
+
+
+class TestInitGlobalProviderBuiltinDependencyFailure:
+ """Same contract as TestInitProvidersParserBuiltinDependencyFailure but
+ for the provider class import path in init_global_provider."""
+
+ @patch("prowler.providers.common.provider.Provider.is_builtin")
+ @patch("prowler.providers.common.provider.import_module")
+ def test_builtin_with_missing_transitive_dep_fails_loudly(
+ self, mock_import, mock_is_builtin
+ ):
+ mock_is_builtin.return_value = True
+ mock_import.side_effect = ImportError("No module named 'boto3'")
+
+ args = Namespace(
+ provider="aws",
+ fixer_config="config.yaml",
+ config_file="config.yaml",
+ )
+
+ Provider._global = None
+ with pytest.raises(SystemExit):
+ Provider.init_global_provider(args)
+ Provider._global = None
+
+ @patch("prowler.providers.common.provider.importlib.metadata.entry_points")
+ def test_load_ep_provider_handles_load_exception(self, mock_ep):
+ """_load_ep_provider returns None when ep.load() raises."""
+ ep = _make_entry_point("broken", "pkg:Cls", "prowler.providers")
+ ep.load.side_effect = Exception("Import failed")
+ mock_ep.return_value = [ep]
+
+ cls = Provider._load_ep_provider("broken")
+
+ assert cls is None
+
+ @patch("prowler.providers.common.provider.import_module")
+ @patch("prowler.providers.common.provider.Provider.get_available_providers")
+ def test_get_providers_help_text_builtin_path(self, mock_providers, mock_import):
+ """get_providers_help_text reads _cli_help_text from a built-in provider module."""
+ import types
+
+ mock_providers.return_value = ["fakebuiltin"]
+
+ mock_cls = type(
+ "FakeBuiltinProvider", (Provider,), {"_cli_help_text": "Built-in Help"}
+ )
+ mock_module = types.ModuleType("fake_module")
+ mock_module.FakeBuiltinProvider = mock_cls
+ mock_import.return_value = mock_module
+
+ help_text = Provider.get_providers_help_text()
+
+ assert help_text["fakebuiltin"] == "Built-in Help"
+
+ @patch("prowler.providers.common.provider.import_module")
+ @patch("prowler.providers.common.provider.Provider.get_available_providers")
+ def test_get_providers_help_text_generic_exception(
+ self, mock_providers, mock_import
+ ):
+ """get_providers_help_text handles generic exceptions with empty string."""
+ mock_providers.return_value = ["broken"]
+ mock_import.side_effect = RuntimeError("Unexpected error")
+
+ help_text = Provider.get_providers_help_text()
+
+ assert help_text["broken"] == ""
+
+
+# ===========================================================================
+# 2. Provider Initialization
+# ===========================================================================
+
+
+class TestProviderInitialization:
+ """Tests 8-9: init_global_provider fallback to entry point."""
+
+ @patch("prowler.providers.common.provider.load_and_validate_config_file")
+ @patch("prowler.providers.common.provider.Provider._load_ep_provider")
+ @patch("prowler.providers.common.provider.import_module")
+ def test_init_global_provider_fallback_to_entry_point(
+ self, mock_import, mock_load_ep, mock_config
+ ):
+ """Test 8: init_global_provider falls back to entry point when built-in fails."""
+ mock_import.side_effect = ImportError("No built-in")
+ mock_load_ep.return_value = FakeExternalProvider
+ mock_config.return_value = {}
+
+ args = Namespace(
+ provider="fakeexternal",
+ fixer_config="config.yaml",
+ config_file="config.yaml",
+ )
+
+ Provider._global = None
+ Provider.init_global_provider(args)
+
+ assert isinstance(Provider._global, FakeExternalProvider)
+ Provider._global = None
+
+ @patch("prowler.providers.common.provider.load_and_validate_config_file")
+ @patch("prowler.providers.common.provider.Provider._load_ep_provider")
+ @patch("prowler.providers.common.provider.import_module")
+ def test_init_global_provider_exits_for_unknown_provider(
+ self, mock_import, mock_load_ep, mock_config
+ ):
+ """Test 9: init_global_provider exits when provider not found anywhere."""
+ mock_import.side_effect = ImportError("No built-in")
+ mock_load_ep.return_value = None
+ mock_config.return_value = {}
+
+ args = Namespace(
+ provider="nonexistent",
+ fixer_config="config.yaml",
+ config_file="config.yaml",
+ )
+
+ with pytest.raises(SystemExit):
+ Provider.init_global_provider(args)
+
+ @patch("prowler.providers.common.provider.load_and_validate_config_file")
+ @patch("prowler.providers.common.provider.Provider._load_ep_provider")
+ @patch("prowler.providers.common.provider.import_module")
+ def test_init_global_provider_wires_instance_returned_by_from_cli_args(
+ self, mock_import, mock_load_ep, mock_config
+ ):
+ """A provider that implements from_cli_args as a pure function (returns
+ the instance without calling set_global_provider from __init__) is
+ correctly wired as the global provider by init_global_provider."""
+ mock_import.side_effect = ImportError("No built-in")
+ mock_load_ep.return_value = FakePureContractProvider
+ mock_config.return_value = {}
+
+ args = Namespace(
+ provider="fakepure",
+ fixer_config="config.yaml",
+ config_file="config.yaml",
+ )
+
+ Provider._global = None
+ Provider.init_global_provider(args)
+
+ assert isinstance(Provider._global, FakePureContractProvider)
+ Provider._global = None
+
+ @pytest.mark.parametrize(
+ "plugin_name",
+ [
+ "awsx",
+ "aws_lite",
+ "azure_gov",
+ "gcp_org",
+ "github_enterprise",
+ "iac_v2",
+ ],
+ )
+ @patch("prowler.providers.common.provider.load_and_validate_config_file")
+ @patch("prowler.providers.common.provider.Provider._load_ep_provider")
+ @patch("prowler.providers.common.provider.import_module")
+ def test_init_global_provider_external_with_builtin_substring_uses_from_cli_args(
+ self, mock_import, mock_load_ep, mock_config, plugin_name
+ ):
+ """Regression guard for the substring footgun in the dispatch chain.
+
+ An external plug-in whose name contains a built-in substring
+ (e.g. `awsx`, `aws_lite`, `azure_gov`, `gcp_org`, `github_enterprise`,
+ `iac_v2`) MUST be routed to the dynamic else and instantiated via
+ `from_cli_args` — not silently captured by the built-in branch whose
+ name happens to be a substring of the plug-in name. See PR #10700
+ review.
+ """
+ mock_import.side_effect = ImportError("No built-in")
+ mock_load_ep.return_value = FakeExternalProvider
+ mock_config.return_value = {}
+
+ # Namespace deliberately omits the kwargs of any built-in branch
+ # (no `aws_retries_max_attempts`, `az_cli_auth`, `personal_access_token`,
+ # etc.). If equality dispatch is broken and the plug-in is misrouted to
+ # a built-in branch, attribute access will raise and the global never
+ # gets wired.
+ args = Namespace(
+ provider=plugin_name,
+ fixer_config="config.yaml",
+ config_file="config.yaml",
+ )
+
+ Provider._global = None
+ Provider.init_global_provider(args)
+
+ assert isinstance(Provider._global, FakeExternalProvider)
+ Provider._global = None
+
+ @patch("prowler.providers.common.provider.logger")
+ @patch("prowler.providers.common.provider.load_and_validate_config_file")
+ @patch("prowler.providers.common.provider.Provider._load_ep_provider")
+ @patch("prowler.providers.common.provider.import_module")
+ @patch("prowler.providers.common.provider.Provider.is_builtin")
+ def test_init_global_provider_warns_when_plugin_shadowed_by_builtin(
+ self, mock_is_builtin, mock_import, mock_load_ep, mock_config, mock_logger
+ ):
+ """Regression guard: when a plug-in registers a provider name that
+ collides with a built-in, the BUILT-IN wins and a warning is emitted
+ naming the shadowed plug-in. Matches the precedence enforced by
+ `_resolve_check_module` and `CheckMetadata.get_bulk` for checks. See
+ PR #10700 review (HugoPBrito).
+ """
+ # Simulate a built-in `aws` that exists, AND a plug-in registered
+ # under the same `aws` name via entry points.
+ mock_is_builtin.return_value = True
+ mock_load_ep.return_value = FakeExternalProvider # plug-in shadow
+ mock_import.return_value = MagicMock(
+ AwsProvider=MagicMock(side_effect=lambda **_kw: None)
+ )
+ mock_config.return_value = {}
+
+ args = Namespace(
+ provider="aws",
+ fixer_config="config.yaml",
+ config_file="config.yaml",
+ aws_retries_max_attempts=3,
+ role=None,
+ session_duration=None,
+ external_id=None,
+ role_session_name=None,
+ mfa=None,
+ profile=None,
+ region=None,
+ excluded_region=None,
+ organizations_role=None,
+ scan_unused_services=False,
+ resource_tag=None,
+ resource_arn=None,
+ mutelist_file=None,
+ )
+
+ Provider._global = None
+ try:
+ Provider.init_global_provider(args)
+ except BaseException:
+ # The AwsProvider mock is fake and the dispatch may sys.exit on
+ # the simulated failure; we only care about the warning emitted
+ # before the dispatch happens.
+ pass
+ finally:
+ Provider._global = None
+
+ # Warning was emitted naming the shadowed plug-in
+ warning_msgs = [
+ call.args[0]
+ for call in mock_logger.warning.call_args_list
+ if call.args and "Plug-in provider 'aws'" in call.args[0]
+ ]
+ assert warning_msgs, "expected a warning about the shadowed plug-in 'aws'"
+ assert "IGNORED" in warning_msgs[0]
+
+
+# ===========================================================================
+# 3. Check Discovery
+# ===========================================================================
+
+
+class TestCheckDiscovery:
+ """Tests 10-14: _recover_ep_checks, recover_checks_from_provider."""
+
+ @patch("prowler.lib.check.utils.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_ep_checks_discovers_checks(self, mock_spec, mock_ep):
+ """Test 10: _recover_ep_checks discovers checks from entry points."""
+ from prowler.lib.check.utils import _recover_ep_checks
+
+ mock_ep.return_value = [
+ _make_entry_point("my_check", "pkg.checks.my_check", "prowler.checks.fake"),
+ ]
+ mock_spec_obj = MagicMock()
+ mock_spec_obj.origin = "/path/to/pkg/checks/my_check.py"
+ mock_spec.return_value = mock_spec_obj
+
+ checks = _recover_ep_checks("fake")
+
+ assert len(checks) == 1
+ assert checks[0][0] == "my_check"
+ assert checks[0][1] == "/path/to/pkg/checks"
+
+ @patch("prowler.lib.check.utils.importlib.metadata.entry_points")
+ def test_recover_ep_checks_empty_without_entry_points(self, mock_ep):
+ """Test 11: _recover_ep_checks returns empty list with no entry points."""
+ from prowler.lib.check.utils import _recover_ep_checks
+
+ mock_ep.return_value = []
+
+ checks = _recover_ep_checks("fake")
+
+ assert checks == []
+
+ @patch("prowler.lib.check.utils.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_ep_checks_handles_broken_entry_point(self, mock_spec, mock_ep):
+ """Test 12: _recover_ep_checks handles failed entry points gracefully."""
+ from prowler.lib.check.utils import _recover_ep_checks
+
+ mock_ep.return_value = [
+ _make_entry_point("broken_check", "pkg.broken", "prowler.checks.fake"),
+ ]
+ mock_spec.side_effect = Exception("Module not found")
+
+ checks = _recover_ep_checks("fake")
+
+ assert checks == []
+
+ @patch("prowler.lib.check.utils._recover_ep_checks")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_checks_handles_external_provider_without_services(
+ self, mock_find_spec, mock_ep_checks
+ ):
+ """Test 13: recover_checks_from_provider doesn't crash for external providers.
+
+ With find_spec returning None (built-in package doesn't exist), discovery
+ falls through to entry points cleanly — no ModuleNotFoundError catch
+ needed.
+ """
+ from prowler.lib.check.utils import recover_checks_from_provider
+
+ mock_find_spec.return_value = None # not a built-in
+ mock_ep_checks.return_value = [("ext_check", "/path/to/check")]
+
+ checks = recover_checks_from_provider("fakeexternal")
+
+ assert len(checks) == 1
+ assert checks[0][0] == "ext_check"
+
+ @patch("prowler.lib.check.utils._recover_ep_checks")
+ @patch("prowler.lib.check.utils.list_modules")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_checks_combines_builtin_and_entry_points(
+ self, mock_find_spec, mock_list_modules, mock_ep_checks
+ ):
+ """Test 14: recover_checks_from_provider combines built-in and entry point checks."""
+ from prowler.lib.check.utils import recover_checks_from_provider
+
+ mock_find_spec.return_value = MagicMock() # built-in package exists
+
+ # Simulate a built-in module
+ builtin_module = MagicMock()
+ builtin_module.name = "prowler.providers.aws.services.ec2.check_a.check_a"
+ builtin_module.module_finder.path = "/builtin/path"
+ mock_list_modules.return_value = [builtin_module]
+
+ mock_ep_checks.return_value = [("check_b", "/external/path")]
+
+ checks = recover_checks_from_provider("aws")
+
+ check_names = [c[0] for c in checks]
+ assert "check_a" in check_names
+ assert "check_b" in check_names
+
+ @patch("prowler.lib.check.utils.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_ep_checks_filters_by_service(self, mock_spec, mock_ep):
+ """Service filter keeps only entry points whose dotted path includes
+ `.services.{service}.` — mirroring the built-in package filter."""
+ from prowler.lib.check.utils import _recover_ep_checks
+
+ mock_ep.return_value = [
+ _make_entry_point(
+ "container_has_no_root_user",
+ "prowler_artifacts_dockerdesktop.services.container.container_has_no_root_user.container_has_no_root_user",
+ "prowler.checks.dockerdesktop",
+ ),
+ _make_entry_point(
+ "image_is_signed",
+ "prowler_artifacts_dockerdesktop.services.image.image_is_signed.image_is_signed",
+ "prowler.checks.dockerdesktop",
+ ),
+ ]
+ mock_spec_obj = MagicMock()
+ mock_spec_obj.origin = "/some/path/check.py"
+ mock_spec.return_value = mock_spec_obj
+
+ checks = _recover_ep_checks("dockerdesktop", service="container")
+
+ assert len(checks) == 1
+ assert checks[0][0] == "container_has_no_root_user"
+
+ @patch("prowler.lib.check.utils.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_ep_checks_without_service_returns_all(self, mock_spec, mock_ep):
+ """Without a service filter, all entry points for the provider are returned."""
+ from prowler.lib.check.utils import _recover_ep_checks
+
+ mock_ep.return_value = [
+ _make_entry_point(
+ "container_has_no_root_user",
+ "prowler_artifacts_dockerdesktop.services.container.container_has_no_root_user.container_has_no_root_user",
+ "prowler.checks.dockerdesktop",
+ ),
+ _make_entry_point(
+ "image_is_signed",
+ "prowler_artifacts_dockerdesktop.services.image.image_is_signed.image_is_signed",
+ "prowler.checks.dockerdesktop",
+ ),
+ ]
+ mock_spec_obj = MagicMock()
+ mock_spec_obj.origin = "/some/path/check.py"
+ mock_spec.return_value = mock_spec_obj
+
+ checks = _recover_ep_checks("dockerdesktop")
+
+ assert len(checks) == 2
+
+ @patch("prowler.lib.check.utils._recover_ep_checks")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_checks_external_provider_with_service(
+ self, mock_find_spec, mock_ep_checks
+ ):
+ """External provider with --service: built-in package doesn't exist,
+ but entry points are still consulted and return the requested service's
+ checks. No premature sys.exit."""
+ from prowler.lib.check.utils import recover_checks_from_provider
+
+ mock_find_spec.return_value = None # not a built-in
+ mock_ep_checks.return_value = [("container_check", "/ext/path")]
+
+ checks = recover_checks_from_provider("dockerdesktop", service="container")
+
+ assert len(checks) == 1
+ assert checks[0][0] == "container_check"
+ mock_ep_checks.assert_called_once_with("dockerdesktop", "container")
+
+ @patch("prowler.lib.check.utils._recover_ep_checks")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_checks_unknown_service_fails_cleanly(
+ self, mock_find_spec, mock_ep_checks
+ ):
+ """A typo or unknown service (not in built-ins nor in entry points)
+ fails with a clear error message, not a silent empty result."""
+ from prowler.lib.check.utils import recover_checks_from_provider
+
+ mock_find_spec.return_value = None
+ mock_ep_checks.return_value = []
+
+ with pytest.raises(SystemExit):
+ recover_checks_from_provider("aws", service="typo_service")
+
+ @patch("prowler.lib.check.utils._recover_ep_checks")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_checks_builtin_with_new_external_service(
+ self, mock_find_spec, mock_ep_checks
+ ):
+ """Built-in provider with a new service added via entry points:
+ the built-in package for that specific service doesn't exist (find_spec
+ returns None), but entry points pick it up. The gate `if not service:`
+ that previously skipped entry points when --service was passed is removed."""
+ from prowler.lib.check.utils import recover_checks_from_provider
+
+ mock_find_spec.return_value = None # built-in for new_aws_service doesn't exist
+ mock_ep_checks.return_value = [("new_check", "/ext/path")]
+
+ checks = recover_checks_from_provider("aws", service="new_aws_service")
+
+ assert len(checks) == 1
+ assert checks[0][0] == "new_check"
+ mock_ep_checks.assert_called_once_with("aws", "new_aws_service")
+
+ @patch("prowler.lib.check.utils._recover_ep_checks")
+ @patch("prowler.lib.check.utils.list_modules")
+ @patch("prowler.lib.check.utils.importlib.util.find_spec")
+ def test_recover_checks_surfaces_error_when_builtin_service_import_fails(
+ self, mock_find_spec, mock_list_modules, mock_ep_checks
+ ):
+ """Regression guard: when a built-in service's package exists but one
+ of its modules fails to import (e.g. a broken transitive dependency),
+ the error must surface via the global exception handler — not be
+ silently swallowed and replaced by an entry-point plug-in that happens
+ to share a name. See PR #10700 review (HugoPBrito)."""
+ from prowler.lib.check.utils import recover_checks_from_provider
+
+ mock_find_spec.return_value = MagicMock() # built-in service exists
+ mock_list_modules.side_effect = ImportError("missing transitive dep: foo")
+
+ # Even if a plug-in registers checks for the same service, it must NOT
+ # silently take over — the import error wins.
+ mock_ep_checks.return_value = [("evil_check", "/evil/path")]
+
+ with pytest.raises(SystemExit):
+ recover_checks_from_provider("aws", service="ec2")
+
+
+# ===========================================================================
+# 4. Check Execution
+# ===========================================================================
+
+
+class TestCheckExecution:
+ """Tests 15-17: _resolve_check_module."""
+
+ @patch("prowler.lib.check.check.importlib.util.find_spec")
+ @patch("prowler.lib.check.check.import_check")
+ def test_resolve_check_module_builtin_first(self, mock_import, mock_find_spec):
+ """Test 15: _resolve_check_module resolves built-in checks first."""
+ from prowler.lib.check.check import _resolve_check_module
+
+ mock_module = MagicMock()
+ mock_import.return_value = mock_module
+ mock_find_spec.return_value = MagicMock() # built-in package exists
+
+ result = _resolve_check_module("aws", "ec2", "my_check")
+
+ assert result is mock_module
+ mock_import.assert_called_once_with(
+ "prowler.providers.aws.services.ec2.my_check.my_check"
+ )
+
+ @patch("prowler.lib.check.check.importlib.util.find_spec")
+ @patch("prowler.lib.check.check.import_check")
+ def test_resolve_check_module_fallback_to_entry_point(
+ self, mock_import_check, mock_find_spec
+ ):
+ """Test 16: _resolve_check_module falls back to entry point when built-in is absent."""
+ from prowler.lib.check.check import _resolve_check_module
+
+ mock_find_spec.return_value = None # built-in does not exist
+
+ mock_ext_module = MagicMock()
+ ep = _make_entry_point(
+ "my_check", "ext_pkg.checks.my_check", "prowler.checks.fake"
+ )
+
+ with (
+ patch("importlib.metadata.entry_points", return_value=[ep]),
+ patch("importlib.import_module", return_value=mock_ext_module) as mock_imp,
+ ):
+ result = _resolve_check_module("fake", "svc", "my_check")
+
+ assert result is mock_ext_module
+ mock_imp.assert_called_with("ext_pkg.checks.my_check")
+ mock_import_check.assert_not_called()
+
+ @patch("prowler.lib.check.check.importlib.util.find_spec")
+ @patch("prowler.lib.check.check.import_check")
+ def test_resolve_check_module_builtin_wins_over_entry_point(
+ self, mock_import_check, mock_find_spec
+ ):
+ """Regression guard: when both a built-in and an entry-point check
+ exist with the same CheckID, the BUILT-IN wins. Plug-ins extend
+ Prowler with new checks but cannot silently override existing
+ built-ins — a security tool prefers fail-loud predictability over
+ permissive overrides. CheckMetadata.get_bulk applies the same
+ precedence (first-write-wins) and emits a warning. See PR #10700
+ review (HugoPBrito)."""
+ from prowler.lib.check.check import _resolve_check_module
+
+ mock_find_spec.return_value = MagicMock() # built-in exists
+ builtin_module = MagicMock()
+ mock_import_check.return_value = builtin_module
+
+ # Plug-in registers same CheckID — must NOT take precedence
+ ep = _make_entry_point(
+ "ec2_instance_public_ip",
+ "plug_pkg.checks.ec2_instance_public_ip",
+ "prowler.checks.aws",
+ )
+
+ with (
+ patch("importlib.metadata.entry_points", return_value=[ep]),
+ patch("importlib.import_module") as mock_imp,
+ ):
+ result = _resolve_check_module("aws", "ec2", "ec2_instance_public_ip")
+
+ assert result is builtin_module
+ mock_import_check.assert_called_once_with(
+ "prowler.providers.aws.services.ec2.ec2_instance_public_ip.ec2_instance_public_ip"
+ )
+ # Plug-in must NOT be loaded when a built-in with the same CheckID exists
+ mock_imp.assert_not_called()
+
+ @patch("prowler.lib.check.check.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.check.importlib.util.find_spec")
+ def test_resolve_check_module_raises_when_not_found(self, mock_find_spec, mock_ep):
+ """Test 17: _resolve_check_module raises ModuleNotFoundError when both fail."""
+ from prowler.lib.check.check import _resolve_check_module
+
+ mock_find_spec.return_value = None
+ mock_ep.return_value = []
+
+ with pytest.raises(ModuleNotFoundError, match="not found"):
+ _resolve_check_module("fake", "svc", "nonexistent_check")
+
+ @patch("prowler.lib.check.check.importlib.util.find_spec")
+ @patch("prowler.lib.check.check.import_check")
+ def test_resolve_check_module_surfaces_error_when_builtin_import_fails(
+ self, mock_import_check, mock_find_spec
+ ):
+ """Regression guard: when no plug-in entry-point overrides the
+ check, a built-in whose module exists but fails to import (e.g.
+ broken transitive dependency) MUST surface the real error instead
+ of being silently treated as 'not found'. See PR #10700 review
+ (HugoPBrito)."""
+ from prowler.lib.check.check import _resolve_check_module
+
+ mock_find_spec.return_value = MagicMock() # built-in module exists
+ mock_import_check.side_effect = ImportError("missing transitive dep: foo")
+
+ # No plug-in override — the built-in's import failure must propagate
+ with patch("importlib.metadata.entry_points", return_value=[]):
+ with pytest.raises(ImportError, match="missing transitive dep"):
+ _resolve_check_module("aws", "ec2", "ec2_instance_public_ip")
+
+
+# ===========================================================================
+# 5. CLI Arguments
+# ===========================================================================
+
+
+class TestCLIArguments:
+ """Tests 18-19: init_providers_parser fallback."""
+
+ @patch("prowler.providers.common.arguments.Provider._load_ep_provider")
+ @patch("prowler.providers.common.arguments.Provider.get_available_providers")
+ @patch("prowler.providers.common.arguments.import_module")
+ def test_init_providers_parser_fallback_to_init_parser(
+ self, mock_import, mock_providers, mock_load_ep
+ ):
+ """Test 18: init_providers_parser falls back to cls.init_parser for external providers."""
+ from prowler.providers.common.arguments import init_providers_parser
+
+ mock_providers.return_value = ["fakeexternal"]
+ mock_import.side_effect = ImportError("No built-in arguments module")
+ mock_load_ep.return_value = FakeExternalProvider
+
+ parser_instance = MagicMock()
+
+ # Should not raise
+ init_providers_parser(parser_instance)
+
+ @patch("prowler.providers.common.arguments.Provider._load_ep_provider")
+ @patch("prowler.providers.common.arguments.Provider.get_available_providers")
+ @patch("prowler.providers.common.arguments.import_module")
+ def test_init_providers_parser_no_crash_without_init_parser(
+ self, mock_import, mock_providers, mock_load_ep
+ ):
+ """Test 19: init_providers_parser doesn't crash if provider has no init_parser."""
+ from prowler.providers.common.arguments import init_providers_parser
+
+ mock_providers.return_value = ["nohelptext"]
+ mock_import.side_effect = ImportError("No built-in")
+ # FakeProviderNoHelpText has no init_parser
+ mock_load_ep.return_value = FakeProviderNoHelpText
+
+ parser_instance = MagicMock()
+
+ # Should not raise
+ init_providers_parser(parser_instance)
+
+ @patch("prowler.providers.common.arguments.Provider._load_ep_provider")
+ @patch("prowler.providers.common.arguments.Provider.get_available_providers")
+ @patch("prowler.providers.common.arguments.import_module")
+ def test_init_providers_parser_handles_init_parser_exception(
+ self, mock_import, mock_providers, mock_load_ep
+ ):
+ """init_providers_parser handles exception when init_parser raises."""
+ from prowler.providers.common.arguments import init_providers_parser
+
+ mock_providers.return_value = ["fakeexternal"]
+ mock_import.side_effect = ImportError("No built-in")
+
+ broken_cls = MagicMock()
+ broken_cls.init_parser.side_effect = RuntimeError("Parser init failed")
+ mock_load_ep.return_value = broken_cls
+
+ parser_instance = MagicMock()
+
+ # Should not raise
+ init_providers_parser(parser_instance)
+
+
+# ===========================================================================
+# 6. Compliance
+# ===========================================================================
+
+
+class TestCompliance:
+ """Tests 20-23: compliance discovery and loading."""
+
+ @patch("prowler.config.config.importlib.metadata.entry_points")
+ def test_get_ep_compliance_dirs_discovers_dirs(self, mock_ep):
+ """Test 20: _get_ep_compliance_dirs discovers compliance directories."""
+ from prowler.config.config import _get_ep_compliance_dirs
+
+ mock_module = MagicMock()
+ mock_module.__path__ = ["/path/to/compliance"]
+ ep = _make_entry_point("fakeexternal", "pkg.compliance", "prowler.compliance")
+ ep.load.return_value = mock_module
+ mock_ep.return_value = [ep]
+
+ dirs = _get_ep_compliance_dirs()
+
+ assert dirs["fakeexternal"] == "/path/to/compliance"
+
+ @patch("prowler.config.config.importlib.metadata.entry_points")
+ def test_get_ep_compliance_dirs_file_fallback(self, mock_ep):
+ """_get_ep_compliance_dirs uses __file__ when module has no __path__."""
+ from prowler.config.config import _get_ep_compliance_dirs
+
+ mock_module = MagicMock(spec=[])
+ mock_module.__file__ = "/path/to/compliance/__init__.py"
+ del mock_module.__path__
+ ep = _make_entry_point("ext", "pkg.compliance", "prowler.compliance")
+ ep.load.return_value = mock_module
+ mock_ep.return_value = [ep]
+
+ dirs = _get_ep_compliance_dirs()
+
+ assert dirs["ext"] == "/path/to/compliance"
+
+ @patch("prowler.config.config.importlib.metadata.entry_points")
+ def test_get_ep_compliance_dirs_handles_load_exception(self, mock_ep):
+ """_get_ep_compliance_dirs handles ep.load() exception gracefully."""
+ from prowler.config.config import _get_ep_compliance_dirs
+
+ ep = _make_entry_point("broken", "pkg.compliance", "prowler.compliance")
+ ep.load.side_effect = Exception("Load failed")
+ mock_ep.return_value = [ep]
+
+ dirs = _get_ep_compliance_dirs()
+
+ assert dirs == {}
+
+ @patch("prowler.config.config._get_ep_compliance_dirs")
+ def test_get_available_compliance_includes_external(self, mock_dirs):
+ """Test 21: get_available_compliance_frameworks includes external compliance."""
+ import json
+ import os
+ import tempfile
+
+ from prowler.config.config import get_available_compliance_frameworks
+
+ # Create a temp dir with a compliance JSON
+ with tempfile.TemporaryDirectory() as tmpdir:
+ json_path = os.path.join(tmpdir, "custom_1.0_ext.json")
+ with open(json_path, "w") as f:
+ json.dump({"Framework": "Custom", "Provider": "ext"}, f)
+
+ mock_dirs.return_value = {"ext": tmpdir}
+
+ frameworks = get_available_compliance_frameworks("ext")
+
+ assert "custom_1.0_ext" in frameworks
+
+ @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+ def test_compliance_get_bulk_loads_external(self, mock_list_modules, mock_ep):
+ """Test 22: Compliance.get_bulk loads external compliance JSON."""
+ import json
+ import os
+ import tempfile
+
+ from prowler.lib.check.compliance_models import Compliance
+
+ mock_list_modules.return_value = []
+
+ # Create a valid compliance JSON
+ with tempfile.TemporaryDirectory() as tmpdir:
+ json_data = {
+ "Framework": "Custom",
+ "Name": "Custom Framework",
+ "Version": "1.0",
+ "Provider": "fakeexternal",
+ "Description": "Test framework",
+ "Requirements": [],
+ }
+ json_path = os.path.join(tmpdir, "custom_1.0_fakeexternal.json")
+ with open(json_path, "w") as f:
+ json.dump(json_data, f)
+
+ mock_module = MagicMock()
+ mock_module.__path__ = [tmpdir]
+ ep = _make_entry_point(
+ "fakeexternal", "pkg.compliance", "prowler.compliance"
+ )
+ ep.load.return_value = mock_module
+ mock_ep.return_value = [ep]
+
+ bulk = Compliance.get_bulk("fakeexternal")
+
+ assert "custom_1.0_fakeexternal" in bulk
+ assert bulk["custom_1.0_fakeexternal"].Framework == "Custom"
+
+ @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+ def test_compliance_get_bulk_file_fallback(self, mock_list_modules, mock_ep):
+ """Compliance.get_bulk uses __file__ when external module has no __path__."""
+ import json
+ import os
+ import tempfile
+
+ from prowler.lib.check.compliance_models import Compliance
+
+ mock_list_modules.return_value = []
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ json_data = {
+ "Framework": "Custom",
+ "Name": "Custom File Fallback",
+ "Version": "1.0",
+ "Provider": "fakeexternal",
+ "Description": "Test",
+ "Requirements": [],
+ }
+ json_path = os.path.join(tmpdir, "custom_file_fakeexternal.json")
+ with open(json_path, "w") as f:
+ json.dump(json_data, f)
+
+ mock_module = MagicMock(spec=[])
+ mock_module.__file__ = os.path.join(tmpdir, "__init__.py")
+ del mock_module.__path__
+ ep = _make_entry_point(
+ "fakeexternal", "pkg.compliance", "prowler.compliance"
+ )
+ ep.load.return_value = mock_module
+ mock_ep.return_value = [ep]
+
+ bulk = Compliance.get_bulk("fakeexternal")
+
+ assert "custom_file_fakeexternal" in bulk
+
+ @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+ def test_compliance_get_bulk_handles_external_exception(
+ self, mock_list_modules, mock_ep
+ ):
+ """Compliance.get_bulk handles exception when loading external compliance."""
+ from prowler.lib.check.compliance_models import Compliance
+
+ mock_list_modules.return_value = []
+
+ ep = _make_entry_point("fakeexternal", "pkg.compliance", "prowler.compliance")
+ ep.load.side_effect = Exception("Load failed")
+ mock_ep.return_value = [ep]
+
+ bulk = Compliance.get_bulk("fakeexternal")
+
+ assert bulk == {}
+
+ @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+ def test_compliance_get_bulk_builtin_wins_on_duplicate(
+ self, mock_list_modules, mock_ep
+ ):
+ """Test 23: Compliance.get_bulk built-in wins on duplicate framework names."""
+ import json
+ import os
+ import tempfile
+
+ from prowler.lib.check.compliance_models import Compliance
+
+ mock_list_modules.return_value = []
+ mock_ep.return_value = []
+
+ # If both exist with same key, built-in (loaded first) should win
+ # Since we have no built-in modules mocked, just verify external loads
+ # The actual dedup logic: `if name not in bulk_compliance_frameworks`
+ with tempfile.TemporaryDirectory() as tmpdir:
+ json_data = {
+ "Framework": "CIS",
+ "Name": "CIS Test",
+ "Version": "1.0",
+ "Provider": "fakeexternal",
+ "Description": "Test",
+ "Requirements": [],
+ }
+ with open(os.path.join(tmpdir, "dup_framework.json"), "w") as f:
+ json.dump(json_data, f)
+
+ mock_module = MagicMock()
+ mock_module.__path__ = [tmpdir]
+ ep = _make_entry_point(
+ "fakeexternal", "pkg.compliance", "prowler.compliance"
+ )
+ ep.load.return_value = mock_module
+ mock_ep.return_value = [ep]
+
+ bulk = Compliance.get_bulk("fakeexternal")
+
+ assert "dup_framework" in bulk
+
+
+# ===========================================================================
+# 7. Parser
+# ===========================================================================
+
+
+class TestParser:
+ """Tests 24-27: parser dynamic discovery."""
+
+ @patch("prowler.lib.cli.parser.Provider.get_providers_help_text")
+ @patch("prowler.lib.cli.parser.Provider.get_available_providers")
+ def test_parser_discovers_new_providers(self, mock_providers, mock_help):
+ """Test 24: Parser discovers providers not in known_providers."""
+ from prowler.lib.cli.parser import ProwlerArgumentParser
+
+ mock_providers.return_value = [
+ "aws",
+ "azure",
+ "gcp",
+ "kubernetes",
+ "m365",
+ "github",
+ "googleworkspace",
+ "cloudflare",
+ "oraclecloud",
+ "openstack",
+ "alibabacloud",
+ "iac",
+ "llm",
+ "image",
+ "nhn",
+ "mongodbatlas",
+ "fakeexternal",
+ ]
+ mock_help.return_value = {"fakeexternal": "Fake External Provider"}
+
+ parser = ProwlerArgumentParser()
+
+ assert "fakeexternal" in parser.parser.format_usage()
+
+ @patch("prowler.lib.cli.parser.Provider.get_providers_help_text")
+ @patch("prowler.lib.cli.parser.Provider.get_available_providers")
+ def test_parser_appends_to_epilog_with_help_text(self, mock_providers, mock_help):
+ """Test 25: Parser appends new providers to epilog with _cli_help_text."""
+ from prowler.lib.cli.parser import ProwlerArgumentParser
+
+ mock_providers.return_value = [
+ "aws",
+ "azure",
+ "gcp",
+ "kubernetes",
+ "m365",
+ "github",
+ "googleworkspace",
+ "cloudflare",
+ "oraclecloud",
+ "openstack",
+ "alibabacloud",
+ "iac",
+ "llm",
+ "image",
+ "nhn",
+ "mongodbatlas",
+ "fakeexternal",
+ ]
+ mock_help.return_value = {"fakeexternal": "Fake External Provider"}
+
+ parser = ProwlerArgumentParser()
+ epilog = parser.parser.epilog
+
+ assert "fakeexternal" in epilog
+ assert "Fake External Provider" in epilog
+
+ @patch("prowler.lib.cli.parser.Provider.get_providers_help_text")
+ @patch("prowler.lib.cli.parser.Provider.get_available_providers")
+ def test_parser_skips_epilog_entry_without_help_text(
+ self, mock_providers, mock_help
+ ):
+ """Test 26: Parser doesn't add epilog entry if _cli_help_text is empty."""
+ from prowler.lib.cli.parser import ProwlerArgumentParser
+
+ mock_providers.return_value = [
+ "aws",
+ "azure",
+ "gcp",
+ "kubernetes",
+ "m365",
+ "github",
+ "googleworkspace",
+ "cloudflare",
+ "oraclecloud",
+ "openstack",
+ "alibabacloud",
+ "iac",
+ "llm",
+ "image",
+ "nhn",
+ "mongodbatlas",
+ "nohelptext",
+ ]
+ mock_help.return_value = {"nohelptext": ""}
+
+ parser = ProwlerArgumentParser()
+ epilog = parser.parser.epilog
+
+ # Should appear in usage/csv but NOT in the descriptive epilog listing
+ assert "nohelptext" in parser.parser.format_usage()
+ # No line with "nohelptext Something" in epilog
+ epilog_lines = [
+ line.strip() for line in epilog.splitlines() if "nohelptext" in line
+ ]
+ assert len(epilog_lines) == 0 or all(
+ "nohelptext" in line and line.strip() == "nohelptext" or "{" in line
+ for line in epilog_lines
+ )
+
+ @patch("prowler.lib.cli.parser.Provider.get_providers_help_text")
+ @patch("prowler.lib.cli.parser.Provider.get_available_providers")
+ def test_parser_does_not_duplicate_known_providers(self, mock_providers, mock_help):
+ """Test 27: Parser doesn't duplicate providers already in the known list."""
+ from prowler.lib.cli.parser import ProwlerArgumentParser
+
+ # No new providers
+ mock_providers.return_value = [
+ "aws",
+ "azure",
+ "gcp",
+ "kubernetes",
+ "m365",
+ "github",
+ "googleworkspace",
+ "cloudflare",
+ "oraclecloud",
+ "openstack",
+ "alibabacloud",
+ "iac",
+ "llm",
+ "image",
+ "nhn",
+ "mongodbatlas",
+ ]
+ mock_help.return_value = {}
+
+ parser = ProwlerArgumentParser()
+ usage = parser.parser.format_usage()
+
+ # aws should appear exactly once in usage
+ assert usage.count("aws") == 1
+
+
+# ===========================================================================
+# 8. Dispatch Fallbacks
+# ===========================================================================
+
+
+class TestDispatchFallbacks:
+ """Tests 28-34: all else clause fallbacks for external providers."""
+
+ def test_stdout_report_calls_get_stdout_detail(self, fake_provider):
+ """Test 28: stdout_report else clause calls provider.get_stdout_detail."""
+ from prowler.lib.outputs.outputs import stdout_report
+
+ finding = MagicMock()
+ finding.check_metadata.Provider = "fakeexternal"
+ finding.status = "FAIL"
+ finding.muted = False
+ finding.status_extended = "test"
+
+ with patch("builtins.print") as mock_print:
+ stdout_report(
+ finding, "\033[31m", True, ["FAIL"], False, provider=fake_provider
+ )
+
+ mock_print.assert_called_once()
+ printed = mock_print.call_args[0][0]
+ assert "fake-detail" in printed
+
+ def test_stdout_report_resolves_provider_when_none(self, fake_provider):
+ """stdout_report resolves provider via get_global_provider when not passed."""
+ from prowler.lib.outputs.outputs import stdout_report
+
+ finding = MagicMock()
+ finding.check_metadata.Provider = "fakeexternal"
+ finding.status = "FAIL"
+ finding.muted = False
+ finding.status_extended = "test"
+
+ with patch("builtins.print") as mock_print:
+ stdout_report(finding, "\033[31m", True, ["FAIL"], False)
+
+ mock_print.assert_called_once()
+ printed = mock_print.call_args[0][0]
+ assert "fake-detail" in printed
+
+ def test_report_propagates_provider_to_stdout_report(self):
+ """Regression guard: report() must pass its local `provider` through to
+ stdout_report so the dynamic else does not fall back to the global
+ singleton. With Provider._global cleared, the call chain still has to
+ work for an external provider — proving the argument is being used
+ instead of `Provider.get_global_provider()`. See PR #10700 review
+ (HugoPBrito)."""
+ from prowler.lib.outputs.outputs import report
+ from prowler.providers.common.provider import Provider
+
+ local_provider = FakeExternalProvider.__new__(FakeExternalProvider)
+
+ finding = MagicMock()
+ finding.status = "PASS"
+ finding.muted = False
+ finding.region = "x"
+ finding.check_metadata.Provider = "fakeexternal"
+ finding.status_extended = "test"
+
+ output_options = MagicMock()
+ output_options.verbose = True
+ output_options.status = []
+ output_options.fixer = False
+
+ # Clear the global singleton so any unintended fallback would crash.
+ previous_global = Provider._global
+ Provider._global = None
+ try:
+ with patch("builtins.print") as mock_print:
+ report([finding], local_provider, output_options)
+
+ # report() prints the finding line plus an empty separator when
+ # verbose is set; we only care that the finding was rendered using
+ # the local provider's `get_stdout_detail` ("fake-detail").
+ printed = "".join(
+ call.args[0] for call in mock_print.call_args_list if call.args
+ )
+ assert "fake-detail" in printed
+ finally:
+ Provider._global = previous_global
+
+ def test_report_sort_calls_get_finding_sort_key(self, fake_provider):
+ """Test 29: report else clause calls provider.get_finding_sort_key."""
+ from prowler.lib.outputs.outputs import report
+
+ finding1 = MagicMock()
+ finding1.status = "PASS"
+ finding1.muted = False
+ finding1.region = "b-region"
+ finding1.check_metadata.Provider = "fakeexternal"
+ finding1.status_extended = "test1"
+
+ finding2 = MagicMock()
+ finding2.status = "PASS"
+ finding2.muted = False
+ finding2.region = "a-region"
+ finding2.check_metadata.Provider = "fakeexternal"
+ finding2.status_extended = "test2"
+
+ output_options = MagicMock()
+ output_options.verbose = False
+ output_options.status = []
+
+ findings = [finding1, finding2]
+ report(findings, fake_provider, output_options)
+
+ # Should be sorted by region (get_finding_sort_key returns "region")
+ assert findings[0].region == "a-region"
+ assert findings[1].region == "b-region"
+
+ def test_display_summary_table_calls_get_summary_entity(self, fake_provider):
+ """Test 30: display_summary_table else clause calls provider.get_summary_entity."""
+ from prowler.lib.outputs.summary_table import display_summary_table
+
+ finding = MagicMock()
+ finding.status = "FAIL"
+ finding.muted = False
+ finding.check_metadata.ServiceName = "test_service"
+ finding.check_metadata.Provider = "fakeexternal"
+ finding.check_metadata.Severity = "high"
+
+ output_options = MagicMock()
+ output_options.output_directory = "/tmp"
+ output_options.output_filename = "test"
+ output_options.output_modes = []
+
+ with patch("builtins.print") as mock_print:
+ display_summary_table([finding], fake_provider, output_options)
+
+ printed_text = " ".join(str(c) for c in mock_print.call_args_list)
+ assert "Fake Host" in printed_text or "fake-host-1" in printed_text
+
+ def test_generate_output_calls_get_finding_output_data(self, fake_provider):
+ """Test 31: finding.generate_output else clause calls provider.get_finding_output_data."""
+ from prowler.lib.check.models import (
+ CheckMetadata,
+ Code,
+ Recommendation,
+ Remediation,
+ )
+ from prowler.lib.outputs.finding import Finding
+
+ metadata = CheckMetadata(
+ Provider="fakeexternal",
+ CheckID="test_check",
+ CheckTitle="Test check title",
+ CheckType=[],
+ ServiceName="test",
+ SubServiceName="",
+ ResourceIdTemplate="",
+ Severity="high",
+ ResourceType="Test",
+ ResourceGroup="",
+ Description="Test description",
+ Risk="Test risk",
+ RelatedUrl="",
+ Remediation=Remediation(
+ Code=Code(CLI="", NativeIaC="", Other="", Terraform=""),
+ Recommendation=Recommendation(
+ Text="Fix it", Url="https://hub.prowler.com/check/test_check"
+ ),
+ ),
+ Categories=[],
+ DependsOn=[],
+ RelatedTo=[],
+ Notes="",
+ )
+
+ check_output = MagicMock()
+ check_output.check_metadata = metadata
+ check_output.status = "FAIL"
+ check_output.status_extended = "test failed"
+ check_output.muted = False
+ check_output.resource = {}
+ check_output.resource_details = ""
+ check_output.resource_tags = {}
+ check_output.compliance = {}
+
+ output_options = MagicMock()
+ output_options.unix_timestamp = False
+ output_options.bulk_checks_metadata = {}
+
+ finding = Finding.generate_output(fake_provider, check_output, output_options)
+
+ assert finding.auth_method == "fake"
+ assert finding.account_uid == "fake-account"
+ assert finding.resource_name == "fake-resource"
+ assert finding.region == "local"
+
+ def test_output_options_calls_get_output_options(self, fake_provider):
+ """Test 32: __main__.py else clause calls provider.get_output_options."""
+ result = fake_provider.get_output_options(MagicMock(), {})
+
+ assert result is not None
+ assert hasattr(result, "output_directory")
+
+ def test_html_assessment_calls_get_html_assessment_summary(self, fake_provider):
+ """Test 33: html.py fallback calls provider.get_html_assessment_summary."""
+ from prowler.lib.outputs.html.html import HTML
+
+ result = HTML.get_assessment_summary(fake_provider)
+
+ assert "Fake Assessment
" in result
+
+ def test_compliance_output_calls_generate_compliance_output(self, fake_provider):
+ """Test 34: __main__.py else clause calls provider.generate_compliance_output."""
+ generated_outputs = {"compliance": []}
+
+ fake_provider.generate_compliance_output(
+ [],
+ {},
+ set(),
+ MagicMock(),
+ generated_outputs,
+ )
+
+ assert "fake-compliance-output" in generated_outputs["compliance"]
+
+
+# ===========================================================================
+# 9. Base Contract Defaults
+# ===========================================================================
+
+
+class TestBaseContractDefaults:
+ """Tests for Provider base class default implementations."""
+
+ def test_from_cli_args_raises_not_implemented(self):
+ """Base Provider.from_cli_args raises NotImplementedError."""
+ with pytest.raises(NotImplementedError):
+ FakeProviderNoHelpText.from_cli_args(MagicMock(), {})
+
+ def test_get_output_options_raises_not_implemented(self):
+ """Base Provider.get_output_options raises NotImplementedError."""
+ provider = FakeProviderNoHelpText()
+ with pytest.raises(NotImplementedError):
+ provider.get_output_options(MagicMock(), {})
+
+ def test_get_stdout_detail_raises_not_implemented(self):
+ """Base Provider.get_stdout_detail raises NotImplementedError."""
+ provider = FakeProviderNoHelpText()
+ with pytest.raises(NotImplementedError):
+ provider.get_stdout_detail(MagicMock())
+
+ def test_get_finding_sort_key_returns_none(self):
+ """Base Provider.get_finding_sort_key returns None."""
+ provider = FakeProviderNoHelpText()
+ assert provider.get_finding_sort_key() is None
+
+ def test_get_summary_entity_raises_not_implemented(self):
+ """Base Provider.get_summary_entity raises NotImplementedError."""
+ provider = FakeProviderNoHelpText()
+ with pytest.raises(NotImplementedError):
+ provider.get_summary_entity()
+
+ def test_get_finding_output_data_raises_not_implemented(self):
+ """Base Provider.get_finding_output_data raises NotImplementedError."""
+ provider = FakeProviderNoHelpText()
+ with pytest.raises(NotImplementedError):
+ provider.get_finding_output_data(MagicMock())
+
+ def test_get_html_assessment_summary_raises_not_implemented(self):
+ """Base Provider.get_html_assessment_summary raises NotImplementedError."""
+ provider = FakeProviderNoHelpText()
+ with pytest.raises(NotImplementedError):
+ provider.get_html_assessment_summary()
+
+ def test_generate_compliance_output_raises_not_implemented(self):
+ """Base Provider.generate_compliance_output raises NotImplementedError."""
+ provider = FakeProviderNoHelpText()
+ with pytest.raises(NotImplementedError):
+ provider.generate_compliance_output([], {}, set(), MagicMock(), {})
+
+ def test_get_mutelist_finding_args_raises_not_implemented(self):
+ """Base Provider.get_mutelist_finding_args raises NotImplementedError."""
+ provider = FakeProviderNoHelpText()
+ with pytest.raises(NotImplementedError):
+ provider.get_mutelist_finding_args()
+
+ def test_display_compliance_table_raises_not_implemented(self):
+ """Base Provider.display_compliance_table raises NotImplementedError."""
+ provider = FakeProviderNoHelpText()
+ with pytest.raises(NotImplementedError):
+ provider.display_compliance_table([], {}, "fw", "out", "/tmp", False)
+
+ def test_is_external_tool_provider_defaults_to_false(self):
+ """Base Provider.is_external_tool_provider returns False."""
+ provider = FakeProviderNoHelpText()
+ assert provider.is_external_tool_provider is False
+
+
+# ===========================================================================
+# 10. Mutelist Dispatch for External Providers
+# ===========================================================================
+
+
+class TestMutelistDispatch:
+ """Tests for mutelist integration with external providers."""
+
+ def test_get_mutelist_finding_args_returns_identity(self, fake_provider):
+ """External provider returns identity kwargs for mutelist."""
+ args = fake_provider.get_mutelist_finding_args()
+
+ assert args == {"host_id": "fake-host-1"}
+
+ def test_mutelist_dispatch_calls_external_provider(self, fake_provider):
+ """execute() uses get_mutelist_finding_args for unknown provider types."""
+ from prowler.lib.check.check import execute
+
+ # Create a mock check that returns one finding
+ finding = MagicMock()
+ finding.status = "FAIL"
+ finding.muted = False
+ finding.check_metadata.Provider = "fakeexternal"
+
+ check = MagicMock()
+ check.execute.return_value = [finding]
+ check.CheckID = "fake_check"
+ check.ServiceName = "fake_service"
+ check.Severity.value = "high"
+
+ # Setup mutelist on the provider
+ fake_provider.mutelist = MagicMock()
+ fake_provider.mutelist.mutelist = {"Accounts": {}}
+ fake_provider.mutelist.is_finding_muted.return_value = True
+
+ output_options = MagicMock()
+ output_options.status = []
+ output_options.unix_timestamp = False
+
+ execute(check, fake_provider, None, output_options)
+
+ # is_finding_muted should have been called with host_id + finding
+ fake_provider.mutelist.is_finding_muted.assert_called_once_with(
+ host_id="fake-host-1", finding=finding
+ )
+
+
+# ===========================================================================
+# 11. Compliance Table Dispatch for External Providers
+# ===========================================================================
+
+
+class TestComplianceTableDispatch:
+ """Tests for compliance table display with external providers."""
+
+ def test_display_compliance_table_delegates_to_provider(self, fake_provider):
+ """display_compliance_table uses provider method for unknown frameworks."""
+ from prowler.lib.outputs.compliance.compliance import (
+ display_compliance_table,
+ )
+
+ fake_provider.display_compliance_table = MagicMock(return_value=True)
+
+ display_compliance_table(
+ [], {}, "custom_1.0_fakeexternal", "out", "/tmp", False
+ )
+
+ fake_provider.display_compliance_table.assert_called_once_with(
+ [],
+ {},
+ "custom_1.0_fakeexternal",
+ "out",
+ "/tmp",
+ False,
+ )
+
+ def test_display_compliance_table_falls_back_to_generic(self, fake_provider):
+ """display_compliance_table falls back to generic when provider returns False."""
+ from prowler.lib.outputs.compliance.compliance import (
+ display_compliance_table,
+ )
+
+ fake_provider.display_compliance_table = MagicMock(return_value=False)
+
+ with patch(
+ "prowler.lib.outputs.compliance.compliance.get_generic_compliance_table"
+ ) as mock_generic:
+ display_compliance_table(
+ [], {}, "custom_1.0_fakeexternal", "out", "/tmp", False
+ )
+
+ mock_generic.assert_called_once()
+
+ def test_display_compliance_table_falls_back_on_not_implemented(self):
+ """display_compliance_table falls back to generic when NotImplementedError."""
+ # Use a provider that doesn't implement display_compliance_table
+ provider = FakeProviderNoHelpText()
+ Provider.set_global_provider(provider)
+
+ with patch(
+ "prowler.lib.outputs.compliance.compliance.get_generic_compliance_table"
+ ) as mock_generic:
+ from prowler.lib.outputs.compliance.compliance import (
+ display_compliance_table,
+ )
+
+ display_compliance_table(
+ [], {}, "unknown_1.0_nohelptext", "out", "/tmp", False
+ )
+
+ mock_generic.assert_called_once()
+ Provider._global = None