From 47a6e28d7133d8fd53be42cd3c73fa2740b070e7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Wed, 25 Sep 2024 09:38:21 +0200 Subject: [PATCH] refactor(output_options): remove output options from provider (#5149) --- prowler/__main__.py | 103 ++++++----- prowler/lib/check/check.py | 49 +++-- prowler/lib/outputs/finding.py | 20 ++- prowler/lib/outputs/outputs.py | 24 ++- prowler/lib/scan/scan.py | 5 +- prowler/providers/aws/aws_provider.py | 13 -- prowler/providers/azure/azure_provider.py | 30 +--- prowler/providers/common/provider.py | 21 --- prowler/providers/gcp/gcp_provider.py | 19 +- .../kubernetes/kubernetes_provider.py | 13 -- tests/lib/check/check_test.py | 6 +- tests/lib/outputs/finding_test.py | 12 +- .../output_options/output_options_test.py | 169 ++++++++++++++++++ tests/lib/outputs/outputs_test.py | 51 ++---- tests/lib/scan/scan_test.py | 2 +- tests/providers/aws/aws_provider_test.py | 95 ---------- tests/providers/aws/utils.py | 7 - tests/providers/azure/azure_provider_test.py | 91 +--------- tests/providers/gcp/gcp_provider_test.py | 98 +--------- .../kubernetes/kubernetes_provider_test.py | 72 -------- 20 files changed, 331 insertions(+), 569 deletions(-) create mode 100644 tests/lib/outputs/output_options/output_options_test.py diff --git a/prowler/__main__.py b/prowler/__main__.py index fd0401e1b5..b2a235b166 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -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 diff --git a/prowler/lib/check/check.py b/prowler/lib/check/check.py index 32a40349db..8a2f210c90 100644 --- a/prowler/lib/check/check.py +++ b/prowler/lib/check/check.py @@ -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: diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py index f7695221c2..aea7c13a59 100644 --- a/prowler/lib/outputs/finding.py +++ b/prowler/lib/outputs/finding.py @@ -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": diff --git a/prowler/lib/outputs/outputs.py b/prowler/lib/outputs/outputs.py index bb7a643b77..d600c851a6 100644 --- a/prowler/lib/outputs/outputs.py +++ b/prowler/lib/outputs/outputs.py @@ -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( diff --git a/prowler/lib/scan/scan.py b/prowler/lib/scan/scan.py index f3f664b987..8cada8a90b 100644 --- a/prowler/lib/scan/scan.py +++ b/prowler/lib/scan/scan.py @@ -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 ] diff --git a/prowler/providers/aws/aws_provider.py b/prowler/providers/aws/aws_provider.py index 63ea4dda2e..83091299a8 100644 --- a/prowler/providers/aws/aws_provider.py +++ b/prowler/providers/aws/aws_provider.py @@ -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: """ diff --git a/prowler/providers/azure/azure_provider.py b/prowler/providers/azure/azure_provider.py index 4f2de1aa9d..96a0d8e575 100644 --- a/prowler/providers/azure/azure_provider.py +++ b/prowler/providers/azure/azure_provider.py @@ -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.""" diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index bfffcae492..bcea5bcfc4 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -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: """ diff --git a/prowler/providers/gcp/gcp_provider.py b/prowler/providers/gcp/gcp_provider.py index 62bfc983ff..9060c6c72d 100644 --- a/prowler/providers/gcp/gcp_provider.py +++ b/prowler/providers/gcp/gcp_provider.py @@ -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: """ diff --git a/prowler/providers/kubernetes/kubernetes_provider.py b/prowler/providers/kubernetes/kubernetes_provider.py index e3e0699dd4..8f7a377516 100644 --- a/prowler/providers/kubernetes/kubernetes_provider.py +++ b/prowler/providers/kubernetes/kubernetes_provider.py @@ -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: """ diff --git a/tests/lib/check/check_test.py b/tests/lib/check/check_test.py index 16348cdca8..4ef1fcdaf9 100644 --- a/tests/lib/check/check_test.py +++ b/tests/lib/check/check_test.py @@ -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 diff --git a/tests/lib/outputs/finding_test.py b/tests/lib/outputs/finding_test.py index ecd0a368d7..40041b405e 100644 --- a/tests/lib/outputs/finding_test.py +++ b/tests/lib/outputs/finding_test.py @@ -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" diff --git a/tests/lib/outputs/output_options/output_options_test.py b/tests/lib/outputs/output_options/output_options_test.py new file mode 100644 index 0000000000..0c788b8fb0 --- /dev/null +++ b/tests/lib/outputs/output_options/output_options_test.py @@ -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 diff --git a/tests/lib/outputs/outputs_test.py b/tests/lib/outputs/outputs_test.py index 52a08e7fa8..624366af5e 100644 --- a/tests/lib/outputs/outputs_test.py +++ b/tests/lib/outputs/outputs_test.py @@ -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( diff --git a/tests/lib/scan/scan_test.py b/tests/lib/scan/scan_test.py index 15d45ba6e0..949347e233 100644 --- a/tests/lib/scan/scan_test.py +++ b/tests/lib/scan/scan_test.py @@ -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 diff --git a/tests/providers/aws/aws_provider_test.py b/tests/providers/aws/aws_provider_test.py index 06c79817b6..e4e5fe491a 100644 --- a/tests/providers/aws/aws_provider_test.py +++ b/tests/providers/aws/aws_provider_test.py @@ -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 diff --git a/tests/providers/aws/utils.py b/tests/providers/aws/utils.py index 738fce0940..7104bef9dd 100644 --- a/tests/providers/aws/utils.py +++ b/tests/providers/aws/utils.py @@ -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 diff --git a/tests/providers/azure/azure_provider_test.py b/tests/providers/azure/azure_provider_test.py index 3a4a6c1797..334c83bf42 100644 --- a/tests/providers/azure/azure_provider_test.py +++ b/tests/providers/azure/azure_provider_test.py @@ -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" diff --git a/tests/providers/gcp/gcp_provider_test.py b/tests/providers/gcp/gcp_provider_test.py index 769c9cba55..99d8e61c4e 100644 --- a/tests/providers/gcp/gcp_provider_test.py +++ b/tests/providers/gcp/gcp_provider_test.py @@ -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() diff --git a/tests/providers/kubernetes/kubernetes_provider_test.py b/tests/providers/kubernetes/kubernetes_provider_test.py index 02e99f6ad1..5edafc1e3f 100644 --- a/tests/providers/kubernetes/kubernetes_provider_test.py +++ b/tests/providers/kubernetes/kubernetes_provider_test.py @@ -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")