feat(fixer): add Prowler Fixer feature!

prowler/__main__.py

@@ -16,14 +16,17 @@
     execute_checks,
     list_categories,
     list_checks_json,
+    list_fixers,
     list_services,
     parse_checks_from_folder,
     print_categories,
     print_checks,
     print_compliance_frameworks,
     print_compliance_requirements,
+    print_fixers,
     print_services,
     remove_custom_checks_module,
+    run_fixer,
 )
 from prowler.lib.check.checks_loader import load_checks_to_execute
 from prowler.lib.check.compliance import update_checks_metadata_with_compliance
@@ -94,6 +97,10 @@
         print_services(list_services(provider))
         sys.exit()
 
+    if args.list_fixer:
+        print_fixers(list_fixers(provider))
+        sys.exit()
+
     # Load checks metadata
     logger.debug("Loading checks metadata from .metadata.json files")
     bulk_checks_metadata = bulk_load_checks_metadata(provider)
@@ -210,6 +217,16 @@
             "There are no checks to execute. Please, check your input arguments"
         )
 
+    # Prowler Fixer
+    if global_provider.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):
+            run_fixer(findings)
+        else:
+            print(f"{Style.BRIGHT}{Fore.GREEN}\nNo findings to fix!{Style.RESET_ALL}\n")
+        sys.exit()
+
     # Extract findings stats
     stats = extract_findings_statistics(findings)
 

prowler/lib/banner.py

@@ -15,7 +15,7 @@
 """
     print(banner)
 
-    if args.verbose:
+    if args.verbose or getattr(args, "fix", None):
         print(
             f"""
 Color code for results:

prowler/lib/check/check.py

@@ -36,6 +36,9 @@
         # Build check path name
         check_name = check_info[0]
         check_path = check_info[1]
+        # Ignore fixer files
+        if check_name.endswith("_fixer"):
+            continue
         # Append metadata file extension
         metadata_file = f"{check_path}/{check_name}.metadata.json"
         # Load metadata
@@ -203,6 +206,20 @@
     return sorted(available_services)
 
 
+def list_fixers(provider: str) -> set:
+    available_fixers = set()
+    checks = recover_checks_from_provider(provider)
+    # Build list of check's metadata files
+    for check_info in checks:
+        # Build check path name
+        check_name = check_info[0]
+        # Ignore non fixer files
+        if not check_name.endswith("_fixer"):
+            continue
+        available_fixers.add(check_name)
+    return sorted(available_fixers)
+
+
 def list_categories(bulk_checks_metadata: dict) -> set:
     available_categories = set()
     for check in bulk_checks_metadata.values():
@@ -239,6 +256,23 @@
     print(message)
 
 
+def print_fixers(fixers_list: set):
+    services_num = len(fixers_list)
+    plural_string = (
+        f"\nThere are {Fore.YELLOW}{services_num}{Style.RESET_ALL} available fixers.\n"
+    )
+    singular_string = (
+        f"\nThere is {Fore.YELLOW}{services_num}{Style.RESET_ALL} available fixer.\n"
+    )
+
+    message = plural_string if services_num > 1 else singular_string
+
+    for service in fixers_list:
+        print(f"- {service}")
+
+    print(message)
+
+
 def print_compliance_frameworks(
     bulk_compliance_frameworks: dict,
 ):
@@ -399,8 +433,16 @@
 
 
 def run_check(check: Check, output_options) -> list:
+    """
+    Run the check and return the findings
+    Args:
+        check (Check): check class
+        output_options (Any): output options
+    Returns:
+        list: list of findings
+    """
     findings = []
-    if output_options.verbose:
+    if output_options.verbose or output_options.fixer:
         print(
             f"\nCheck ID: {check.CheckID} - {Fore.MAGENTA}{check.ServiceName}{Fore.YELLOW} [{check.Severity}]{Style.RESET_ALL}"
         )
@@ -419,6 +461,45 @@
         return findings
 
 
