diff --git a/prowler/__main__.py b/prowler/__main__.py
index b46a8fa579..a0e94a885f 100644
--- a/prowler/__main__.py
+++ b/prowler/__main__.py
@@ -69,11 +69,11 @@ from prowler.lib.outputs.compliance.cis.cis_gcp import GCPCIS
from prowler.lib.outputs.compliance.cis.cis_github import GithubCIS
from prowler.lib.outputs.compliance.cis.cis_googleworkspace import GoogleWorkspaceCIS
from prowler.lib.outputs.compliance.cis.cis_kubernetes import KubernetesCIS
+from prowler.lib.outputs.compliance.cis.cis_m365 import M365CIS
+from prowler.lib.outputs.compliance.cis.cis_oraclecloud import OracleCloudCIS
from prowler.lib.outputs.compliance.cisa_scuba.cisa_scuba_googleworkspace import (
GoogleWorkspaceCISASCuBA,
)
-from prowler.lib.outputs.compliance.cis.cis_m365 import M365CIS
-from prowler.lib.outputs.compliance.cis.cis_oraclecloud import OracleCloudCIS
from prowler.lib.outputs.compliance.compliance import display_compliance_table
from prowler.lib.outputs.compliance.csa.csa_alibabacloud import AlibabaCloudCSA
from prowler.lib.outputs.compliance.csa.csa_aws import AWSCSA
@@ -405,6 +405,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:
@@ -1282,6 +1285,30 @@ def prowler():
)
generated_outputs["compliance"].append(generic_compliance)
generic_compliance.batch_write_data_to_file()
+ else:
+ # Dynamic fallback: any external/custom provider
+ try:
+ global_provider.generate_compliance_output(
+ finding_outputs,
+ bulk_compliance_frameworks,
+ input_compliance_frameworks,
+ output_options,
+ generated_outputs,
+ )
+ except NotImplementedError:
+ # Last resort: generic compliance
+ for compliance_name in input_compliance_frameworks:
+ filename = (
+ f"{output_options.output_directory}/compliance/"
+ f"{output_options.output_filename}_{compliance_name}.csv"
+ )
+ generic_compliance = GenericCompliance(
+ findings=finding_outputs,
+ compliance=bulk_compliance_frameworks[compliance_name],
+ file_path=filename,
+ )
+ generated_outputs["compliance"].append(generic_compliance)
+ generic_compliance.batch_write_data_to_file()
# AWS Security Hub Integration
if provider == "aws":
diff --git a/prowler/config/config.py b/prowler/config/config.py
index debea713aa..42aa2374b3 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
@@ -76,13 +77,36 @@ EXTERNAL_TOOL_PROVIDERS = frozenset({"iac", "llm", "image"})
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:
+ pass
+ 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 provider in providers:
- compliance_dir = f"{actual_directory}/../compliance/{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:
@@ -91,6 +115,17 @@ def get_available_compliance_frameworks(provider=None):
available_compliance_frameworks.append(
file.name.removesuffix(".json")
)
+ # External compliance via entry points
+ ep_dirs = _get_ep_compliance_dirs()
+ for prov, path in ep_dirs.items():
+ if provider and prov != provider:
+ continue
+ if os.path.isdir(path):
+ for file in os.scandir(path):
+ if file.is_file() and file.name.endswith(".json"):
+ available_compliance_frameworks.append(
+ file.name.removesuffix(".json")
+ )
return available_compliance_frameworks
diff --git a/prowler/lib/check/check.py b/prowler/lib/check/check.py
index 8e365acedc..8bb50d8871 100644
--- a/prowler/lib/check/check.py
+++ b/prowler/lib/check/check.py
@@ -362,6 +362,29 @@ 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 — tries built-in path first, then entry points."""
+ # Built-in path
+ builtin_path = f"prowler.providers.{provider_type}.services.{service}.{check_name}.{check_name}"
+ try:
+ return import_check(builtin_path)
+ except ModuleNotFoundError:
+ pass
+
+ # Entry point lookup
+ import importlib.metadata
+
+ 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
@@ -502,9 +525,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()
@@ -582,9 +606,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()
diff --git a/prowler/lib/check/compliance_models.py b/prowler/lib/check/compliance_models.py
index d1f3b8c35d..b17bccebe2 100644
--- a/prowler/lib/check/compliance_models.py
+++ b/prowler/lib/check/compliance_models.py
@@ -1,3 +1,4 @@
+import importlib.metadata
import os
import sys
from enum import Enum
@@ -390,26 +391,55 @@ class Compliance(BaseModel):
"""Bulk load all compliance frameworks specification into a dict"""
try:
bulk_compliance_frameworks = {}
+ # Built-in compliance from prowler/compliance/{provider}/
available_compliance_framework_modules = list_compliance_modules()
for compliance_framework in available_compliance_framework_modules:
if provider in compliance_framework.name:
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 e:
+ logger.warning(
+ f"Failed to load external compliance for '{ep.name}': {e}"
+ )
+
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 43d6018cbf..26fa884027 100644
--- a/prowler/lib/check/models.py
+++ b/prowler/lib/check/models.py
@@ -11,10 +11,11 @@ from typing import Any, Dict, Optional, Set
from pydantic.v1 import BaseModel, Field, ValidationError, validator
from pydantic.v1.error_wrappers import ErrorWrapper
-from prowler.config.config import EXTERNAL_TOOL_PROVIDERS, Provider
+from prowler.config.config import EXTERNAL_TOOL_PROVIDERS
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(
@@ -466,7 +467,7 @@ class CheckMetadata(BaseModel):
# If the bulk checks metadata is not provided, get it
if not bulk_checks_metadata:
bulk_checks_metadata = {}
- available_providers = [p.value for p in Provider]
+ available_providers = ProviderABC.get_available_providers()
for provider_name in available_providers:
bulk_checks_metadata.update(CheckMetadata.get_bulk(provider_name))
if provider:
@@ -491,7 +492,7 @@ class CheckMetadata(BaseModel):
# Loaded here, as it is not always needed
if not bulk_compliance_frameworks:
bulk_compliance_frameworks = {}
- available_providers = [p.value for p in Provider]
+ available_providers = ProviderABC.get_available_providers()
for provider in available_providers:
bulk_compliance_frameworks = Compliance.get_bulk(provider=provider)
checks_from_compliance_framework = (
diff --git a/prowler/lib/check/utils.py b/prowler/lib/check/utils.py
index 45f16cde32..b81d76aa76 100644
--- a/prowler/lib/check/utils.py
+++ b/prowler/lib/check/utils.py
@@ -1,10 +1,30 @@
import importlib
+import importlib.metadata
+import os
import sys
from pkgutil import walk_packages
from prowler.lib.logger import logger
+def _recover_ep_checks(provider: str) -> list[tuple]:
+ """Discover external checks registered via entry points for a provider.
+
+ 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:
+ 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 e:
+ logger.warning(f"Failed to discover external check '{ep.name}': {e}")
+ return checks
+
+
def recover_checks_from_provider(
provider: str, service: str = None, include_fixers: bool = False
) -> list[tuple]:
@@ -19,24 +39,34 @@ def recover_checks_from_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
+ try:
+ 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)
+ except ModuleNotFoundError:
+ if service:
+ logger.critical(
+ f"Service {service} was not found for the {provider} provider."
+ )
+ sys.exit(1)
+ # No built-in services for this provider (e.g., external provider)
+
+ # External checks registered via entry points
+ if not service:
+ checks.extend(_recover_ep_checks(provider))
+
except Exception as e:
logger.critical(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}")
sys.exit(1)
diff --git a/prowler/lib/cli/parser.py b/prowler/lib/cli/parser.py
index d2afa70e15..aa270829f8 100644
--- a/prowler/lib/cli/parser.py
+++ b/prowler/lib/cli/parser.py
@@ -19,19 +19,58 @@ from prowler.providers.common.arguments import (
validate_asff_usage,
validate_provider_arguments,
)
+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} ...",
- epilog="""
+ usage=f"prowler [-h] [--version] {{aws,azure,gcp,kubernetes,m365,github,googleworkspace,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel,dashboard,iac,image{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,iac,llm,image,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel{extra_providers_csv}}}
aws AWS Provider
azure Azure Provider
gcp GCP Provider
@@ -48,13 +87,13 @@ Available Cloud Providers:
image Container Image Provider
nhn NHN Provider (Unofficial)
mongodbatlas MongoDB Atlas Provider (Beta)
- 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
""",
diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py
index 26fbe327ee..e367518fd6 100644
--- a/prowler/lib/outputs/finding.py
+++ b/prowler/lib/outputs/finding.py
@@ -470,6 +470,11 @@ class Finding(BaseModel):
check_output, "fixed_version", ""
)
+ else:
+ # Dynamic fallback: any external/custom provider
+ provider_data = provider.get_finding_output_data(check_output)
+ output_data.update(provider_data)
+
# check_output Unique ID
# TODO: move this to a function
# TODO: in Azure, GCP and K8s there are findings without resource_name
diff --git a/prowler/lib/outputs/html/html.py b/prowler/lib/outputs/html/html.py
index d547ea3b95..1cd097815b 100644
--- a/prowler/lib/outputs/html/html.py
+++ b/prowler/lib/outputs/html/html.py
@@ -1417,11 +1417,13 @@ class HTML(Output):
# Azure_provider --> azure
# Kubernetes_provider --> kubernetes
- # Dynamically get the Provider quick inventory handler
- provider_html_assessment_summary_function = (
- f"get_{provider.type}_assessment_summary"
- )
- return getattr(HTML, provider_html_assessment_summary_function)(provider)
+ # Try static method first, fall back to provider method
+ method_name = f"get_{provider.type}_assessment_summary"
+ if hasattr(HTML, method_name):
+ return getattr(HTML, method_name)(provider)
+ else:
+ # Dynamic fallback: any external/custom provider
+ return provider.get_html_assessment_summary()
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
diff --git a/prowler/lib/outputs/outputs.py b/prowler/lib/outputs/outputs.py
index 1e2f6f2058..9f21d2e3ee 100644
--- a/prowler/lib/outputs/outputs.py
+++ b/prowler/lib/outputs/outputs.py
@@ -7,39 +7,46 @@ from prowler.lib.outputs.common import Status
from prowler.lib.outputs.finding import Finding
-def stdout_report(finding, color, verbose, status, fix):
+def stdout_report(finding, color, verbose, status, fix, provider=None):
if finding.check_metadata.Provider == "aws":
details = finding.region
- if finding.check_metadata.Provider == "azure":
+ elif finding.check_metadata.Provider == "azure":
details = finding.location
- if finding.check_metadata.Provider == "gcp":
+ elif finding.check_metadata.Provider == "gcp":
details = finding.location.lower()
- if finding.check_metadata.Provider == "kubernetes":
+ elif finding.check_metadata.Provider == "kubernetes":
details = finding.namespace.lower()
- if finding.check_metadata.Provider == "github":
+ elif finding.check_metadata.Provider == "github":
details = finding.owner
- if finding.check_metadata.Provider == "m365":
+ elif finding.check_metadata.Provider == "m365":
details = finding.location
- if finding.check_metadata.Provider == "mongodbatlas":
+ elif finding.check_metadata.Provider == "mongodbatlas":
details = finding.location
- if finding.check_metadata.Provider == "nhn":
+ elif finding.check_metadata.Provider == "nhn":
details = finding.location
- if finding.check_metadata.Provider == "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
diff --git a/prowler/lib/outputs/summary_table.py b/prowler/lib/outputs/summary_table.py
index 635139117d..973bf8e0ed 100644
--- a/prowler/lib/outputs/summary_table.py
+++ b/prowler/lib/outputs/summary_table.py
@@ -107,6 +107,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/providers/common/arguments.py b/prowler/providers/common/arguments.py
index 727568a68f..a2c923e111 100644
--- a/prowler/providers/common/arguments.py
+++ b/prowler/providers/common/arguments.py
@@ -23,6 +23,16 @@ def init_providers_parser(self):
),
init_provider_arguments_function,
)(self)
+ except ImportError:
+ # External provider — try 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"Failed to init parser for external provider '{provider}': {error}"
+ )
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py
index debc1c4ef2..b5eaa00299 100644
--- a/prowler/providers/common/provider.py
+++ b/prowler/providers/common/provider.py
@@ -1,4 +1,5 @@
import importlib
+import importlib.metadata
import pkgutil
import sys
from abc import ABC, abstractmethod
@@ -135,6 +136,69 @@ class Provider(ABC):
"""
return set()
+ # --- Dynamic provider contract methods (not @abstractmethod for incremental migration) ---
+
+ _cli_help_text: str = ""
+
+ @classmethod
+ def from_cli_args(cls, arguments: Namespace, fixer_config: dict) -> "Provider":
+ """Instantiate the provider from CLI arguments."""
+ 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()"
+ )
+
+ @property
+ def is_external_tool_provider(self) -> bool:
+ """True for providers that delegate scanning to an external tool."""
+ return False
+
+ # --- End dynamic provider contract methods ---
+
@staticmethod
def get_global_provider() -> "Provider":
return Provider._global
@@ -146,13 +210,21 @@ class Provider(ABC):
@staticmethod
def init_global_provider(arguments: Namespace) -> None:
try:
- provider_class_path = (
- f"{providers_path}.{arguments.provider}.{arguments.provider}_provider"
- )
- provider_class_name = f"{arguments.provider.capitalize()}Provider"
- provider_class = getattr(
- import_module(provider_class_path), provider_class_name
- )
+ # Try built-in provider first, fall back to entry point
+ provider_class = 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
+ )
+ except (ImportError, AttributeError):
+ # External provider — load via entry point
+ 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"
+ )
fixer_config = load_and_validate_config_file(
arguments.provider, arguments.fixer_config
@@ -378,6 +450,9 @@ class Provider(ABC):
mutelist_path=arguments.mutelist_file,
fixer_config=fixer_config,
)
+ else:
+ # Dynamic fallback: any external/custom provider
+ provider_class.from_cli_args(arguments, fixer_config)
except TypeError as error:
logger.critical(
@@ -390,17 +465,65 @@ class Provider(ABC):
)
sys.exit(1)
+ # Cache for entry-point provider classes {name: class}
+ _ep_providers: dict = {}
+
@staticmethod
def get_available_providers() -> list[str]:
"""get_available_providers returns a list of the available providers"""
- providers = []
- # Dynamically import the package based on its string path
+ providers = set()
+ # Built-in providers from local package
prowler_providers = importlib.import_module(providers_path)
- # Iterate over all modules found in the prowler_providers package
for _, provider, ispkg in pkgutil.iter_modules(prowler_providers.__path__):
if provider != "common" and ispkg:
- providers.append(provider)
- return providers
+ providers.add(provider)
+ # External providers registered via entry points
+ for ep in importlib.metadata.entry_points(group="prowler.providers"):
+ providers.add(ep.name)
+ return sorted(providers)
+
+ @staticmethod
+ def _load_ep_provider(name: str):
+ """Load an external provider class from entry points, with 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:
+ pass
+ 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:
+ help_text[name] = ""
+ return help_text
@staticmethod
def update_provider_config(audit_config: dict, variable: str, value: str):
diff --git a/tests/providers/external/__init__.py b/tests/providers/external/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
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 0000000000..fb56134d33
--- /dev/null
+++ b/tests/providers/external/test_dynamic_provider_loading.py
@@ -0,0 +1,938 @@
+"""
+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_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 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.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"] == ""
+
+
+# ===========================================================================
+# 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)
+
+
+# ===========================================================================
+# 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.list_modules")
+ def test_recover_checks_handles_external_provider_without_services(
+ self, mock_list_modules, mock_ep_checks
+ ):
+ """Test 13: recover_checks_from_provider doesn't crash for external providers."""
+ from prowler.lib.check.utils import recover_checks_from_provider
+
+ mock_list_modules.side_effect = ModuleNotFoundError("No services")
+ 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")
+ def test_recover_checks_combines_builtin_and_entry_points(
+ self, 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
+
+ # 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
+
+
+# ===========================================================================
+# 4. Check Execution
+# ===========================================================================
+
+
+class TestCheckExecution:
+ """Tests 15-17: _resolve_check_module."""
+
+ @patch("prowler.lib.check.check.import_check")
+ def test_resolve_check_module_builtin_first(self, mock_import):
+ """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
+
+ 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.import_check")
+ def test_resolve_check_module_fallback_to_entry_point(self, mock_import_check):
+ """Test 16: _resolve_check_module falls back to entry point."""
+ from prowler.lib.check.check import _resolve_check_module
+
+ mock_import_check.side_effect = ModuleNotFoundError("Not built-in")
+
+ 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")
+
+ @patch("prowler.lib.check.check.importlib.metadata.entry_points")
+ @patch("prowler.lib.check.check.import_check")
+ def test_resolve_check_module_raises_when_not_found(self, mock_import, mock_ep):
+ """Test 17: _resolve_check_module raises ModuleNotFoundError when both fail."""
+ from prowler.lib.check.check import _resolve_check_module
+
+ mock_import.side_effect = ModuleNotFoundError("Not built-in")
+ mock_ep.return_value = []
+
+ with pytest.raises(ModuleNotFoundError, match="not found"):
+ _resolve_check_module("fake", "svc", "nonexistent_check")
+
+
+# ===========================================================================
+# 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)
+
+
+# ===========================================================================
+# 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._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_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_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"]