Merge branch 'prowler-4.0-dev' into cloudtrail-threat-detection

prowler/__main__.py

@@ -193,6 +193,7 @@ def prowler():
             global_provider,
             custom_checks_metadata,
             getattr(args, "mutelist_file", None),
+            args.config_file,
         )
     else:
         logger.error(

prowler/lib/check/check.py

@@ -21,7 +21,7 @@
 from prowler.lib.logger import logger
 from prowler.lib.mutelist.mutelist import mutelist_findings
 from prowler.lib.outputs.outputs import report
-from prowler.lib.utils.utils import open_file, parse_json_file
+from prowler.lib.utils.utils import open_file, parse_json_file, print_boxes
 from prowler.providers.common.common import get_global_provider
 from prowler.providers.common.models import Audit_Metadata
 
@@ -423,6 +423,7 @@ def execute_checks(
     global_provider: Any,
     custom_checks_metadata: Any,
     mutelist_file: str,
+    config_file: str,
 ) -> list:
     # List to store all the check's findings
     all_findings = []
@@ -485,11 +486,22 @@ def execute_checks(
                     f"{check_name} - {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                 )
     else:
-        # Print the mutelist (if any) that is being used
+        # Prepare your messages
+        messages = [
+            f"{Style.BRIGHT}Config File: {Style.RESET_ALL}{Fore.YELLOW}{config_file}{Style.RESET_ALL}"
+        ]
         if mutelist_file:
-            print(
-                f"{Style.BRIGHT}Using the following Mute List: {Style.RESET_ALL}{Fore.YELLOW}{mutelist_file}{Style.RESET_ALL}\n"
+            messages.append(
+                f"{Style.BRIGHT}Mute List File: {Style.RESET_ALL}{Fore.YELLOW}{mutelist_file}{Style.RESET_ALL}"
             )
+        if global_provider.type == "aws":
+            messages.append(
+                f"{Style.BRIGHT}Scanning unused services and resources: {Style.RESET_ALL}{Fore.YELLOW}{global_provider.scan_unused_services}{Style.RESET_ALL}"
+            )
+        report_title = (
+            f"{Style.BRIGHT}Using the following configuration:{Style.RESET_ALL}"
+        )
+        print_boxes(messages, report_title)
         # Default execution
         checks_num = len(checks_to_execute)
         plural_string = "checks"

prowler/lib/utils/utils.py

@@ -2,6 +2,7 @@
 import json
 import os
 import pwd
+import re
 import sys
 import tempfile
 from datetime import datetime
@@ -12,6 +13,7 @@
 from time import mktime
 from typing import Optional
 
+from colorama import Style
 from detect_secrets import SecretsCollection
 from detect_secrets.settings import default_settings
 
@@ -162,3 +164,47 @@ def is_owned_by_root(file_path: str) -> bool:
             f"{file_path}: {e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}"
         )
         return False
+
+
+def strip_ansi_codes(s: str):
+    """
+    Strips ANSI escape codes from a string.
+    Args:
+        s (str): The string to strip.
+    Returns:
+        str: The string without ANSI escape codes.
+    """
+    ansi_escape = re.compile(r"(?:\x1B[@-_][0-?]*[ -/]*[@-~])")
+    return ansi_escape.sub("", s)
+
+
+def print_boxes(messages: list, report_title: str):
+    """
+    Prints a series of messages in a box format.
+    Args:
+        messages (list): A list of messages to print.
+    """
+    # Determine the length of the longest message
+    max_length = max(len(strip_ansi_codes(message)) for message in messages)
+    # Adding a bit of padding for aesthetics
+    padding = 4
+    box_width = max_length + padding
+    # Print the top border
+    print(f"{Style.BRIGHT}+{'-' * (box_width)}+{Style.RESET_ALL}")
+    # Print report title
+    space_padding = box_width - len(strip_ansi_codes(report_title)) - 1
+    print(
+        f"{Style.BRIGHT}|{Style.RESET_ALL} {report_title}{' ' * space_padding}{Style.BRIGHT}|{Style.RESET_ALL}"
+    )
+    print(f"{Style.BRIGHT}+{'-' * (box_width)}+{Style.RESET_ALL}")
+
+    # Print each message centered within the box
+    for message in messages:
+        space_padding = box_width - len(strip_ansi_codes(message)) - 1
+        # Ensure message is properly padded
+        print(
+            f"{Style.BRIGHT}|{Style.RESET_ALL} {message}{' ' * space_padding}{Style.BRIGHT}|{Style.RESET_ALL}"
+        )
+
+    # Print the bottom border
+    print(f"{Style.BRIGHT}+{'-' * (box_width)}+{Style.RESET_ALL}\n")

prowler/providers/aws/aws_provider.py

@@ -14,7 +14,7 @@
 from prowler.lib.check.check import list_modules, recover_checks_from_service
 from prowler.lib.logger import logger
 from prowler.lib.mutelist.mutelist import parse_mutelist_file
-from prowler.lib.utils.utils import open_file, parse_json_file
+from prowler.lib.utils.utils import open_file, parse_json_file, print_boxes
 from prowler.providers.aws.config import (
     AWS_STS_GLOBAL_ENDPOINT_REGION,
     BOTO3_USER_AGENT_EXTRA,
@@ -491,30 +491,23 @@ def print_credentials(self):
         profile = (
             self._identity.profile if self._identity.profile is not None else "default"
         )
-        # TODO: rename AWS Filter Region to AWS Regions, and UserId to User ID
-        # review new banner
-        #       report = f"""
-        # The current audit for AWS will use the following credentials:
-
-        # CLI Profile: {Fore.YELLOW}[{profile}]{Style.RESET_ALL} Regions: {Fore.YELLOW}[{regions}]{Style.RESET_ALL}
-        # Account: {Fore.YELLOW}[{self._identity.account}]{Style.RESET_ALL} User ID: {Fore.YELLOW}[{self._identity.user_id}]{Style.RESET_ALL}
-        # Caller Identity ARN: {Fore.YELLOW}[{self._identity.identity_arn}]{Style.RESET_ALL}
-        # """
-        report = f"""
-This report is being generated using credentials below:
-
-AWS-CLI Profile: {Fore.YELLOW}[{profile}]{Style.RESET_ALL} AWS Filter Region: {Fore.YELLOW}[{regions}]{Style.RESET_ALL}
-AWS Account: {Fore.YELLOW}[{self._identity.account}]{Style.RESET_ALL} UserId: {Fore.YELLOW}[{self._identity.user_id}]{Style.RESET_ALL}
-Caller Identity ARN: {Fore.YELLOW}[{self._identity.identity_arn}]{Style.RESET_ALL}
-"""
+        report_lines = [
+            f"{Style.BRIGHT}AWS-CLI Profile: {Style.RESET_ALL}{Fore.YELLOW}{profile}{Style.RESET_ALL}",
+            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}{Fore.YELLOW}{self._identity.identity_arn}{Style.RESET_ALL}",
+        ]
         # If -A is set, print Assumed Role ARN
         if (
             hasattr(self, "_assumed_role")
             and self._assumed_role.info.role_arn is not None
         ):
-            report += f"""Assumed Role ARN: {Fore.YELLOW}[{self._assumed_role.info.role_arn.arn}]{Style.RESET_ALL}
-        """
-        print(report)
+            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}"
+        print_boxes(report_lines, report_title)
 
     def generate_regional_clients(
         self,

prowler/providers/azure/azure_provider.py

@@ -11,6 +11,7 @@
 from prowler.config.config import load_and_validate_config_file
 from prowler.lib.logger import logger
 from prowler.lib.mutelist.mutelist import parse_mutelist_file
+from prowler.lib.utils.utils import print_boxes
 from prowler.providers.azure.lib.regions.regions import get_regions_config
 from prowler.providers.azure.models import (
     AzureIdentityInfo,
@@ -175,14 +176,14 @@ def print_credentials(self):
         for key, value in self._identity.subscriptions.items():
             intermediate = key + ": " + value
             printed_subscriptions.append(intermediate)
-        report = f"""
-This report is being generated using the identity below:
-
-Azure Tenant ID: {Fore.YELLOW}[{self._identity.tenant_ids[0]}]{Style.RESET_ALL} Azure Tenant Domain: {Fore.YELLOW}[{self._identity.tenant_domain}]{Style.RESET_ALL} Azure Region: {Fore.YELLOW}[{self.region_config.name}]{Style.RESET_ALL}
-Azure Subscriptions: {Fore.YELLOW}{printed_subscriptions}{Style.RESET_ALL}
-Azure Identity Type: {Fore.YELLOW}[{self._identity.identity_type}]{Style.RESET_ALL} Azure Identity ID: {Fore.YELLOW}[{self._identity.identity_id}]{Style.RESET_ALL}
-"""
-        print(report)
+        report_lines = [
+            f"{Style.BRIGHT}Azure Tenant Domain:{Style.RESET_ALL} {Fore.YELLOW}{self._identity.tenant_domain}{Style.RESET_ALL} {Style.BRIGHT}Azure Tenant ID:{Style.RESET_ALL} {Fore.YELLOW}{self._identity.tenant_ids[0]}{Style.RESET_ALL}",
+            f"{Style.BRIGHT}Azure Region:{Style.RESET_ALL} {Fore.YELLOW}{self.region_config.name}{Style.RESET_ALL}",
+            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}"
+        print_boxes(report_lines, report_title)
 
     # TODO: setup_session or setup_credentials?
     def setup_session(

prowler/providers/gcp/gcp_provider.py

@@ -10,6 +10,7 @@
 from prowler.config.config import load_and_validate_config_file
 from prowler.lib.logger import logger
 from prowler.lib.mutelist.mutelist import parse_mutelist_file
+from prowler.lib.utils.utils import print_boxes
 from prowler.providers.common.models import Audit_Metadata
 from prowler.providers.common.provider import Provider
 from prowler.providers.gcp.models import (
@@ -36,7 +37,7 @@ def __init__(self, arguments):
         input_project_ids = arguments.project_ids
         credentials_file = arguments.credentials_file
 
-        self._session, default_project_id = self.setup_session(credentials_file)
+        self._session = self.setup_session(credentials_file)
 
         self._project_ids = []
         self._projects = {}
@@ -66,7 +67,6 @@ def __init__(self, arguments):
 
         self._identity = GCPIdentityInfo(
             profile=getattr(self.session, "_service_account_email", "default"),
-            default_project_id=default_project_id,
         )
 
         # TODO: move this to the providers, pending for AWS, GCP, AZURE and K8s
@@ -148,9 +148,10 @@ def setup_session(self, credentials_file):
             if credentials_file:
                 self.__set_gcp_creds_env_var__(credentials_file)
 
-            return auth.default(
+            credentials, _ = auth.default(
                 scopes=["https://www.googleapis.com/auth/cloud-platform"]
             )
+            return credentials
         except Exception as error:
             logger.critical(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -167,12 +168,12 @@ def __set_gcp_creds_env_var__(self, credentials_file):
     def print_credentials(self):
         # TODO: Beautify audited profile, set "default" if there is no profile set
         # TODO: improve print_credentials with more data like name, number, organization
-        report = f"""
-This report is being generated using credentials below:
-
-GCP Account: {Fore.YELLOW}[{self.identity.profile}]{Style.RESET_ALL}  GCP Project IDs: {Fore.YELLOW}[{", ".join(self.project_ids)}]{Style.RESET_ALL}
-"""
-        print(report)
+        report_lines = [
+            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}"
+        print_boxes(report_lines, report_title)
 
     def get_projects(self) -> dict[str, GCPProject]:
         try:

prowler/providers/gcp/lib/service/service.py

@@ -24,7 +24,6 @@ def __init__(
         self.service = service.lower() if not service.islower() else service
         self.credentials = provider.session
         self.api_version = api_version
-        self.default_project_id = provider.identity.default_project_id
         self.region = region
         self.client = self.__generate_client__(
             self.service, api_version, self.credentials

prowler/providers/gcp/models.py

@@ -8,7 +8,6 @@
 
 class GCPIdentityInfo(BaseModel):
     profile: str
-    default_project_id: str
 
 
 class GCPOrganization(BaseModel):

prowler/providers/kubernetes/kubernetes_provider.py

@@ -8,6 +8,7 @@
 from prowler.config.config import load_and_validate_config_file
 from prowler.lib.logger import logger
 from prowler.lib.mutelist.mutelist import parse_mutelist_file
+from prowler.lib.utils.utils import print_boxes
 from prowler.providers.common.models import Audit_Metadata
 from prowler.providers.common.provider import Provider
 from prowler.providers.kubernetes.models import (
@@ -265,19 +266,18 @@ def print_credentials(self):
         Prints the Kubernetes credentials.
         """
         if self._identity.context == "In-Cluster":
-            report = f"""
-This report is being generated using the Kubernetes configuration below:
-
-Kubernetes Pod: {Fore.YELLOW}[prowler]{Style.RESET_ALL}  Namespace: {Fore.YELLOW}[{self.get_pod_current_namespace()}]{Style.RESET_ALL}
-"""
-            print(report)
+            report_lines = [
+                f"{Style.BRIGHT}Kubernetes Pod:{Style.RESET_ALL} {Fore.YELLOW}prowler{Style.RESET_ALL}",
+                f"{Style.BRIGHT}Namespace:{Style.RESET_ALL} {Fore.YELLOW}{self.get_pod_current_namespace()}{Style.RESET_ALL}",
+            ]
         else:
             roles = self.get_context_user_roles()
             roles_str = ", ".join(roles) if roles else "No associated Roles"
-
-            report = f"""
-This report is being generated using the Kubernetes configuration below:
-
-Kubernetes Cluster: {Fore.YELLOW}[{self._identity.cluster}]{Style.RESET_ALL} User: {Fore.YELLOW}[{self._identity.user}]{Style.RESET_ALL} Namespaces: {Fore.YELLOW}[{', '.join(self.namespaces)}]{Style.RESET_ALL} Roles: {Fore.YELLOW}[{roles_str}]{Style.RESET_ALL}
-"""
-            print(report)
+            report_lines = [
+                f"{Style.BRIGHT}Kubernetes Cluster:{Style.RESET_ALL} {Fore.YELLOW}{self._identity.cluster}{Style.RESET_ALL}",
+                f"{Style.BRIGHT}User:{Style.RESET_ALL} {Fore.YELLOW}{self._identity.user}{Style.RESET_ALL}",
+                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}"
+        print_boxes(report_lines, report_title)

tests/lib/outputs/slack_test.py

@@ -44,7 +44,6 @@ def test_create_message_identity_azure(self):
     def test_create_message_identity_gcp(self):
         gcp_provider = set_mocked_gcp_provider(
             project_ids=["test-project1", "test-project2"],
-            default_project_id="test-project1",
         )
 
         assert create_message_identity(gcp_provider) == (

tests/lib/utils/utils_test.py

@@ -15,6 +15,7 @@
     open_file,
     outputs_unix_timestamp,
     parse_json_file,
+    strip_ansi_codes,
     validate_ip_address,
 )
 
@@ -162,3 +163,21 @@ def test_is_owned_by_root(self):
         assert not is_owned_by_root("not_existing_file")
         # Not valid for darwin systems
         # assert is_owned_by_root("/etc/passwd")
+
+
+class TestStripAnsiCodes:
+    def test_strip_ansi_codes_no_alteration(self):
+        input_string = "\x1B[31mHello\x1B[0m World"
+        expected_output = "Hello World"
+
+        actual_output = strip_ansi_codes(input_string)
+
+        assert actual_output == expected_output
+
+    def test_strip_ansi_codes_empty_string(self):
+        input_string = ""
+        expected_output = ""
+
+        actual_output = strip_ansi_codes(input_string)
+
+        assert actual_output == expected_output

tests/providers/gcp/gcp_fixtures.py

@@ -7,15 +7,14 @@
 
 
 def set_mocked_gcp_provider(
-    project_ids: list[str] = [], default_project_id: str = "", profile: str = ""
+    project_ids: list[str] = [], profile: str = ""
 ) -> GcpProvider:
     provider = MagicMock()
     provider.type = "gcp"
     provider.session = None
     provider.project_ids = project_ids
     provider.identity = GCPIdentityInfo(
         profile=profile,
-        default_project_id=default_project_id,
     )
 
     return provider

tests/providers/gcp/gcp_provider_test.py

@@ -28,7 +28,7 @@ def test_gcp_provider(self):
         }
         with patch(
             "prowler.providers.gcp.gcp_provider.GcpProvider.setup_session",
-            return_value=(None, ""),
+            return_value=None,
         ), patch(
             "prowler.providers.gcp.gcp_provider.GcpProvider.get_projects",
             return_value=projects,
@@ -40,9 +40,7 @@ def test_gcp_provider(self):
             assert gcp_provider.session is None
             assert gcp_provider.project_ids == ["test-project"]
             assert gcp_provider.projects == projects
-            assert gcp_provider.identity == GCPIdentityInfo(
-                profile="default", default_project_id=""
-            )
+            assert gcp_provider.identity == GCPIdentityInfo(profile="default")
             assert gcp_provider.audit_config == {"shodan_api_key": None}
 
     @freeze_time(datetime.today())
@@ -72,7 +70,7 @@ def test_gcp_provider_output_options(self):
         }
         with patch(
             "prowler.providers.gcp.gcp_provider.GcpProvider.setup_session",
-            return_value=(None, ""),
+            return_value=None,
         ), patch(
             "prowler.providers.gcp.gcp_provider.GcpProvider.get_projects",
             return_value=projects,