+def run_fixer(check_findings: list):
+    """
+    Run the fixer for the check if it exists and there are any FAIL findings
+    Args:
+        check_findings (list): list of findings
+    """
+    try:
+        # Map findings to each check
+        findings_dict = {}
+        for finding in check_findings:
+            if finding.check_metadata.CheckID not in findings_dict:
+                findings_dict[finding.check_metadata.CheckID] = []
+            findings_dict[finding.check_metadata.CheckID].append(finding)
+
+        for check, findings in findings_dict.items():
+            # Check if there are any FAIL findings for the check
+            if any("FAIL" in finding.status for finding in findings):
+                try:
+                    check_module_path = f"prowler.providers.{findings[0].check_metadata.Provider}.services.{findings[0].check_metadata.ServiceName}.{check}.{check}_fixer"
+                    lib = import_check(check_module_path)
+                    fixer = getattr(lib, "fixer")
+                except AttributeError:
+                    logger.error(f"Fixer method not implemented for check {check}")
+                else:
+                    print(
+                        f"\nFixing fails for check {Fore.YELLOW}{check}{Style.RESET_ALL}..."
+                    )
+                    for finding in findings:
+                        if finding.status == "FAIL":
+                            print(
+                                f"\t{orange_color}FIXING{Style.RESET_ALL} {finding.region}... {(Fore.GREEN + 'DONE') if fixer(finding.region) else (Fore.RED + 'ERROR')}{Style.RESET_ALL}"
+                            )
+                    print()
+    except Exception as error:
+        logger.error(
+            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+        )
+
+
 def execute_checks(
     checks_to_execute: list,
     global_provider: Any,
@@ -569,14 +650,18 @@
         lib = import_check(check_module_path)
         # Recover functions from check
         check_to_execute = getattr(lib, check_name)
-        c = check_to_execute()
+        check_class = check_to_execute()
 
         # Update check metadata to reflect that in the outputs
-        if custom_checks_metadata and custom_checks_metadata["Checks"].get(c.CheckID):
-            c = update_check_metadata(c, custom_checks_metadata["Checks"][c.CheckID])
+        if custom_checks_metadata and custom_checks_metadata["Checks"].get(
+            check_class.CheckID
+        ):
+            check_class = update_check_metadata(
+                check_class, custom_checks_metadata["Checks"][check_class.CheckID]
+            )
 
         # Run check
-        check_findings = run_check(c, global_provider.output_options)
+        check_findings = run_check(check_class, global_provider.output_options)
 
         # Update Audit Status
         services_executed.add(service)

prowler/lib/check/checks_loader.py

@@ -101,7 +101,7 @@ def load_checks_to_execute(
 
         # Only execute threat detection checks if threat-detection category is set
         if "threat-detection" not in categories:
-            for threat_detection_check in check_categories["threat-detection"]:
+            for threat_detection_check in check_categories.get("threat-detection", []):
                 checks_to_execute.discard(threat_detection_check)
 
         # Check Aliases

prowler/lib/cli/parser.py

@@ -291,6 +291,7 @@ def __init_list_checks_parser__(self):
         )
         list_group.add_argument(
             "--list-compliance",
+            "--list-compliances",
             action="store_true",
             help="List all available compliance frameworks",
         )
@@ -305,6 +306,12 @@ def __init_list_checks_parser__(self):
             action="store_true",
             help="List the available check's categories",
         )
