mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
refactor(output_options): remove output options from provider (#5149)
This commit is contained in:
Pedro Martín
+62
-41
@@ -69,8 +69,12 @@ from prowler.lib.outputs.slack.slack import Slack
|
||||
from prowler.lib.outputs.summary_table import display_summary_table
|
||||
from prowler.providers.aws.lib.s3.s3 import S3
|
||||
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
|
||||
from prowler.providers.aws.models import AWSOutputOptions
|
||||
from prowler.providers.azure.models import AzureOutputOptions
|
||||
from prowler.providers.common.provider import Provider
|
||||
from prowler.providers.common.quick_inventory import run_provider_quick_inventory
|
||||
from prowler.providers.gcp.models import GCPOutputOptions
|
||||
from prowler.providers.kubernetes.models import KubernetesOutputOptions
|
||||
|
||||
|
||||
def prowler():
|
||||
@@ -236,7 +240,22 @@ def prowler():
|
||||
global_provider.mutelist = args.mutelist_file
|
||||
|
||||
# Setup Output Options
|
||||
global_provider.output_options = (args, bulk_checks_metadata)
|
||||
if provider == "aws":
|
||||
output_options = AWSOutputOptions(
|
||||
args, bulk_checks_metadata, global_provider.identity
|
||||
)
|
||||
elif provider == "azure":
|
||||
output_options = AzureOutputOptions(
|
||||
args, bulk_checks_metadata, global_provider.identity
|
||||
)
|
||||
elif provider == "gcp":
|
||||
output_options = GCPOutputOptions(
|
||||
args, bulk_checks_metadata, global_provider.identity
|
||||
)
|
||||
elif provider == "kubernetes":
|
||||
output_options = KubernetesOutputOptions(
|
||||
args, bulk_checks_metadata, global_provider.identity
|
||||
)
|
||||
|
||||
# Run the quick inventory for the provider if available
|
||||
if hasattr(args, "quick_inventory") and args.quick_inventory:
|
||||
@@ -252,6 +271,7 @@ def prowler():
|
||||
global_provider,
|
||||
custom_checks_metadata,
|
||||
args.config_file,
|
||||
output_options,
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
@@ -259,7 +279,7 @@ def prowler():
|
||||
)
|
||||
|
||||
# Prowler Fixer
|
||||
if global_provider.output_options.fixer:
|
||||
if output_options.fixer:
|
||||
print(f"{Style.BRIGHT}\nRunning Prowler Fixer, please wait...{Style.RESET_ALL}")
|
||||
# Check if there are any FAIL findings
|
||||
if any("FAIL" in finding.status for finding in findings):
|
||||
@@ -305,7 +325,8 @@ def prowler():
|
||||
# TODO: this part is needed since the checks generates a Check_Report_XXX and the output uses Finding
|
||||
# This will be refactored for the outputs generate directly the Finding
|
||||
finding_outputs = [
|
||||
Finding.generate_output(global_provider, finding) for finding in findings
|
||||
Finding.generate_output(global_provider, finding, output_options)
|
||||
for finding in findings
|
||||
]
|
||||
|
||||
generated_outputs = {"regular": [], "compliance": []}
|
||||
@@ -313,8 +334,8 @@ def prowler():
|
||||
if args.output_formats:
|
||||
for mode in args.output_formats:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/"
|
||||
f"{global_provider.output_options.output_filename}"
|
||||
f"{output_options.output_directory}/"
|
||||
f"{output_options.output_filename}"
|
||||
)
|
||||
if mode == "csv":
|
||||
csv_output = CSV(
|
||||
@@ -356,16 +377,16 @@ def prowler():
|
||||
)
|
||||
|
||||
# Compliance Frameworks
|
||||
input_compliance_frameworks = set(
|
||||
global_provider.output_options.output_modes
|
||||
).intersection(get_available_compliance_frameworks(provider))
|
||||
input_compliance_frameworks = set(output_options.output_modes).intersection(
|
||||
get_available_compliance_frameworks(provider)
|
||||
)
|
||||
if provider == "aws":
|
||||
for compliance_name in input_compliance_frameworks:
|
||||
if compliance_name.startswith("cis_"):
|
||||
# Generate CIS Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
cis = AWSCIS(
|
||||
findings=finding_outputs,
|
||||
@@ -378,8 +399,8 @@ def prowler():
|
||||
elif compliance_name == "mitre_attack_aws":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack = AWSMitreAttack(
|
||||
findings=finding_outputs,
|
||||
@@ -392,8 +413,8 @@ def prowler():
|
||||
elif compliance_name.startswith("ens_"):
|
||||
# Generate ENS Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
ens = AWSENS(
|
||||
findings=finding_outputs,
|
||||
@@ -406,8 +427,8 @@ def prowler():
|
||||
elif compliance_name.startswith("aws_well_architected_framework"):
|
||||
# Generate AWS Well-Architected Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
aws_well_architected = AWSWellArchitected(
|
||||
findings=finding_outputs,
|
||||
@@ -420,8 +441,8 @@ def prowler():
|
||||
elif compliance_name.startswith("iso27001_"):
|
||||
# Generate ISO27001 Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
iso27001 = AWSISO27001(
|
||||
findings=finding_outputs,
|
||||
@@ -447,8 +468,8 @@ def prowler():
|
||||
kisa_ismsp.batch_write_data_to_file()
|
||||
else:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
generic_compliance = GenericCompliance(
|
||||
findings=finding_outputs,
|
||||
@@ -464,8 +485,8 @@ def prowler():
|
||||
if compliance_name.startswith("cis_"):
|
||||
# Generate CIS Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
cis = AzureCIS(
|
||||
findings=finding_outputs,
|
||||
@@ -478,8 +499,8 @@ def prowler():
|
||||
elif compliance_name == "mitre_attack_azure":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack = AzureMitreAttack(
|
||||
findings=finding_outputs,
|
||||
@@ -491,8 +512,8 @@ def prowler():
|
||||
mitre_attack.batch_write_data_to_file()
|
||||
else:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
generic_compliance = GenericCompliance(
|
||||
findings=finding_outputs,
|
||||
@@ -508,8 +529,8 @@ def prowler():
|
||||
if compliance_name.startswith("cis_"):
|
||||
# Generate CIS Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
cis = GCPCIS(
|
||||
findings=finding_outputs,
|
||||
@@ -522,8 +543,8 @@ def prowler():
|
||||
elif compliance_name == "mitre_attack_gcp":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack = GCPMitreAttack(
|
||||
findings=finding_outputs,
|
||||
@@ -535,8 +556,8 @@ def prowler():
|
||||
mitre_attack.batch_write_data_to_file()
|
||||
else:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
generic_compliance = GenericCompliance(
|
||||
findings=finding_outputs,
|
||||
@@ -552,8 +573,8 @@ def prowler():
|
||||
if compliance_name.startswith("cis_"):
|
||||
# Generate CIS Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
cis = KubernetesCIS(
|
||||
findings=finding_outputs,
|
||||
@@ -565,8 +586,8 @@ def prowler():
|
||||
cis.batch_write_data_to_file()
|
||||
else:
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
f"{output_options.output_directory}/compliance/"
|
||||
f"{output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
generic_compliance = GenericCompliance(
|
||||
findings=finding_outputs,
|
||||
@@ -609,7 +630,7 @@ def prowler():
|
||||
aws_partition=global_provider.identity.partition,
|
||||
aws_session=global_provider.session.current_session,
|
||||
findings=asff_output.data,
|
||||
send_only_fails=global_provider.output_options.send_sh_only_fails,
|
||||
send_only_fails=output_options.send_sh_only_fails,
|
||||
aws_security_hub_available_regions=security_hub_regions,
|
||||
)
|
||||
# Send the findings to Security Hub
|
||||
@@ -635,7 +656,7 @@ def prowler():
|
||||
display_summary_table(
|
||||
findings,
|
||||
global_provider,
|
||||
global_provider.output_options,
|
||||
output_options,
|
||||
)
|
||||
# Only display compliance table if there are findings (not all MANUAL) and it is a default execution
|
||||
if (
|
||||
@@ -654,13 +675,13 @@ def prowler():
|
||||
findings,
|
||||
bulk_checks_metadata,
|
||||
compliance,
|
||||
global_provider.output_options.output_filename,
|
||||
global_provider.output_options.output_directory,
|
||||
output_options.output_filename,
|
||||
output_options.output_directory,
|
||||
compliance_overview,
|
||||
)
|
||||
if compliance_overview:
|
||||
print(
|
||||
f"\nDetailed compliance results are in {Fore.YELLOW}{global_provider.output_options.output_directory}/compliance/{Style.RESET_ALL}\n"
|
||||
f"\nDetailed compliance results are in {Fore.YELLOW}{output_options.output_directory}/compliance/{Style.RESET_ALL}\n"
|
||||
)
|
||||
|
||||
# If custom checks were passed, remove the modules
|
||||
|
||||
+34
-15
@@ -435,6 +435,7 @@ def execute_checks(
|
||||
global_provider: Any,
|
||||
custom_checks_metadata: Any,
|
||||
config_file: str,
|
||||
output_options: Any,
|
||||
) -> list:
|
||||
# List to store all the check's findings
|
||||
all_findings = []
|
||||
@@ -471,7 +472,7 @@ def execute_checks(
|
||||
)
|
||||
|
||||
# Execution with the --only-logs flag
|
||||
if global_provider.output_options.only_logs:
|
||||
if output_options.only_logs:
|
||||
for check_name in checks_to_execute:
|
||||
# Recover service from check name
|
||||
service = check_name.split("_")[0]
|
||||
@@ -481,6 +482,7 @@ def execute_checks(
|
||||
check_name,
|
||||
global_provider,
|
||||
custom_checks_metadata,
|
||||
output_options,
|
||||
)
|
||||
all_findings.extend(check_findings)
|
||||
|
||||
@@ -544,6 +546,7 @@ def execute_checks(
|
||||
check_name,
|
||||
global_provider,
|
||||
custom_checks_metadata,
|
||||
output_options,
|
||||
)
|
||||
all_findings.extend(check_findings)
|
||||
services_executed.add(service)
|
||||
@@ -575,7 +578,21 @@ def execute(
|
||||
check_name: str,
|
||||
global_provider: Any,
|
||||
custom_checks_metadata: Any,
|
||||
output_options: Any = None,
|
||||
):
|
||||
"""
|
||||
Execute the check and report the findings
|
||||
|
||||
Args:
|
||||
service (str): service name
|
||||
check_name (str): check name
|
||||
global_provider (Any): provider object
|
||||
custom_checks_metadata (Any): custom checks metadata
|
||||
output_options (Any): output options, depending on the provider
|
||||
|
||||
Returns:
|
||||
list: list of findings
|
||||
"""
|
||||
try:
|
||||
# Import check module
|
||||
check_module_path = f"prowler.providers.{global_provider.type}.services.{service}.{check_name}.{check_name}"
|
||||
@@ -593,20 +610,24 @@ def execute(
|
||||
)
|
||||
|
||||
# Run check
|
||||
verbose = (
|
||||
global_provider.output_options.verbose
|
||||
or global_provider.output_options.fixer
|
||||
)
|
||||
check_findings = run_check(
|
||||
check_class, verbose, global_provider.output_options.only_logs
|
||||
)
|
||||
verbose = False
|
||||
if hasattr(output_options, "verbose"):
|
||||
verbose = output_options.verbose
|
||||
elif hasattr(output_options, "fixer"):
|
||||
verbose = output_options.fixer
|
||||
|
||||
only_logs = False
|
||||
if hasattr(output_options, "only_logs"):
|
||||
only_logs = output_options.only_logs
|
||||
|
||||
check_findings = run_check(check_class, verbose, only_logs)
|
||||
|
||||
# Exclude findings per status
|
||||
if global_provider.output_options.status:
|
||||
if hasattr(output_options, "status") and output_options.status:
|
||||
check_findings = [
|
||||
finding
|
||||
for finding in check_findings
|
||||
if finding.status in global_provider.output_options.status
|
||||
if finding.status in output_options.status
|
||||
]
|
||||
|
||||
# Mutelist findings
|
||||
@@ -628,7 +649,7 @@ def execute(
|
||||
|
||||
# Refactor(Outputs)
|
||||
# Report the check's findings
|
||||
report(check_findings, global_provider)
|
||||
report(check_findings, global_provider, output_options)
|
||||
|
||||
# Refactor(Outputs)
|
||||
if os.environ.get("PROWLER_REPORT_LIB_PATH"):
|
||||
@@ -638,10 +659,8 @@ def execute(
|
||||
outputs_module = importlib.import_module(lib)
|
||||
custom_report_interface = getattr(outputs_module, "report")
|
||||
|
||||
# TODO: review this call and see if we can remove the global_provider.output_options since it is contained in the global_provider
|
||||
custom_report_interface(
|
||||
check_findings, global_provider.output_options, global_provider
|
||||
)
|
||||
# TODO: review this call and see if we can remove the output_options since it is contained in the global_provider
|
||||
custom_report_interface(check_findings, output_options, global_provider)
|
||||
except Exception:
|
||||
sys.exit(1)
|
||||
except ModuleNotFoundError:
|
||||
|
||||
@@ -88,29 +88,37 @@ class Finding(BaseModel):
|
||||
|
||||
@classmethod
|
||||
def generate_output(
|
||||
cls, provider: Provider, check_output: Check_Report
|
||||
cls, provider: Provider, check_output: Check_Report, output_options
|
||||
) -> "Finding":
|
||||
"""Generates the output for a finding based on the provider and output options
|
||||
|
||||
Args:
|
||||
provider (Provider): the provider object
|
||||
check_output (Check_Report): the check output object
|
||||
output_options: the output options object, depending on the provider
|
||||
Returns:
|
||||
finding_output (Finding): the finding output object
|
||||
|
||||
"""
|
||||
output_options = provider.output_options
|
||||
# TODO: think about get_provider_data_mapping
|
||||
provider_data_mapping = get_provider_data_mapping(provider)
|
||||
|
||||
# TODO: move fill_common_finding_data
|
||||
common_finding_data = fill_common_finding_data(
|
||||
check_output, output_options.unix_timestamp
|
||||
)
|
||||
unix_timestamp = False
|
||||
if hasattr(output_options, "unix_timestamp"):
|
||||
unix_timestamp = output_options.unix_timestamp
|
||||
|
||||
common_finding_data = fill_common_finding_data(check_output, unix_timestamp)
|
||||
output_data = {}
|
||||
output_data.update(provider_data_mapping)
|
||||
output_data.update(common_finding_data)
|
||||
|
||||
bulk_checks_metadata = {}
|
||||
if hasattr(output_options, "bulk_checks_metadata"):
|
||||
bulk_checks_metadata = output_options.bulk_checks_metadata
|
||||
|
||||
output_data["compliance"] = get_check_compliance(
|
||||
check_output, provider.type, output_options.bulk_checks_metadata
|
||||
check_output, provider.type, bulk_checks_metadata
|
||||
)
|
||||
try:
|
||||
if provider.type == "aws":
|
||||
|
||||
@@ -25,10 +25,12 @@ def stdout_report(finding, color, verbose, status, fix):
|
||||
)
|
||||
|
||||
|
||||
# TODO: Only pass check_findings, provider.output_options and provider.type
|
||||
def report(check_findings, provider):
|
||||
# TODO: Only pass check_findings, output_options and provider.type
|
||||
def report(check_findings, provider, output_options):
|
||||
try:
|
||||
output_options = provider.output_options
|
||||
verbose = False
|
||||
if hasattr(output_options, "verbose"):
|
||||
verbose = output_options.verbose
|
||||
if check_findings:
|
||||
# TO-DO Generic Function
|
||||
if provider.type == "aws":
|
||||
@@ -39,21 +41,27 @@ def report(check_findings, provider):
|
||||
|
||||
for finding in check_findings:
|
||||
# Print findings by stdout
|
||||
status = []
|
||||
if hasattr(output_options, "status"):
|
||||
status = output_options.status
|
||||
fixer = False
|
||||
if hasattr(output_options, "fixer"):
|
||||
fixer = output_options.fixer
|
||||
color = set_report_color(finding.status, finding.muted)
|
||||
stdout_report(
|
||||
finding,
|
||||
color,
|
||||
output_options.verbose,
|
||||
output_options.status,
|
||||
output_options.fixer,
|
||||
verbose,
|
||||
status,
|
||||
fixer,
|
||||
)
|
||||
|
||||
else: # No service resources in the whole account
|
||||
color = set_report_color("MANUAL")
|
||||
if output_options.verbose:
|
||||
if verbose:
|
||||
print(f"\t{color}INFO{Style.RESET_ALL} There are no resources")
|
||||
# Separator between findings and bar
|
||||
if output_options.verbose:
|
||||
if verbose:
|
||||
print()
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
|
||||
@@ -106,6 +106,7 @@ class Scan:
|
||||
check_name,
|
||||
self._provider,
|
||||
custom_checks_metadata,
|
||||
output_options=None,
|
||||
)
|
||||
|
||||
# Store findings
|
||||
@@ -131,7 +132,9 @@ class Scan:
|
||||
)
|
||||
|
||||
findings = [
|
||||
Finding.generate_output(self._provider, finding)
|
||||
Finding.generate_output(
|
||||
self._provider, finding, output_options=None
|
||||
)
|
||||
for finding in check_findings
|
||||
]
|
||||
|
||||
|
||||
@@ -52,7 +52,6 @@ from prowler.providers.aws.models import (
|
||||
AWSIdentityInfo,
|
||||
AWSMFAInfo,
|
||||
AWSOrganizationsInfo,
|
||||
AWSOutputOptions,
|
||||
AWSSession,
|
||||
)
|
||||
from prowler.providers.common.models import Audit_Metadata, Connection
|
||||
@@ -68,7 +67,6 @@ class AwsProvider(Provider):
|
||||
_audit_config: dict
|
||||
_scan_unused_services: bool = False
|
||||
_enabled_regions: set = set()
|
||||
_output_options: AWSOutputOptions
|
||||
# TODO: this is not optional, enforce for all providers
|
||||
audit_metadata: Audit_Metadata
|
||||
|
||||
@@ -304,17 +302,6 @@ class AwsProvider(Provider):
|
||||
def fixer_config(self):
|
||||
return self._fixer_config
|
||||
|
||||
@property
|
||||
def output_options(self):
|
||||
return self._output_options
|
||||
|
||||
@output_options.setter
|
||||
def output_options(self, options: tuple):
|
||||
arguments, bulk_checks_metadata = options
|
||||
self._output_options = AWSOutputOptions(
|
||||
arguments, bulk_checks_metadata, self._identity
|
||||
)
|
||||
|
||||
@property
|
||||
def mutelist(self) -> AWSMutelist:
|
||||
"""
|
||||
|
||||
@@ -29,11 +29,7 @@ from prowler.providers.azure.exceptions.exceptions import (
|
||||
from prowler.providers.azure.lib.arguments.arguments import validate_azure_region
|
||||
from prowler.providers.azure.lib.mutelist.mutelist import AzureMutelist
|
||||
from prowler.providers.azure.lib.regions.regions import get_regions_config
|
||||
from prowler.providers.azure.models import (
|
||||
AzureIdentityInfo,
|
||||
AzureOutputOptions,
|
||||
AzureRegionConfig,
|
||||
)
|
||||
from prowler.providers.azure.models import AzureIdentityInfo, AzureRegionConfig
|
||||
from prowler.providers.common.models import Audit_Metadata, Connection
|
||||
from prowler.providers.common.provider import Provider
|
||||
|
||||
@@ -53,7 +49,6 @@ class AzureProvider(Provider):
|
||||
_audit_config (dict): The audit configuration for the Azure provider.
|
||||
_region_config (AzureRegionConfig): The region configuration for the Azure provider.
|
||||
_locations (dict): A dictionary containing the available locations for the Azure provider.
|
||||
_output_options (AzureOutputOptions): The output options for the Azure provider.
|
||||
_mutelist (AzureMutelist): The mutelist object associated with the Azure provider.
|
||||
audit_metadata (Audit_Metadata): The audit metadata for the Azure provider.
|
||||
|
||||
@@ -81,7 +76,6 @@ class AzureProvider(Provider):
|
||||
_audit_config: dict
|
||||
_region_config: AzureRegionConfig
|
||||
_locations: dict
|
||||
_output_options: AzureOutputOptions
|
||||
_mutelist: AzureMutelist
|
||||
# TODO: this is not optional, enforce for all providers
|
||||
audit_metadata: Audit_Metadata
|
||||
@@ -197,28 +191,6 @@ class AzureProvider(Provider):
|
||||
"""Returns the fixer configuration."""
|
||||
return self._fixer_config
|
||||
|
||||
@property
|
||||
def output_options(self):
|
||||
"""Returns the output options for the Azure provider."""
|
||||
return self._output_options
|
||||
|
||||
@output_options.setter
|
||||
def output_options(self, options: tuple):
|
||||
"""Set output options for the Azure provider.
|
||||
|
||||
Sets the output options for the Azure provider using the provided arguments and bulk checks metadata.
|
||||
|
||||
Args:
|
||||
options (tuple): A tuple containing the arguments and bulk checks metadata.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
arguments, bulk_checks_metadata = options
|
||||
self._output_options = AzureOutputOptions(
|
||||
arguments, bulk_checks_metadata, self._identity
|
||||
)
|
||||
|
||||
@property
|
||||
def mutelist(self) -> AzureMutelist:
|
||||
"""Mutelist object associated with this Azure provider."""
|
||||
|
||||
@@ -37,7 +37,6 @@ class Provider(ABC):
|
||||
identity (property): The identity of the provider for auditing.
|
||||
session (property): The session of the provider for auditing.
|
||||
audit_config (property): The audit configuration of the provider.
|
||||
output_options (property): The output configuration of the provider for auditing.
|
||||
|
||||
Methods:
|
||||
print_credentials(): Displays the provider's credentials used for auditing in the command-line interface.
|
||||
@@ -109,26 +108,6 @@ class Provider(ABC):
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def output_options(self) -> str:
|
||||
"""
|
||||
output_options method returns the provider's audit output configuration.
|
||||
|
||||
This method needs to be created in each provider.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
@output_options.setter
|
||||
@abstractmethod
|
||||
def output_options(self, value: str) -> Any:
|
||||
"""
|
||||
output_options.setter sets the provider's audit output configuration.
|
||||
|
||||
This method needs to be created in each provider.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
@abstractmethod
|
||||
def get_output_mapping(self) -> dict:
|
||||
"""
|
||||
|
||||
@@ -23,12 +23,7 @@ from prowler.providers.gcp.exceptions.exceptions import (
|
||||
GCPTestConnectionError,
|
||||
)
|
||||
from prowler.providers.gcp.lib.mutelist.mutelist import GCPMutelist
|
||||
from prowler.providers.gcp.models import (
|
||||
GCPIdentityInfo,
|
||||
GCPOrganization,
|
||||
GCPOutputOptions,
|
||||
GCPProject,
|
||||
)
|
||||
from prowler.providers.gcp.models import GCPIdentityInfo, GCPOrganization, GCPProject
|
||||
|
||||
|
||||
class GcpProvider(Provider):
|
||||
@@ -38,7 +33,6 @@ class GcpProvider(Provider):
|
||||
_excluded_project_ids: list
|
||||
_identity: GCPIdentityInfo
|
||||
_audit_config: dict
|
||||
_output_options: GCPOutputOptions
|
||||
_mutelist: GCPMutelist
|
||||
# TODO: this is not optional, enforce for all providers
|
||||
audit_metadata: Audit_Metadata
|
||||
@@ -178,17 +172,6 @@ class GcpProvider(Provider):
|
||||
def fixer_config(self):
|
||||
return self._fixer_config
|
||||
|
||||
@property
|
||||
def output_options(self):
|
||||
return self._output_options
|
||||
|
||||
@output_options.setter
|
||||
def output_options(self, options: tuple):
|
||||
arguments, bulk_checks_metadata = options
|
||||
self._output_options = GCPOutputOptions(
|
||||
arguments, bulk_checks_metadata, self._identity
|
||||
)
|
||||
|
||||
@property
|
||||
def mutelist(self) -> GCPMutelist:
|
||||
"""
|
||||
|
||||
@@ -21,7 +21,6 @@ from prowler.providers.kubernetes.exceptions.exceptions import (
|
||||
from prowler.providers.kubernetes.lib.mutelist.mutelist import KubernetesMutelist
|
||||
from prowler.providers.kubernetes.models import (
|
||||
KubernetesIdentityInfo,
|
||||
KubernetesOutputOptions,
|
||||
KubernetesSession,
|
||||
)
|
||||
|
||||
@@ -32,7 +31,6 @@ class KubernetesProvider(Provider):
|
||||
_namespaces: list
|
||||
_audit_config: dict
|
||||
_identity: KubernetesIdentityInfo
|
||||
_output_options: KubernetesOutputOptions
|
||||
_mutelist: dict
|
||||
# TODO: this is not optional, enforce for all providers
|
||||
audit_metadata: Audit_Metadata
|
||||
@@ -106,17 +104,6 @@ class KubernetesProvider(Provider):
|
||||
def fixer_config(self):
|
||||
return self._fixer_config
|
||||
|
||||
@property
|
||||
def output_options(self):
|
||||
return self._output_options
|
||||
|
||||
@output_options.setter
|
||||
def output_options(self, options: tuple):
|
||||
arguments, bulk_checks_metadata = options
|
||||
self._output_options = KubernetesOutputOptions(
|
||||
arguments, bulk_checks_metadata, self._identity
|
||||
)
|
||||
|
||||
@property
|
||||
def mutelist(self) -> KubernetesMutelist:
|
||||
"""
|
||||
|
||||
@@ -826,6 +826,7 @@ class TestCheck:
|
||||
expected_checks=["accessanalyzer_enabled"]
|
||||
),
|
||||
custom_checks_metadata=None,
|
||||
output_options=None,
|
||||
)
|
||||
assert len(findings) == 1
|
||||
|
||||
@@ -843,6 +844,8 @@ class TestCheck:
|
||||
)
|
||||
]
|
||||
status = ["PASS"]
|
||||
output_options = mock.MagicMock()
|
||||
output_options.status = status
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.accessanalyzer.accessanalyzer_service.AccessAnalyzer",
|
||||
accessanalyzer_client,
|
||||
@@ -851,9 +854,10 @@ class TestCheck:
|
||||
service="accessanalyzer",
|
||||
check_name="accessanalyzer_enabled",
|
||||
global_provider=set_mocked_aws_provider(
|
||||
status=status, expected_checks=["accessanalyzer_enabled"]
|
||||
expected_checks=["accessanalyzer_enabled"]
|
||||
),
|
||||
custom_checks_metadata=None,
|
||||
output_options=output_options,
|
||||
)
|
||||
assert len(findings) == 0
|
||||
|
||||
|
||||
@@ -215,10 +215,8 @@ class TestFinding:
|
||||
check_output.region = "us-west-1"
|
||||
output_options = MagicMock()
|
||||
output_options.unix_timestamp = 1234567890
|
||||
global_provider = MagicMock()
|
||||
global_provider.output_options = output_options
|
||||
# Call the method under test
|
||||
finding_output = Finding.generate_output(provider, check_output)
|
||||
finding_output = Finding.generate_output(provider, check_output, output_options)
|
||||
|
||||
# Assertions to verify expected behavior
|
||||
assert finding_output is not None
|
||||
@@ -293,10 +291,8 @@ class TestFinding:
|
||||
check_output.region = "us-west-1"
|
||||
output_options = MagicMock()
|
||||
output_options.unix_timestamp = 1234567890
|
||||
global_provider = MagicMock()
|
||||
global_provider.output_options = output_options
|
||||
# Call the method under test
|
||||
finding_output = Finding.generate_output(provider, check_output)
|
||||
finding_output = Finding.generate_output(provider, check_output, output_options)
|
||||
|
||||
# Assertions to verify expected behavior
|
||||
assert finding_output is not None
|
||||
@@ -367,7 +363,7 @@ class TestFinding:
|
||||
output_options = MagicMock()
|
||||
output_options.unix_timestamp = 1234567890
|
||||
|
||||
finding_output = Finding.generate_output(provider, check_output)
|
||||
finding_output = Finding.generate_output(provider, check_output, output_options)
|
||||
|
||||
assert finding_output is not None
|
||||
assert finding_output.auth_method == "Principal: mock_auth"
|
||||
@@ -438,7 +434,7 @@ class TestFinding:
|
||||
output_options = MagicMock()
|
||||
output_options.unix_timestamp = 1234567890
|
||||
|
||||
finding_output = Finding.generate_output(provider, check_output)
|
||||
finding_output = Finding.generate_output(provider, check_output, output_options)
|
||||
|
||||
assert finding_output is not None
|
||||
assert finding_output.auth_method == "in-cluster"
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
from argparse import Namespace
|
||||
from datetime import datetime
|
||||
from unittest import mock
|
||||
|
||||
from freezegun import freeze_time
|
||||
|
||||
from prowler.config.config import output_file_timestamp
|
||||
from prowler.providers.aws.models import AWSOutputOptions
|
||||
from prowler.providers.azure.models import AzureOutputOptions
|
||||
from prowler.providers.gcp.models import GCPOutputOptions
|
||||
from prowler.providers.kubernetes.models import KubernetesOutputOptions
|
||||
|
||||
|
||||
class Test_Output_Options:
|
||||
@freeze_time(datetime.today())
|
||||
def test_set_output_options_aws_no_output_filename(self):
|
||||
arguments = Namespace()
|
||||
arguments.status = ["FAIL"]
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
arguments.verbose = True
|
||||
arguments.security_hub = True
|
||||
arguments.shodan = None
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.send_sh_only_fails = True
|
||||
|
||||
identity = mock.MagicMock()
|
||||
identity.account = "123456789012"
|
||||
|
||||
output_options = AWSOutputOptions(arguments, {}, identity)
|
||||
|
||||
assert output_options.status == ["FAIL"]
|
||||
assert output_options.output_modes == ["csv", "json-asff"]
|
||||
assert output_options.output_directory == "output_test_directory"
|
||||
assert output_options.verbose
|
||||
assert output_options.security_hub_enabled
|
||||
assert not output_options.shodan_api_key
|
||||
assert not output_options.only_logs
|
||||
assert not output_options.unix_timestamp
|
||||
assert output_options.send_sh_only_fails
|
||||
assert (
|
||||
output_options.output_filename
|
||||
== f"prowler-output-{identity.account}-{output_file_timestamp}"
|
||||
)
|
||||
assert output_options.bulk_checks_metadata == {}
|
||||
|
||||
@freeze_time(datetime.today())
|
||||
def test_set_output_options_aws(self):
|
||||
arguments = Namespace()
|
||||
arguments.status = []
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
arguments.verbose = True
|
||||
arguments.output_filename = "output_test_filename"
|
||||
arguments.security_hub = True
|
||||
arguments.shodan = None
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.send_sh_only_fails = True
|
||||
|
||||
identity = mock.MagicMock()
|
||||
identity.account = "123456789012"
|
||||
|
||||
output_options = AWSOutputOptions(arguments, {}, identity)
|
||||
|
||||
assert isinstance(output_options, AWSOutputOptions)
|
||||
assert output_options.security_hub_enabled
|
||||
assert output_options.send_sh_only_fails
|
||||
assert output_options.status == []
|
||||
assert output_options.output_modes == ["csv", "json-asff"]
|
||||
assert output_options.output_directory == arguments.output_directory
|
||||
assert output_options.bulk_checks_metadata == {}
|
||||
assert output_options.verbose
|
||||
assert output_options.output_filename == arguments.output_filename
|
||||
|
||||
@freeze_time(datetime.today())
|
||||
def test_azure_provider_output_options_with_domain(self):
|
||||
arguments = Namespace()
|
||||
# Output Options
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
output_directory = arguments.output_directory
|
||||
arguments.status = []
|
||||
arguments.verbose = True
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.shodan = None
|
||||
|
||||
identity = mock.MagicMock()
|
||||
identity.tenant_domain = "test-domain"
|
||||
|
||||
output_options = AzureOutputOptions(
|
||||
arguments,
|
||||
{},
|
||||
identity,
|
||||
)
|
||||
|
||||
assert isinstance(output_options, AzureOutputOptions)
|
||||
assert output_options.status == []
|
||||
assert output_options.output_modes == [
|
||||
"csv",
|
||||
]
|
||||
assert output_options.output_directory == output_directory
|
||||
assert output_options.bulk_checks_metadata == {}
|
||||
assert output_options.verbose
|
||||
assert (
|
||||
output_options.output_filename
|
||||
== f"prowler-output-{identity.tenant_domain}-{output_file_timestamp}"
|
||||
)
|
||||
|
||||
@freeze_time(datetime.today())
|
||||
def test_gcp_output_options(self):
|
||||
arguments = Namespace()
|
||||
# Output options
|
||||
arguments.status = []
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
arguments.verbose = True
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.shodan = None
|
||||
|
||||
identity = mock.MagicMock()
|
||||
identity.profile = "test-profile"
|
||||
|
||||
output_optionss = GCPOutputOptions(
|
||||
arguments,
|
||||
{},
|
||||
identity,
|
||||
)
|
||||
|
||||
assert isinstance(output_optionss, GCPOutputOptions)
|
||||
assert output_optionss.status == []
|
||||
assert output_optionss.output_modes == [
|
||||
"csv",
|
||||
]
|
||||
assert output_optionss.output_directory == arguments.output_directory
|
||||
assert output_optionss.bulk_checks_metadata == {}
|
||||
assert output_optionss.verbose
|
||||
assert f"prowler-output-{identity.profile}" in output_optionss.output_filename
|
||||
|
||||
def test_set_output_options_kubernetes(self):
|
||||
arguments = Namespace()
|
||||
arguments.status = []
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
arguments.verbose = True
|
||||
arguments.output_filename = "output_test_filename"
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.shodan = None
|
||||
|
||||
identity = mock.MagicMock()
|
||||
identity.context = "test-context"
|
||||
|
||||
output_options = KubernetesOutputOptions(
|
||||
arguments,
|
||||
{},
|
||||
identity,
|
||||
)
|
||||
|
||||
assert isinstance(output_options, KubernetesOutputOptions)
|
||||
assert output_options.status == []
|
||||
assert output_options.output_modes == ["csv"]
|
||||
assert output_options.output_directory == arguments.output_directory
|
||||
assert output_options.bulk_checks_metadata == {}
|
||||
assert output_options.verbose
|
||||
assert output_options.output_filename == arguments.output_filename
|
||||
@@ -466,12 +466,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "aws"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_1, finding_2]
|
||||
@@ -509,12 +508,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "aws"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_1, finding_2]
|
||||
@@ -552,12 +550,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "aws"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_1, finding_2]
|
||||
@@ -595,12 +592,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "aws"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_1, finding_2]
|
||||
@@ -640,12 +636,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "azure"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -685,12 +680,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "azure"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -730,12 +724,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "azure"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -775,12 +768,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "azure"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -818,12 +810,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "gcp"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -861,12 +852,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "gcp"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -904,12 +894,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "gcp"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -947,12 +936,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "gcp"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -990,12 +978,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "kubernetes"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -1033,12 +1020,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "kubernetes"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -1076,12 +1062,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "kubernetes"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -1119,12 +1104,11 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "kubernetes"
|
||||
provider.output_options = output_options
|
||||
|
||||
# Assertions
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
# Call the report method
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
check_findings_sorted = [finding_2, finding_1]
|
||||
@@ -1148,10 +1132,9 @@ class TestOutputs:
|
||||
|
||||
provider = MagicMock()
|
||||
provider.type = "azure"
|
||||
provider.output_options = output_options
|
||||
|
||||
with mock.patch("builtins.print") as mocked_print:
|
||||
report(check_findings, provider)
|
||||
report(check_findings, provider, output_options)
|
||||
|
||||
# Assertions
|
||||
mocked_print.assert_any_call(
|
||||
|
||||
@@ -64,7 +64,7 @@ def mock_generate_output():
|
||||
with mock.patch(
|
||||
"prowler.lib.outputs.finding.Finding.generate_output", autospec=True
|
||||
) as mock_gen_output:
|
||||
mock_gen_output.side_effect = lambda provider, finding: finding
|
||||
mock_gen_output.side_effect = lambda provider, finding, output_options: finding
|
||||
yield mock_gen_output
|
||||
|
||||
|
||||
|
||||
@@ -2,17 +2,14 @@ import json
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
from argparse import Namespace
|
||||
from datetime import datetime, timedelta
|
||||
from json import dumps
|
||||
from os import rmdir
|
||||
from re import search
|
||||
from unittest import mock
|
||||
|
||||
import botocore
|
||||
import botocore.exceptions
|
||||
from boto3 import client, resource, session
|
||||
from freezegun import freeze_time
|
||||
from mock import patch
|
||||
from moto import mock_aws
|
||||
from pytest import raises
|
||||
@@ -42,7 +39,6 @@ from prowler.providers.aws.models import (
|
||||
AWSCredentials,
|
||||
AWSMFAInfo,
|
||||
AWSOrganizationsInfo,
|
||||
AWSOutputOptions,
|
||||
)
|
||||
from prowler.providers.common.models import Connection
|
||||
from prowler.providers.common.provider import Provider
|
||||
@@ -1038,97 +1034,6 @@ aws:
|
||||
|
||||
assert aws_provider.audit_resources == [AWS_ACCOUNT_ARN]
|
||||
|
||||
@mock_aws
|
||||
@freeze_time(datetime.today())
|
||||
def test_set_provider_output_options_aws_no_output_filename(self):
|
||||
arguments = Namespace()
|
||||
arguments.status = ["FAIL"]
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
arguments.verbose = True
|
||||
arguments.security_hub = True
|
||||
arguments.shodan = "test-api-key"
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.send_sh_only_fails = True
|
||||
|
||||
aws_provider = AwsProvider()
|
||||
# This is needed since the output_options requires to get the global provider to get the audit config
|
||||
with patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
):
|
||||
|
||||
aws_provider.output_options = arguments, {}
|
||||
|
||||
assert isinstance(aws_provider.output_options, AWSOutputOptions)
|
||||
assert aws_provider.output_options.security_hub_enabled
|
||||
assert aws_provider.output_options.send_sh_only_fails
|
||||
assert aws_provider.output_options.status == ["FAIL"]
|
||||
assert aws_provider.output_options.output_modes == ["csv", "json-asff"]
|
||||
assert (
|
||||
aws_provider.output_options.output_directory
|
||||
== arguments.output_directory
|
||||
)
|
||||
assert aws_provider.output_options.bulk_checks_metadata == {}
|
||||
assert aws_provider.output_options.verbose
|
||||
assert (
|
||||
f"prowler-output-{AWS_ACCOUNT_NUMBER}"
|
||||
in aws_provider.output_options.output_filename
|
||||
)
|
||||
# Flaky due to the millisecond part of the timestamp
|
||||
# assert (
|
||||
# aws_provider.output_options.output_filename
|
||||
# == f"prowler-output-{AWS_ACCOUNT_NUMBER}-{datetime.today().strftime('%Y%m%d%H%M%S')}"
|
||||
# )
|
||||
|
||||
# Delete testing directory
|
||||
rmdir(f"{arguments.output_directory}/compliance")
|
||||
rmdir(arguments.output_directory)
|
||||
|
||||
@mock_aws
|
||||
@freeze_time(datetime.today())
|
||||
def test_set_provider_output_options_aws(self):
|
||||
arguments = Namespace()
|
||||
arguments.status = []
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
arguments.verbose = True
|
||||
arguments.output_filename = "output_test_filename"
|
||||
arguments.security_hub = True
|
||||
arguments.shodan = "test-api-key"
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.send_sh_only_fails = True
|
||||
|
||||
aws_provider = AwsProvider()
|
||||
# This is needed since the output_options requires to get the global provider to get the audit config
|
||||
with patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
):
|
||||
|
||||
aws_provider.output_options = arguments, {}
|
||||
|
||||
assert isinstance(aws_provider.output_options, AWSOutputOptions)
|
||||
assert aws_provider.output_options.security_hub_enabled
|
||||
assert aws_provider.output_options.send_sh_only_fails
|
||||
assert aws_provider.output_options.status == []
|
||||
assert aws_provider.output_options.output_modes == ["csv", "json-asff"]
|
||||
assert (
|
||||
aws_provider.output_options.output_directory
|
||||
== arguments.output_directory
|
||||
)
|
||||
assert aws_provider.output_options.bulk_checks_metadata == {}
|
||||
assert aws_provider.output_options.verbose
|
||||
assert (
|
||||
aws_provider.output_options.output_filename == arguments.output_filename
|
||||
)
|
||||
|
||||
# Delete testing directory
|
||||
rmdir(f"{arguments.output_directory}/compliance")
|
||||
rmdir(arguments.output_directory)
|
||||
|
||||
@mock_aws
|
||||
def test_validate_credentials_commercial_partition_with_regions(self):
|
||||
# Create a mock IAM user
|
||||
|
||||
@@ -120,13 +120,6 @@ def set_mocked_aws_provider(
|
||||
# AWS Provider
|
||||
provider = AwsProvider()
|
||||
|
||||
# Set output options
|
||||
provider.output_options = arguments, {}
|
||||
|
||||
# Output options
|
||||
|
||||
provider.output_options = arguments, {}
|
||||
|
||||
# Mock Session
|
||||
provider._session.session_config = None
|
||||
provider._session.original_session = original_session
|
||||
|
||||
@@ -1,13 +1,9 @@
|
||||
from argparse import Namespace
|
||||
from datetime import datetime
|
||||
from os import rmdir
|
||||
from unittest.mock import patch
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from azure.core.credentials import AccessToken
|
||||
from azure.identity import DefaultAzureCredential
|
||||
from freezegun import freeze_time
|
||||
from mock import MagicMock
|
||||
|
||||
from prowler.config.config import (
|
||||
@@ -22,11 +18,7 @@ from prowler.providers.azure.exceptions.exceptions import (
|
||||
AzureNoAuthenticationMethodError,
|
||||
AzureTenantIDNoBrowserAuthError,
|
||||
)
|
||||
from prowler.providers.azure.models import (
|
||||
AzureIdentityInfo,
|
||||
AzureOutputOptions,
|
||||
AzureRegionConfig,
|
||||
)
|
||||
from prowler.providers.azure.models import AzureIdentityInfo, AzureRegionConfig
|
||||
from prowler.providers.common.models import Connection
|
||||
|
||||
|
||||
@@ -203,87 +195,6 @@ class TestAzureProvider:
|
||||
== "[1919] Azure Tenant ID (--tenant-id) is required for browser authentication mode"
|
||||
)
|
||||
|
||||
@freeze_time(datetime.today())
|
||||
def test_azure_provider_output_options_with_domain(self):
|
||||
arguments = Namespace()
|
||||
subscription_id = None
|
||||
tenant_id = None
|
||||
|
||||
# We need to set exactly one auth method
|
||||
az_cli_auth = None
|
||||
sp_env_auth = True
|
||||
browser_auth = None
|
||||
managed_identity_auth = None
|
||||
|
||||
config_file = default_config_file_path
|
||||
fixer_config = default_fixer_config_file_path
|
||||
azure_region = "AzureCloud"
|
||||
|
||||
# Output Options
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
output_directory = arguments.output_directory
|
||||
arguments.status = []
|
||||
arguments.verbose = True
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.shodan = "test-api-key"
|
||||
|
||||
tenant_domain = "test-domain"
|
||||
with patch(
|
||||
"prowler.providers.azure.azure_provider.AzureProvider.setup_identity",
|
||||
return_value=AzureIdentityInfo(tenant_domain=tenant_domain),
|
||||
), patch(
|
||||
"prowler.providers.azure.azure_provider.AzureProvider.get_locations",
|
||||
return_value={},
|
||||
), patch(
|
||||
"prowler.providers.azure.azure_provider.AzureProvider.setup_session",
|
||||
return_value=DefaultAzureCredential(),
|
||||
):
|
||||
azure_provider = AzureProvider(
|
||||
az_cli_auth,
|
||||
sp_env_auth,
|
||||
browser_auth,
|
||||
managed_identity_auth,
|
||||
tenant_id,
|
||||
azure_region,
|
||||
subscription_id,
|
||||
config_file,
|
||||
fixer_config,
|
||||
)
|
||||
# This is needed since the output_options requires to get the global provider to get the audit config
|
||||
with patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=azure_provider,
|
||||
):
|
||||
|
||||
azure_provider.output_options = arguments, {}
|
||||
|
||||
assert isinstance(azure_provider.output_options, AzureOutputOptions)
|
||||
assert azure_provider.output_options.status == []
|
||||
assert azure_provider.output_options.output_modes == [
|
||||
"csv",
|
||||
]
|
||||
assert (
|
||||
azure_provider.output_options.output_directory == output_directory
|
||||
)
|
||||
assert azure_provider.output_options.bulk_checks_metadata == {}
|
||||
assert azure_provider.output_options.verbose
|
||||
# Flaky due to the millisecond part of the timestamp
|
||||
# assert (
|
||||
# azure_provider.output_options.output_filename
|
||||
# == f"prowler-output-{azure_provider.identity.tenant_domain}-{datetime.today().strftime('%Y%m%d%H%M%S')}"
|
||||
# )
|
||||
assert (
|
||||
f"prowler-output-{azure_provider.identity.tenant_domain}"
|
||||
in azure_provider.output_options.output_filename
|
||||
)
|
||||
|
||||
# Delete testing directory
|
||||
# TODO: move this to a fixtures file
|
||||
rmdir(f"{arguments.output_directory}/compliance")
|
||||
rmdir(arguments.output_directory)
|
||||
|
||||
def test_test_connection_browser_auth(self):
|
||||
with patch(
|
||||
"prowler.providers.azure.azure_provider.DefaultAzureCredential"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from argparse import Namespace
|
||||
from datetime import datetime
|
||||
from os import environ, rmdir
|
||||
from os import environ
|
||||
|
||||
from freezegun import freeze_time
|
||||
from mock import MagicMock, patch
|
||||
@@ -11,8 +11,7 @@ from prowler.config.config import (
|
||||
load_and_validate_config_file,
|
||||
)
|
||||
from prowler.providers.gcp.gcp_provider import GcpProvider
|
||||
from prowler.providers.gcp.models import GCPIdentityInfo, GCPOutputOptions, GCPProject
|
||||
from tests.providers.gcp.gcp_fixtures import mock_api_client
|
||||
from prowler.providers.gcp.models import GCPIdentityInfo, GCPProject
|
||||
|
||||
|
||||
class TestGCPProvider:
|
||||
@@ -72,99 +71,6 @@ class TestGCPProvider:
|
||||
assert gcp_provider.identity == GCPIdentityInfo(profile="default")
|
||||
assert gcp_provider.audit_config == {"shodan_api_key": None}
|
||||
|
||||
@freeze_time(datetime.today())
|
||||
def test_gcp_provider_output_options(self):
|
||||
arguments = Namespace()
|
||||
arguments.project_id = []
|
||||
arguments.excluded_project_id = []
|
||||
arguments.list_project_id = False
|
||||
arguments.credentials_file = ""
|
||||
arguments.impersonate_service_account = ""
|
||||
arguments.config_file = default_config_file_path
|
||||
arguments.fixer_config = default_fixer_config_file_path
|
||||
|
||||
# Output options
|
||||
arguments.status = []
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
arguments.verbose = True
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.shodan = "test-api-key"
|
||||
|
||||
projects = {
|
||||
"test-project": GCPProject(
|
||||
number="55555555",
|
||||
id="project/55555555",
|
||||
name="test-project",
|
||||
labels={"test": "value"},
|
||||
lifecycle_state="",
|
||||
)
|
||||
}
|
||||
|
||||
mocked_service = MagicMock()
|
||||
|
||||
mocked_service.projects.list.return_value = MagicMock(
|
||||
execute=MagicMock(return_value={"projects": projects})
|
||||
)
|
||||
with patch(
|
||||
"prowler.providers.gcp.gcp_provider.GcpProvider.setup_session",
|
||||
return_value=(None, None),
|
||||
), patch(
|
||||
"prowler.providers.gcp.gcp_provider.GcpProvider.get_projects",
|
||||
return_value=projects,
|
||||
), patch(
|
||||
"prowler.providers.gcp.gcp_provider.GcpProvider.update_projects_with_organizations",
|
||||
return_value=None,
|
||||
), patch(
|
||||
"prowler.providers.gcp.gcp_provider.discovery.build",
|
||||
new=mock_api_client,
|
||||
), patch(
|
||||
"prowler.providers.gcp.gcp_provider.discovery.build",
|
||||
return_value=mocked_service,
|
||||
):
|
||||
gcp_provider = GcpProvider(
|
||||
arguments.project_id,
|
||||
arguments.excluded_project_id,
|
||||
arguments.credentials_file,
|
||||
arguments.impersonate_service_account,
|
||||
arguments.list_project_id,
|
||||
arguments.config_file,
|
||||
arguments.fixer_config,
|
||||
)
|
||||
# This is needed since the output_options requires to get the global provider to get the audit config
|
||||
with patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=gcp_provider,
|
||||
):
|
||||
gcp_provider.output_options = arguments, {}
|
||||
|
||||
assert isinstance(gcp_provider.output_options, GCPOutputOptions)
|
||||
assert gcp_provider.output_options.status == []
|
||||
assert gcp_provider.output_options.output_modes == [
|
||||
"csv",
|
||||
]
|
||||
assert (
|
||||
gcp_provider.output_options.output_directory
|
||||
== arguments.output_directory
|
||||
)
|
||||
assert gcp_provider.output_options.bulk_checks_metadata == {}
|
||||
assert gcp_provider.output_options.verbose
|
||||
assert (
|
||||
f"prowler-output-{gcp_provider.identity.profile}"
|
||||
in gcp_provider.output_options.output_filename
|
||||
)
|
||||
# Flaky due to the millisecond part of the timestamp
|
||||
# assert (
|
||||
# gcp_provider.output_options.output_filename
|
||||
# == f"prowler-output-{gcp_provider.identity.profile}-{datetime.today().strftime('%Y%m%d%H%M%S')}"
|
||||
# )
|
||||
|
||||
# Delete testing directory
|
||||
# TODO: move this to a fixtures file
|
||||
rmdir(f"{arguments.output_directory}/compliance")
|
||||
rmdir(arguments.output_directory)
|
||||
|
||||
@freeze_time(datetime.today())
|
||||
def test_is_project_matching(self):
|
||||
arguments = Namespace()
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
from argparse import Namespace
|
||||
from os import rmdir
|
||||
from unittest.mock import patch
|
||||
|
||||
from kubernetes import client
|
||||
@@ -11,7 +10,6 @@ from prowler.config.config import (
|
||||
from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
|
||||
from prowler.providers.kubernetes.models import (
|
||||
KubernetesIdentityInfo,
|
||||
KubernetesOutputOptions,
|
||||
KubernetesSession,
|
||||
)
|
||||
from tests.providers.kubernetes.kubernetes_fixtures import KUBERNETES_CONFIG
|
||||
@@ -80,76 +78,6 @@ class TestKubernetesProvider:
|
||||
|
||||
assert kubernetes_provider.audit_config == KUBERNETES_CONFIG
|
||||
|
||||
def test_set_provider_output_options_kubernetes(self):
|
||||
arguments = Namespace()
|
||||
arguments.kubeconfig_file = "dummy_path"
|
||||
arguments.context = None
|
||||
arguments.only_logs = False
|
||||
arguments.namespace = None
|
||||
arguments.config_file = default_config_file_path
|
||||
arguments.fixer_config = default_fixer_config_file_path
|
||||
arguments.status = []
|
||||
arguments.output_formats = ["csv"]
|
||||
arguments.output_directory = "output_test_directory"
|
||||
arguments.verbose = True
|
||||
arguments.output_filename = "output_test_filename"
|
||||
arguments.only_logs = False
|
||||
arguments.unix_timestamp = False
|
||||
arguments.shodan = "test-api-key"
|
||||
|
||||
context = {
|
||||
"name": "test-context",
|
||||
"context": {
|
||||
"user": "test-user",
|
||||
"cluster": "test-cluster",
|
||||
},
|
||||
}
|
||||
with patch(
|
||||
"prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.setup_session",
|
||||
return_value=KubernetesSession(
|
||||
api_client=client.ApiClient,
|
||||
context=context,
|
||||
),
|
||||
), patch(
|
||||
"prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
|
||||
return_value=["namespace-1"],
|
||||
):
|
||||
|
||||
kubernetes_provider = KubernetesProvider(
|
||||
arguments.kubeconfig_file,
|
||||
arguments.context,
|
||||
arguments.namespace,
|
||||
arguments.config_file,
|
||||
arguments.fixer_config,
|
||||
)
|
||||
# This is needed since the output_options requires to get the global provider to get the audit config
|
||||
with patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=kubernetes_provider,
|
||||
):
|
||||
|
||||
kubernetes_provider.output_options = arguments, {}
|
||||
|
||||
assert isinstance(
|
||||
kubernetes_provider.output_options, KubernetesOutputOptions
|
||||
)
|
||||
assert kubernetes_provider.output_options.status == []
|
||||
assert kubernetes_provider.output_options.output_modes == ["csv"]
|
||||
assert (
|
||||
kubernetes_provider.output_options.output_directory
|
||||
== arguments.output_directory
|
||||
)
|
||||
assert kubernetes_provider.output_options.bulk_checks_metadata == {}
|
||||
assert kubernetes_provider.output_options.verbose
|
||||
assert (
|
||||
kubernetes_provider.output_options.output_filename
|
||||
== arguments.output_filename
|
||||
)
|
||||
|
||||
# Delete testing directory
|
||||
rmdir(f"{arguments.output_directory}/compliance")
|
||||
rmdir(arguments.output_directory)
|
||||
|
||||
# @patch("kubernetes.client.RbacAuthorizationV1Api")
|
||||
# @patch("kubernetes.config.list_kube_config_contexts")
|
||||
# @patch("kubernetes.config.load_incluster_config")
|
||||
|
||||
Reference in New Issue
Block a user