diff --git a/prowler/__main__.py b/prowler/__main__.py index b8f90e783b..04e4c4df11 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -45,7 +45,6 @@ from prowler.providers.aws.lib.security_hub.security_hub import ( verify_security_hub_integration_enabled_per_region, ) from prowler.providers.common.common import set_global_provider_object -from prowler.providers.common.outputs import set_provider_output_options def prowler(): @@ -174,16 +173,14 @@ def prowler(): # Sort final check list checks_to_execute = sorted(checks_to_execute) - # Parse Mute List - mutelist_file = "" + # Setup Mute List + # TODO: this should be available for all the providers + # Move the argument to the Prowler level to be available for all if hasattr(args, "mutelist_file"): - mutelist_file = global_provider.get_mutelist(args.mutelist_file) + global_provider.mutelist = args.mutelist_file - # Set output options based on the selected provider - # TODO: this is going to be removed an include in the Provider as a new common object - audit_output_options = set_provider_output_options( - provider, args, global_provider.identity, mutelist_file, bulk_checks_metadata - ) + # Setup Output Options + global_provider.output_options = (args, bulk_checks_metadata) # TODO: adapt the quick inventory for the new AWS provider # Run the quick inventory for the provider if available @@ -198,7 +195,6 @@ def prowler(): findings = execute_checks( checks_to_execute, global_provider, - audit_output_options, custom_checks_metadata, ) else: @@ -230,14 +226,19 @@ def prowler(): # Close json file if exists if "json" in mode: close_json( - audit_output_options.output_filename, args.output_directory, mode + global_provider.output_options.output_filename, + args.output_directory, + mode, ) if mode == "html": add_html_footer( - audit_output_options.output_filename, args.output_directory + global_provider.output_options.output_filename, + args.output_directory, ) fill_html_overview_statistics( - stats, audit_output_options.output_filename, args.output_directory + stats, + global_provider.output_options.output_filename, + args.output_directory, ) # Send output to S3 if needed (-B / -D) if provider == "aws" and ( @@ -250,7 +251,7 @@ def prowler(): output_bucket = args.output_bucket_no_assume bucket_session = global_provider.session.original_session send_to_s3_bucket( - audit_output_options.output_filename, + global_provider.output_options.output_filename, args.output_directory, mode, output_bucket, @@ -281,7 +282,10 @@ def prowler(): # Prepare the findings to be sent to Security Hub security_hub_findings_per_region = prepare_security_hub_findings( - findings, provider, audit_output_options, aws_security_enabled_regions + findings, + provider, + global_provider.output_options, + aws_security_enabled_regions, ) # Send the findings to Security Hub @@ -311,7 +315,7 @@ def prowler(): display_summary_table( findings, global_provider, - audit_output_options, + global_provider.output_options, ) if findings: @@ -325,13 +329,13 @@ def prowler(): findings, bulk_checks_metadata, compliance, - audit_output_options.output_filename, - audit_output_options.output_directory, + global_provider.output_options.output_filename, + global_provider.output_options.output_directory, compliance_overview, ) if compliance_overview: print( - f"\nDetailed compliance results are in {Fore.YELLOW}{audit_output_options.output_directory}/compliance/{Style.RESET_ALL}\n" + f"\nDetailed compliance results are in {Fore.YELLOW}{global_provider.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 389db45956..01ec4a74c8 100644 --- a/prowler/lib/check/check.py +++ b/prowler/lib/check/check.py @@ -24,7 +24,6 @@ from prowler.lib.utils.utils import open_file, parse_json_file from prowler.providers.aws.lib.mutelist.mutelist import mutelist_findings from prowler.providers.common.common import get_global_provider from prowler.providers.common.models import Audit_Metadata -from prowler.providers.common.outputs import Provider_Output_Options # Load all checks metadata @@ -398,7 +397,7 @@ def import_check(check_path: str) -> ModuleType: return lib -def run_check(check: Check, output_options: Provider_Output_Options) -> list: +def run_check(check: Check, output_options) -> list: findings = [] if output_options.verbose: print( @@ -422,7 +421,6 @@ def run_check(check: Check, output_options: Provider_Output_Options) -> list: def execute_checks( checks_to_execute: list, global_provider: Any, - audit_output_options: Provider_Output_Options, custom_checks_metadata: Any, ) -> list: # List to store all the check's findings @@ -460,7 +458,7 @@ def execute_checks( ) # Execution with the --only-logs flag - if audit_output_options.only_logs: + if global_provider.output_options.only_logs: for check_name in checks_to_execute: # Recover service from check name service = check_name.split("_")[0] @@ -469,7 +467,6 @@ def execute_checks( service, check_name, global_provider.type, - audit_output_options, global_provider.identity, services_executed, checks_executed, @@ -514,7 +511,6 @@ def execute_checks( check_findings = execute( service, check_name, - audit_output_options, global_provider, services_executed, checks_executed, @@ -541,7 +537,6 @@ def execute_checks( def execute( service: str, check_name: str, - audit_output_options: Provider_Output_Options, global_provider: Any, services_executed: set, checks_executed: set, @@ -559,7 +554,7 @@ def execute( c = update_check_metadata(c, custom_checks_metadata["Checks"][c.CheckID]) # Run check - check_findings = run_check(c, audit_output_options) + check_findings = run_check(c, global_provider.output_options) # Update Audit Status services_executed.add(service) @@ -569,15 +564,15 @@ def execute( ) # Mute List findings - if audit_output_options.mutelist_file: + if hasattr(global_provider, "mutelist") and global_provider.mutelist: check_findings = mutelist_findings( - audit_output_options.mutelist_file, - global_provider.audited_account, + global_provider.mutelist, + global_provider.identity.account, check_findings, ) # Report the check's findings - report(check_findings, audit_output_options, global_provider) + report(check_findings, global_provider) if os.environ.get("PROWLER_REPORT_LIB_PATH"): try: @@ -586,8 +581,9 @@ 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, audit_output_options, global_provider + check_findings, global_provider.output_options, global_provider ) except Exception: sys.exit(1) diff --git a/prowler/lib/outputs/models.py b/prowler/lib/outputs/models.py index b0d0a6d579..4ccfdd8291 100644 --- a/prowler/lib/outputs/models.py +++ b/prowler/lib/outputs/models.py @@ -48,7 +48,7 @@ def get_check_compliance(finding, provider_type, output_options) -> dict: def generate_provider_output_csv(provider, finding, mode: str, fd, output_options): """ - set_provider_output_options configures automatically the outputs based on the selected provider and returns the Provider_Output_Options object. + generate_provider_output_csv creates the provider's CSV output """ try: # Dynamically load the Provider_Output_Options class diff --git a/prowler/lib/outputs/outputs.py b/prowler/lib/outputs/outputs.py index d7f4f7e0e3..c869cd1856 100644 --- a/prowler/lib/outputs/outputs.py +++ b/prowler/lib/outputs/outputs.py @@ -34,8 +34,9 @@ def stdout_report(finding, color, verbose, status): ) -def report(check_findings, output_options, provider): +def report(check_findings, provider): try: + output_options = provider.output_options file_descriptors = {} if check_findings: # TO-DO Generic Function diff --git a/prowler/lib/outputs/summary_table.py b/prowler/lib/outputs/summary_table.py index 22a2426f67..cb11a1eed9 100644 --- a/prowler/lib/outputs/summary_table.py +++ b/prowler/lib/outputs/summary_table.py @@ -11,13 +11,12 @@ from prowler.config.config import ( json_ocsf_file_suffix, ) from prowler.lib.logger import logger -from prowler.providers.common.outputs import Provider_Output_Options def display_summary_table( findings: list, provider, - output_options: Provider_Output_Options, + output_options, ): output_directory = output_options.output_directory output_filename = output_options.output_filename diff --git a/prowler/providers/aws/aws_provider.py b/prowler/providers/aws/aws_provider.py index 1763ef3fe7..80bb85c390 100644 --- a/prowler/providers/aws/aws_provider.py +++ b/prowler/providers/aws/aws_provider.py @@ -33,6 +33,7 @@ from prowler.providers.aws.models import ( AWSIdentityInfo, AWSMFAInfo, AWSOrganizationsInfo, + AWSOutputOptions, AWSSession, ) from prowler.providers.common.models import Audit_Metadata @@ -48,6 +49,9 @@ class AwsProvider(Provider): _audit_config: dict = {} _ignore_unused_services: bool = False _enabled_regions: set = set() + # TODO: enforce the mutelist for the Provider class + _mutelist: dict = {} + _output_options: AWSOutputOptions # TODO: this is not optional, enforce for all providers audit_metadata: Audit_Metadata @@ -254,6 +258,31 @@ class AwsProvider(Provider): def audit_config(self): return self._audit_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): + return self._mutelist + + @mutelist.setter + def mutelist(self, mutelist_path): + if mutelist_path: + mutelist = parse_mutelist_file( + self._session.current_session, self._identity.account, mutelist_path + ) + else: + mutelist = {} + self._mutelist = mutelist + # TODO: This can be moved to another class since it doesn't need self def get_organizations_info( self, organizations_session: Session, aws_account_id: str @@ -774,17 +803,6 @@ Caller Identity ARN: {Fore.YELLOW}[{self._identity.identity_arn}]{Style.RESET_AL ) sys.exit(1) - # TODO: maybe create a function in the provider with a default empty string - def get_mutelist(self, mutelist_file): - # Parse content from Mute List file and get it, if necessary, from S3 - if mutelist_file: - mutelist_file = parse_mutelist_file( - self.session.session, self._identity.account, mutelist_file - ) - else: - mutelist_file = None - return mutelist_file - def read_aws_regions_file() -> dict: # Get JSON locally diff --git a/prowler/providers/aws/lib/mutelist/mutelist.py b/prowler/providers/aws/lib/mutelist/mutelist.py index e416ee8c28..7cd13e7177 100644 --- a/prowler/providers/aws/lib/mutelist/mutelist.py +++ b/prowler/providers/aws/lib/mutelist/mutelist.py @@ -3,6 +3,7 @@ import sys from typing import Any import yaml +from boto3 import Session from boto3.dynamodb.conditions import Attr from schema import Optional, Schema @@ -32,34 +33,34 @@ mutelist_schema = Schema( ) -def parse_mutelist_file(session, aws_account, mutelist_file): +def parse_mutelist_file(session: Session, aws_account: str, mutelist_path: str): try: # Check if file is a S3 URI - if re.search("^s3://([^/]+)/(.*?([^/]+))$", mutelist_file): - bucket = mutelist_file.split("/")[2] - key = ("/").join(mutelist_file.split("/")[3:]) + if re.search("^s3://([^/]+)/(.*?([^/]+))$", mutelist_path): + bucket = mutelist_path.split("/")[2] + key = ("/").join(mutelist_path.split("/")[3:]) s3_client = session.client("s3") mutelist = yaml.safe_load( s3_client.get_object(Bucket=bucket, Key=key)["Body"] )["Mute List"] # Check if file is a Lambda Function ARN - elif re.search(r"^arn:(\w+):lambda:", mutelist_file): - lambda_region = mutelist_file.split(":")[3] + elif re.search(r"^arn:(\w+):lambda:", mutelist_path): + lambda_region = mutelist_path.split(":")[3] lambda_client = session.client("lambda", region_name=lambda_region) lambda_response = lambda_client.invoke( - FunctionName=mutelist_file, InvocationType="RequestResponse" + FunctionName=mutelist_path, InvocationType="RequestResponse" ) lambda_payload = lambda_response["Payload"].read() mutelist = yaml.safe_load(lambda_payload)["Mute List"] # Check if file is a DynamoDB ARN elif re.search( r"^arn:aws(-cn|-us-gov)?:dynamodb:[a-z]{2}-[a-z-]+-[1-9]{1}:[0-9]{12}:table\/[a-zA-Z0-9._-]+$", - mutelist_file, + mutelist_path, ): mutelist = {"Accounts": {}} - table_region = mutelist_file.split(":")[3] + table_region = mutelist_path.split(":")[3] dynamodb_resource = session.resource("dynamodb", region_name=table_region) - dynamo_table = dynamodb_resource.Table(mutelist_file.split("/")[1]) + dynamo_table = dynamodb_resource.Table(mutelist_path.split("/")[1]) response = dynamo_table.scan( FilterExpression=Attr("Accounts").is_in([aws_account, "*"]) ) @@ -90,7 +91,7 @@ def parse_mutelist_file(session, aws_account, mutelist_file): "Exceptions" ] = item["Exceptions"] else: - with open(mutelist_file) as f: + with open(mutelist_path) as f: mutelist = yaml.safe_load(f)["Mute List"] try: mutelist_schema.validate(mutelist) diff --git a/prowler/providers/aws/models.py b/prowler/providers/aws/models.py index d934d2b0fa..26bd6c4a5b 100644 --- a/prowler/providers/aws/models.py +++ b/prowler/providers/aws/models.py @@ -4,7 +4,9 @@ from datetime import datetime from boto3.session import Session from botocore.config import Config +from prowler.config.config import output_file_timestamp from prowler.providers.aws.lib.arn.models import ARN +from prowler.providers.common.models import ProviderOutputOptions @dataclass @@ -75,3 +77,38 @@ class AWSCallerIdentity: class AWSMFAInfo: arn: str totp: str + + +class AWSOutputOptions(ProviderOutputOptions): + security_hub_enabled: bool + + def __init__(self, arguments, bulk_checks_metadata, identity): + # First call Provider_Output_Options init + super().__init__(arguments, bulk_checks_metadata) + + # Confire Shodan API + # TODO: review shodan for the new AWS provider + # if arguments.shodan: + # audit_info = change_config_var( + # "shodan_api_key", arguments.shodan, audit_info + # ) + + # Check if custom output filename was input, if not, set the default + if ( + not hasattr(arguments, "output_filename") + or arguments.output_filename is None + ): + self.output_filename = ( + f"prowler-output-{identity.account}-{output_file_timestamp}" + ) + else: + self.output_filename = arguments.output_filename + + # Security Hub Outputs + self.security_hub_enabled = arguments.security_hub + self.send_sh_only_fails = arguments.send_sh_only_fails + if arguments.security_hub: + if not self.output_modes: + self.output_modes = ["json-asff"] + else: + self.output_modes.append("json-asff") diff --git a/prowler/providers/azure/azure_provider.py b/prowler/providers/azure/azure_provider.py index a4851b4421..e57ec6146e 100644 --- a/prowler/providers/azure/azure_provider.py +++ b/prowler/providers/azure/azure_provider.py @@ -11,7 +11,11 @@ from msgraph import GraphServiceClient from prowler.lib.logger import logger from prowler.providers.azure.lib.regions.regions import get_regions_config -from prowler.providers.azure.models import AzureIdentityInfo, AzureRegionConfig +from prowler.providers.azure.models import ( + AzureIdentityInfo, + AzureOutputOptions, + AzureRegionConfig, +) from prowler.providers.common.models import Audit_Metadata from prowler.providers.common.provider import Provider @@ -23,6 +27,9 @@ class AzureProvider(Provider): _audit_config: Optional[dict] _region_config: AzureRegionConfig _locations: dict + _output_options: AzureOutputOptions + # TODO: enforce the mutelist for the Provider class + # _mutelist: dict = {} # TODO: this is not optional, enforce for all providers audit_metadata: Audit_Metadata @@ -83,6 +90,32 @@ class AzureProvider(Provider): def audit_config(self): return self._audit_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 = AzureOutputOptions( + arguments, bulk_checks_metadata, self._identity + ) + + # TODO: pending to implement + # @property + # def mutelist(self): + # return self._mutelist + + # @mutelist.setter + # def mutelist(self, mutelist_path): + # if mutelist_path: + # mutelist = parse_mutelist_file( + # self._session.current_session, self._identity.account, mutelist_path + # ) + # else: + # mutelist = {} + # self._mutelist = mutelist + # TODO: this should be moved to the argparse, if not we need to enforce it from the Provider def validate_arguments( self, az_cli_auth, sp_env_auth, browser_auth, managed_entity_auth, tenant_id diff --git a/prowler/providers/azure/models.py b/prowler/providers/azure/models.py index 21a47b1744..dec40d0307 100644 --- a/prowler/providers/azure/models.py +++ b/prowler/providers/azure/models.py @@ -1,5 +1,8 @@ from pydantic import BaseModel +from prowler.config.config import output_file_timestamp +from prowler.providers.common.models import ProviderOutputOptions + class AzureIdentityInfo(BaseModel): identity_id: str = "" @@ -15,3 +18,30 @@ class AzureRegionConfig(BaseModel): authority: str = None base_url: str = "" credential_scopes: list = [] + + +class AzureOutputOptions(ProviderOutputOptions): + def __init__(self, arguments, bulk_checks_metadata, identity): + # First call Provider_Output_Options init + super().__init__(arguments, bulk_checks_metadata) + + # Confire Shodan API + # TODO: review shodan for the new AWS provider + # if arguments.shodan: + # audit_info = change_config_var( + # "shodan_api_key", arguments.shodan, audit_info + # ) + + # Check if custom output filename was input, if not, set the default + if ( + not hasattr(arguments, "output_filename") + or arguments.output_filename is None + ): + if identity.domain != "Unknown tenant domain (missing AAD permissions)": + self.output_filename = ( + f"prowler-output-{identity.domain}-{output_file_timestamp}" + ) + else: + self.output_filename = f"prowler-output-{'-'.join(identity.tenant_ids)}-{output_file_timestamp}" + else: + self.output_filename = arguments.output_filename diff --git a/prowler/providers/common/models.py b/prowler/providers/common/models.py index 898dd9e8d9..8d4c8bb726 100644 --- a/prowler/providers/common/models.py +++ b/prowler/providers/common/models.py @@ -1,3 +1,6 @@ +from os import makedirs +from os.path import isdir + from pydantic import BaseModel @@ -9,3 +12,31 @@ class Audit_Metadata(BaseModel): expected_checks: list completed_checks: int audit_progress: int + + +class ProviderOutputOptions: + status: bool + output_modes: list + output_directory: str + bulk_checks_metadata: dict + verbose: str + output_filename: str + only_logs: bool + unix_timestamp: bool + + def __init__(self, arguments, bulk_checks_metadata): + self.status = arguments.status + self.output_modes = arguments.output_modes + self.output_directory = arguments.output_directory + self.verbose = arguments.verbose + self.bulk_checks_metadata = bulk_checks_metadata + self.only_logs = arguments.only_logs + self.unix_timestamp = arguments.unix_timestamp + # Check output directory, if it is not created -> create it + if arguments.output_directory: + if not isdir(arguments.output_directory): + if arguments.output_modes: + makedirs(arguments.output_directory, exist_ok=True) + if not isdir(arguments.output_directory + "/compliance"): + if arguments.output_modes: + makedirs(arguments.output_directory + "/compliance", exist_ok=True) diff --git a/prowler/providers/common/outputs.py b/prowler/providers/common/outputs.py index 597f57ac1f..a8dc7722b9 100644 --- a/prowler/providers/common/outputs.py +++ b/prowler/providers/common/outputs.py @@ -1,34 +1,37 @@ import importlib -import sys -from dataclasses import dataclass -from os import makedirs -from os.path import isdir -from prowler.config.config import output_file_timestamp -from prowler.lib.logger import logger +# TODO: remove after fixing tests +# import sys +# from dataclasses import dataclass +# from os import makedirs +# from os.path import isdir + +# from prowler.lib.logger import logger -def set_provider_output_options( - provider: str, arguments, identity, mutelist_file, bulk_checks_metadata -): - """ - set_provider_output_options configures automatically the outputs based on the selected provider and returns the Provider_Output_Options object. - """ - try: - # Dynamically load the Provider_Output_Options class - provider_output_class = f"{provider.capitalize()}_Output_Options" - provider_output_options = getattr( - importlib.import_module(__name__), provider_output_class - )(arguments, identity, mutelist_file, bulk_checks_metadata) - except Exception as error: - logger.critical( - f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" - ) - sys.exit(1) - else: - return provider_output_options +# TODO: remove after fixing tests +# def set_provider_output_options( +# provider: str, arguments, identity, mutelist_file, bulk_checks_metadata +# ): +# """ +# set_provider_output_options configures automatically the outputs based on the selected provider and returns the Provider_Output_Options object. +# """ +# try: +# # Dynamically load the Provider_Output_Options class +# provider_output_class = f"{provider.capitalize()}_Output_Options" +# provider_output_options = getattr( +# importlib.import_module(__name__), provider_output_class +# )(arguments, identity, mutelist_file, bulk_checks_metadata) +# except Exception as error: +# logger.critical( +# f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" +# ) +# sys.exit(1) +# else: +# return provider_output_options +# TODO: review this function, probably is not needed anymore def get_provider_output_model(provider_type): """ get_provider_output_model returns the model _Check_Output_CSV for each provider @@ -43,128 +46,33 @@ def get_provider_output_model(provider_type): return output_provider_model -@dataclass -class Provider_Output_Options: - status: bool - output_modes: list - output_directory: str - mutelist_file: str - bulk_checks_metadata: dict - verbose: str - output_filename: str - only_logs: bool - unix_timestamp: bool +# TODO: remove after fixing tests +# @dataclass +# class Provider_Output_Options: +# status: bool +# output_modes: list +# output_directory: str +# mutelist_file: str +# bulk_checks_metadata: dict +# verbose: str +# output_filename: str +# only_logs: bool +# unix_timestamp: bool - def __init__(self, arguments, mutelist_file, bulk_checks_metadata): - self.status = arguments.status - self.output_modes = arguments.output_modes - self.output_directory = arguments.output_directory - self.verbose = arguments.verbose - self.bulk_checks_metadata = bulk_checks_metadata - self.mutelist_file = mutelist_file - self.only_logs = arguments.only_logs - self.unix_timestamp = arguments.unix_timestamp - # Check output directory, if it is not created -> create it - if arguments.output_directory: - if not isdir(arguments.output_directory): - if arguments.output_modes: - makedirs(arguments.output_directory, exist_ok=True) - if not isdir(arguments.output_directory + "/compliance"): - if arguments.output_modes: - makedirs(arguments.output_directory + "/compliance", exist_ok=True) - - -class Azure_Output_Options(Provider_Output_Options): - def __init__(self, arguments, identity, mutelist_file, bulk_checks_metadata): - # First call Provider_Output_Options init - super().__init__(arguments, mutelist_file, bulk_checks_metadata) - - # Confire Shodan API - # TODO: review shodan for the new AWS provider - # if arguments.shodan: - # audit_info = change_config_var( - # "shodan_api_key", arguments.shodan, audit_info - # ) - - # Check if custom output filename was input, if not, set the default - if ( - not hasattr(arguments, "output_filename") - or arguments.output_filename is None - ): - if identity.domain != "Unknown tenant domain (missing AAD permissions)": - self.output_filename = ( - f"prowler-output-{identity.domain}-{output_file_timestamp}" - ) - else: - self.output_filename = f"prowler-output-{'-'.join(identity.tenant_ids)}-{output_file_timestamp}" - else: - self.output_filename = arguments.output_filename - - -class Gcp_Output_Options(Provider_Output_Options): - def __init__(self, arguments, identity, mutelist_file, bulk_checks_metadata): - # First call Provider_Output_Options init - super().__init__(arguments, mutelist_file, bulk_checks_metadata) - - # Check if custom output filename was input, if not, set the default - if ( - not hasattr(arguments, "output_filename") - or arguments.output_filename is None - ): - self.output_filename = ( - f"prowler-output-{identity.profile}-{output_file_timestamp}" - ) - else: - self.output_filename = arguments.output_filename - - -class Kubernetes_Output_Options(Provider_Output_Options): - def __init__(self, arguments, identity, mutelist_file, bulk_checks_metadata): - # First call Provider_Output_Options init - super().__init__(arguments, mutelist_file, bulk_checks_metadata) - # TODO move the below if to Provider_Output_Options - # Check if custom output filename was input, if not, set the default - if ( - not hasattr(arguments, "output_filename") - or arguments.output_filename is None - ): - self.output_filename = ( - f"prowler-output-{identity.context}-{output_file_timestamp}" - ) - else: - self.output_filename = arguments.output_filename - - -class Aws_Output_Options(Provider_Output_Options): - security_hub_enabled: bool - - def __init__(self, arguments, identity, mutelist_file, bulk_checks_metadata): - # First call Provider_Output_Options init - super().__init__(arguments, mutelist_file, bulk_checks_metadata) - - # Confire Shodan API - # TODO: review shodan for the new AWS provider - # if arguments.shodan: - # audit_info = change_config_var( - # "shodan_api_key", arguments.shodan, audit_info - # ) - - # Check if custom output filename was input, if not, set the default - if ( - not hasattr(arguments, "output_filename") - or arguments.output_filename is None - ): - self.output_filename = ( - f"prowler-output-{identity.account}-{output_file_timestamp}" - ) - else: - self.output_filename = arguments.output_filename - - # Security Hub Outputs - self.security_hub_enabled = arguments.security_hub - self.send_sh_only_fails = arguments.send_sh_only_fails - if arguments.security_hub: - if not self.output_modes: - self.output_modes = ["json-asff"] - else: - self.output_modes.append("json-asff") +# def __init__(self, arguments, mutelist_file, bulk_checks_metadata): +# self.status = arguments.status +# self.output_modes = arguments.output_modes +# self.output_directory = arguments.output_directory +# self.verbose = arguments.verbose +# self.bulk_checks_metadata = bulk_checks_metadata +# self.mutelist_file = mutelist_file +# self.only_logs = arguments.only_logs +# self.unix_timestamp = arguments.unix_timestamp +# # Check output directory, if it is not created -> create it +# if arguments.output_directory: +# if not isdir(arguments.output_directory): +# if arguments.output_modes: +# makedirs(arguments.output_directory, exist_ok=True) +# if not isdir(arguments.output_directory + "/compliance"): +# if arguments.output_modes: +# makedirs(arguments.output_directory + "/compliance", exist_ok=True) diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 7b76b5ac7e..6477f62897 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -62,6 +62,24 @@ class Provider(ABC): def setup_session(self): pass + @property + @abstractmethod + def output_options(self): + """ + output_options method returns the provider's audit output configuration. + + This method needs to be created in each provider. + """ + + @output_options.setter + @abstractmethod + def output_options(self): + """ + output_options.setter sets the provider's audit output configuration. + + This method needs to be created in each provider. + """ + # TODO: probably this won't be here since we want to do the arguments validation during the parse() def validate_arguments(self): pass diff --git a/prowler/providers/gcp/gcp_provider.py b/prowler/providers/gcp/gcp_provider.py index 54a3e06711..0cc0cf54a1 100644 --- a/prowler/providers/gcp/gcp_provider.py +++ b/prowler/providers/gcp/gcp_provider.py @@ -1,6 +1,5 @@ import os import sys -from dataclasses import dataclass from typing import Optional from colorama import Fore, Style @@ -11,12 +10,7 @@ from googleapiclient import discovery from prowler.lib.logger import logger from prowler.providers.common.models import Audit_Metadata from prowler.providers.common.provider import Provider - - -@dataclass -class GCPIdentityInfo: - profile: str - default_project_id: str +from prowler.providers.gcp.models import GCPIdentityInfo, GCPOutputOptions class GcpProvider(Provider): @@ -25,6 +19,9 @@ class GcpProvider(Provider): _project_ids: list _identity: GCPIdentityInfo _audit_config: Optional[dict] + _output_options: GCPOutputOptions + # TODO: enforce the mutelist for the Provider class + # _mutelist: dict = {} # TODO: this is not optional, enforce for all providers audit_metadata: Audit_Metadata @@ -79,6 +76,32 @@ class GcpProvider(Provider): def audit_config(self): return self._audit_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 + ) + + # TODO: pending to implement + # @property + # def mutelist(self): + # return self._mutelist + + # @mutelist.setter + # def mutelist(self, mutelist_path): + # if mutelist_path: + # mutelist = parse_mutelist_file( + # self._session.current_session, self._identity.account, mutelist_path + # ) + # else: + # mutelist = {} + # self._mutelist = mutelist + def setup_session(self, credentials_file): try: if credentials_file: diff --git a/prowler/providers/gcp/models.py b/prowler/providers/gcp/models.py new file mode 100644 index 0000000000..ec8212a961 --- /dev/null +++ b/prowler/providers/gcp/models.py @@ -0,0 +1,27 @@ +from dataclasses import dataclass + +from prowler.config.config import output_file_timestamp +from prowler.providers.common.models import ProviderOutputOptions + + +@dataclass +class GCPIdentityInfo: + profile: str + default_project_id: str + + +class GCPOutputOptions(ProviderOutputOptions): + def __init__(self, arguments, bulk_checks_metadata, identity): + # First call ProviderOutputOptions init + super().__init__(arguments, bulk_checks_metadata) + + # Check if custom output filename was input, if not, set the default + if ( + not hasattr(arguments, "output_filename") + or arguments.output_filename is None + ): + self.output_filename = ( + f"prowler-output-{identity.profile}-{output_file_timestamp}" + ) + else: + self.output_filename = arguments.output_filename diff --git a/prowler/providers/kubernetes/kubernetes_provider.py b/prowler/providers/kubernetes/kubernetes_provider.py index cebe4ee8fa..a48e52d498 100644 --- a/prowler/providers/kubernetes/kubernetes_provider.py +++ b/prowler/providers/kubernetes/kubernetes_provider.py @@ -1,7 +1,6 @@ import os import sys from argparse import Namespace -from dataclasses import dataclass from typing import Optional from colorama import Fore, Style @@ -10,24 +9,11 @@ from kubernetes import client, config from prowler.lib.logger import logger from prowler.providers.common.models import Audit_Metadata from prowler.providers.common.provider import Provider - - -@dataclass -class KubernetesIdentityInfo: - context: str - cluster: str - user: str - - -@dataclass -class KubernetesSession: - """ - KubernetesSession stores the Kubernetes session's configuration. - - """ - - api_client: client.ApiClient - context: dict +from prowler.providers.kubernetes.models import ( + KubernetesIdentityInfo, + KubernetesOutputOptions, + KubernetesSession, +) class KubernetesProvider(Provider): @@ -36,6 +22,9 @@ class KubernetesProvider(Provider): _namespaces: list _audit_config: Optional[dict] _identity: KubernetesIdentityInfo + _output_options: KubernetesOutputOptions + # TODO: enforce the mutelist for the Provider class + # _mutelist: dict = {} # TODO: this is not optional, enforce for all providers audit_metadata: Audit_Metadata @@ -83,6 +72,32 @@ class KubernetesProvider(Provider): def audit_config(self): return self._audit_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 + ) + + # TODO: pending to implement + # @property + # def mutelist(self): + # return self._mutelist + + # @mutelist.setter + # def mutelist(self, mutelist_path): + # if mutelist_path: + # mutelist = parse_mutelist_file( + # self._session.current_session, self._identity.account, mutelist_path + # ) + # else: + # mutelist = {} + # self._mutelist = mutelist + def setup_session(self, kubeconfig_file, input_context) -> KubernetesSession: """ Sets up the Kubernetes session. diff --git a/prowler/providers/kubernetes/models.py b/prowler/providers/kubernetes/models.py new file mode 100644 index 0000000000..acc15d3929 --- /dev/null +++ b/prowler/providers/kubernetes/models.py @@ -0,0 +1,41 @@ +from dataclasses import dataclass + +from kubernetes import client + +from prowler.config.config import output_file_timestamp +from prowler.providers.common.models import ProviderOutputOptions + + +@dataclass +class KubernetesIdentityInfo: + context: str + cluster: str + user: str + + +@dataclass +class KubernetesSession: + """ + KubernetesSession stores the Kubernetes session's configuration. + + """ + + api_client: client.ApiClient + context: dict + + +class KubernetesOutputOptions(ProviderOutputOptions): + def __init__(self, arguments, bulk_checks_metadata, identity): + # First call ProviderOutputOptions init + super().__init__(arguments, bulk_checks_metadata) + # TODO move the below if to ProviderOutputOptions + # Check if custom output filename was input, if not, set the default + if ( + not hasattr(arguments, "output_filename") + or arguments.output_filename is None + ): + self.output_filename = ( + f"prowler-output-{identity.context}-{output_file_timestamp}" + ) + else: + self.output_filename = arguments.output_filename