+        list_group.add_argument(
+            "--list-fixer",
+            "--list-fixers",
+            action="store_true",
+            help="List fixers available for the provider",
+        )
 
     def __init_mutelist_parser__(self):
         mutelist_subparser = self.common_providers_parser.add_argument_group(

prowler/lib/outputs/outputs.py

@@ -23,7 +23,7 @@
 from prowler.lib.outputs.utils import unroll_dict
 
 
-def stdout_report(finding, color, verbose, status):
+def stdout_report(finding, color, verbose, status, fix):
     if finding.check_metadata.Provider == "aws":
         details = finding.region
     if finding.check_metadata.Provider == "azure":
@@ -33,7 +33,7 @@
     if finding.check_metadata.Provider == "kubernetes":
         details = finding.namespace.lower()
 
-    if verbose and (not status or finding.status in status):
+    if (verbose or fix) and (not status or finding.status in status):
         if finding.muted:
             print(
                 f"\t{color}MUTED ({finding.status}){Style.RESET_ALL} {details}: {finding.status_extended}"
@@ -57,7 +57,7 @@
                 check_findings.sort(key=lambda x: x.subscription)
 
             # Generate the required output files
-            if output_options.output_modes:
+            if output_options.output_modes and not output_options.fixer:
                 # We have to create the required output files
                 file_descriptors = fill_file_descriptors(
                     output_options.output_modes,
@@ -70,7 +70,11 @@
                 # Print findings by stdout
                 color = set_report_color(finding.status, finding.muted)
                 stdout_report(
-                    finding, color, output_options.verbose, output_options.status
+                    finding,
+                    color,
+                    output_options.verbose,
+                    output_options.status,
+                    output_options.fixer,
                 )
 
                 if file_descriptors:

prowler/providers/aws/aws_provider.py

@@ -496,8 +496,7 @@
             f"{Style.BRIGHT}AWS Regions: {Style.RESET_ALL}{Fore.YELLOW}{regions}{Style.RESET_ALL}",
             f"{Style.BRIGHT}AWS Account: {Style.RESET_ALL}{Fore.YELLOW}{self._identity.account}{Style.RESET_ALL}",
             f"{Style.BRIGHT}User Id: {Style.RESET_ALL}{Fore.YELLOW}{self._identity.user_id}{Style.RESET_ALL}",
-            f"{Style.BRIGHT}Caller Identity ARN: {Style.RESET_ALL}",
-            f"{Fore.YELLOW}{self._identity.identity_arn}{Style.RESET_ALL}",
+            f"{Style.BRIGHT}Caller Identity ARN: {Style.RESET_ALL}{Fore.YELLOW}{self._identity.identity_arn}{Style.RESET_ALL}",
         ]
         # If -A is set, print Assumed Role ARN
         if (
@@ -507,7 +506,9 @@
             report_lines.append(
                 f"Assumed Role ARN: {Fore.YELLOW}[{self._assumed_role.info.role_arn.arn}]{Style.RESET_ALL}"
             )
-        report_title = f"{Style.BRIGHT}Prowler is using the AWS credentials below:{Style.RESET_ALL}"
+        report_title = (
+            f"{Style.BRIGHT}Using the AWS credentials below:{Style.RESET_ALL}"
+        )
         print_boxes(report_lines, report_title)
 
     def generate_regional_clients(

prowler/providers/aws/lib/arguments/arguments.py

@@ -159,6 +159,14 @@ def init_parser(self):
         help="Scan unused services",
     )
 
+    # Prowler Fixer
+    prowler_fixer_subparser = aws_parser.add_argument_group("Prowler Fixer")
+    prowler_fixer_subparser.add_argument(
+        "--fixer",
+        action="store_true",
+        help="Fix the failed findings that can be fixed by Prowler",
+    )
+
 
 def validate_session_duration(duration):
     """validate_session_duration validates that the AWS STS Assume Role Session Duration is between 900 and 43200 seconds."""

prowler/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer.py

@@ -0,0 +1,33 @@
+from prowler.lib.logger import logger
+from prowler.providers.aws.services.ec2.ec2_client import ec2_client
+
+
+def fixer(region):
+    """
+    Enable EBS encryption by default in a region.
+    Requires the ec2:EnableEbsEncryptionByDefault permission:
+    {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Action": "ec2:EnableEbsEncryptionByDefault",
+                "Resource": "*"
+            }
+        ]
+    }
+    Args:
+        region (str): AWS region
+    Returns:
+        bool: True if EBS encryption by default is enabled, False otherwise
+    """
+    try:
+        regional_client = ec2_client.regional_clients[region]
+        return regional_client.enable_ebs_encryption_by_default()[
+            "EbsEncryptionByDefault"
+        ]
+    except Exception as error:
+        logger.error(
+            f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+        )
+        return False

prowler/providers/azure/azure_provider.py

@@ -182,7 +182,9 @@
             f"{Style.BRIGHT}Azure Subscriptions:{Style.RESET_ALL} {Fore.YELLOW}{printed_subscriptions}{Style.RESET_ALL}",
             f"{Style.BRIGHT}Azure Identity Type:{Style.RESET_ALL} {Fore.YELLOW}{self._identity.identity_type}{Style.RESET_ALL} {Style.BRIGHT}Azure Identity ID:{Style.RESET_ALL} {Fore.YELLOW}{self._identity.identity_id}{Style.RESET_ALL}",
         ]
-        report_title = f"{Style.BRIGHT}Prowler is using the Azure credentials below:{Style.RESET_ALL}"
+        report_title = (
+            f"{Style.BRIGHT}Using the Azure credentials below:{Style.RESET_ALL}"
+        )
         print_boxes(report_lines, report_title)
 
     # TODO: setup_session or setup_credentials?

prowler/providers/common/models.py

@@ -35,13 +35,14 @@ def __init__(self, arguments, bulk_checks_metadata):
         self.only_logs = arguments.only_logs
         self.unix_timestamp = arguments.unix_timestamp
         self.shodan_api_key = arguments.shodan
+        self.fixer = getattr(arguments, "fixer", None)
 
         # Shodan API Key
         if arguments.shodan:
             update_provider_config("shodan_api_key", arguments.shodan)
 
         # Check output directory, if it is not created -> create it
-        if arguments.output_directory:
+        if arguments.output_directory and not self.fixer:
             if not isdir(arguments.output_directory):
                 if arguments.output_formats:
                     makedirs(arguments.output_directory, exist_ok=True)

prowler/providers/gcp/gcp_provider.py

@@ -172,7 +172,9 @@
             f"{Style.BRIGHT}GCP Account:{Style.RESET_ALL} {Fore.YELLOW}{self.identity.profile}{Style.RESET_ALL}",
             f"{Style.BRIGHT}GCP Project IDs:{Style.RESET_ALL} {Fore.YELLOW}{', '.join(self.project_ids)}{Style.RESET_ALL}",
         ]
-        report_title = f"{Style.BRIGHT}Prowler is using the GCP credentials below:{Style.RESET_ALL}"
+        report_title = (
+            f"{Style.BRIGHT}Using the GCP credentials below:{Style.RESET_ALL}"
+        )
         print_boxes(report_lines, report_title)
 
     def get_projects(self) -> dict[str, GCPProject]:

prowler/providers/kubernetes/kubernetes_provider.py

@@ -279,5 +279,7 @@
                 f"{Style.BRIGHT}Namespaces:{Style.RESET_ALL} {Fore.YELLOW}{', '.join(self.namespaces)}{Style.RESET_ALL}",
                 f"{Style.BRIGHT}Roles:{Style.RESET_ALL} {Fore.YELLOW}{roles_str}{Style.RESET_ALL}",
             ]
-        report_title = f"{Style.BRIGHT}Prowler is using the Kubernetes credentials below:{Style.RESET_ALL}"
+        report_title = (
+            f"{Style.BRIGHT}Using the Kubernetes credentials below:{Style.RESET_ALL}"
+        )
         print_boxes(report_lines, report_title)