feat: Enhance dynamic provider loading and compliance framework discovery

- Implemented dynamic loading of external providers via entry points, allowing for greater flexibility in provider integration.
  - Added functionality to discover compliance directories from entry points, enabling external compliance frameworks to be loaded seamlessly.
  - Refactored check module resolution to prioritize built-in checks while falling back to entry points if necessary.
  - Improved compliance framework loading to include both built-in and external sources, ensuring comprehensive compliance coverage.
  - Enhanced CLI argument parsing to support external providers, improving user experience and configurability.
  - Introduced extensive unit tests to validate dynamic loading, compliance discovery, and overall integration of external providers.
This commit is contained in:
StylusFrost
2026-04-15 13:22:57 +02:00
parent 62f114f5d0
commit f8333baf24
15 changed files with 1356 additions and 78 deletions
+29 -2
View File
@@ -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":
+38 -3
View File
@@ -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
+31 -6
View File
@@ -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()
+35 -5
View File
@@ -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}")
+4 -3
View File
@@ -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 = (
+48 -18
View File
@@ -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)
+44 -5
View File
@@ -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
""",
+5
View File
@@ -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
+7 -5
View File
@@ -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}"
+29 -19
View File
@@ -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
+3
View File
@@ -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):
+10
View File
@@ -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}"
+135 -12
View File
@@ -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):
View File
@@ -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 "<div>Fake Assessment</div>"
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 "<div>Fake Assessment</div>" 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"]