fix(autoscaling): Add exception manage while decoding UserData (#4675)

Co-authored-by: Rubén De la Torre Vico <[email protected]>
Co-authored-by: Pepe Fagoaga <[email protected]>
Co-authored-by: Sergio <[email protected]>

Author: 4 people

prowler/config/config.py

@@ -61,6 +61,7 @@ def get_available_compliance_frameworks():
 default_config_file_path = (
     f"{pathlib.Path(os.path.dirname(os.path.realpath(__file__)))}/config.yaml"
 )
+encoding_format_utf_8 = "utf-8"
 
 
 def check_current_version():
@@ -102,8 +103,7 @@ def load_and_validate_config_file(provider: str, config_file_path: str) -> dict:
     load_and_validate_config_file reads the Prowler config file in YAML format from the default location or the file passed with the --config-file flag
     """
     try:
-        with open(config_file_path) as f:
-            config = {}
+        with open(config_file_path, "r", encoding=encoding_format_utf_8) as f:
             config_file = yaml.safe_load(f)
 
             # Not to introduce a breaking change we have to allow the old format config file without any provider keys

prowler/lib/utils/utils.py

@@ -12,13 +12,14 @@
 from detect_secrets import SecretsCollection
 from detect_secrets.settings import default_settings
 
+from prowler.config.config import encoding_format_utf_8
 from prowler.lib.logger import logger
 
 
 def open_file(input_file: str, mode: str = "r") -> TextIOWrapper:
     """open_file returns a handler to the file using the specified mode."""
     try:
-        f = open(input_file, mode)
+        f = open(input_file, mode, encoding=encoding_format_utf_8)
     except OSError as os_error:
         if os_error.strerror == "Too many open files":
             logger.critical(
@@ -66,7 +67,7 @@ def file_exists(filename: str):
 
 def hash_sha512(string: str) -> str:
     """hash_sha512 returns the first 9 bytes of the SHA512 representation for the given string."""
-    return sha512(string.encode("utf-8")).hexdigest()[0:9]
+    return sha512(string.encode(encoding_format_utf_8)).hexdigest()[0:9]
 
 
 def detect_secrets_scan(data):

prowler/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration.py

@@ -6,7 +6,9 @@
 from detect_secrets import SecretsCollection
 from detect_secrets.settings import default_settings
 
+from prowler.config.config import encoding_format_utf_8
 from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.lib.logger import logger
 from prowler.providers.aws.services.autoscaling.autoscaling_client import (
     autoscaling_client,
 )
@@ -25,12 +27,23 @@ def execute(self):
                 temp_user_data_file = tempfile.NamedTemporaryFile(delete=False)
                 user_data = b64decode(configuration.user_data)
 
-                if user_data[0:2] == b"\x1f\x8b":  # GZIP magic number
-                    user_data = zlib.decompress(user_data, zlib.MAX_WBITS | 32).decode(
-                        "utf-8"
+                try:
+                    if user_data[0:2] == b"\x1f\x8b":  # GZIP magic number
+                        user_data = zlib.decompress(
+                            user_data, zlib.MAX_WBITS | 32
+                        ).decode(encoding_format_utf_8)
+                    else:
+                        user_data = user_data.decode(encoding_format_utf_8)
+                except UnicodeDecodeError as error:
+                    logger.warning(
+                        f"{configuration.region} -- Unable to decode user data in autoscaling launch configuration {configuration.name}: {error}"
                     )
-                else:
-                    user_data = user_data.decode("utf-8")
+                    continue
+                except Exception as error:
+                    logger.warning(
+                        f"{configuration.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                    )
+                    continue
 
                 temp_user_data_file.write(
                     bytes(user_data, encoding="raw_unicode_escape")

prowler/providers/aws/services/ec2/ec2_instance_secrets_user_data/ec2_instance_secrets_user_data.py

@@ -6,6 +6,7 @@
 from detect_secrets import SecretsCollection
 from detect_secrets.settings import default_settings
 
+from prowler.config.config import encoding_format_utf_8
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.ec2.ec2_client import ec2_client
 
@@ -26,9 +27,9 @@ def execute(self):
                     if user_data[0:2] == b"\x1f\x8b":  # GZIP magic number
                         user_data = zlib.decompress(
                             user_data, zlib.MAX_WBITS | 32
-                        ).decode("utf-8")
+                        ).decode(encoding_format_utf_8)
                     else:
-                        user_data = user_data.decode("utf-8")
+                        user_data = user_data.decode(encoding_format_utf_8)
 
                     temp_user_data_file.write(
                         bytes(user_data, encoding="raw_unicode_escape")

prowler/providers/aws/services/iam/iam_service.py

@@ -5,6 +5,7 @@
 from botocore.client import ClientError
 from pydantic import BaseModel
 
+from prowler.config.config import encoding_format_utf_8
 from prowler.lib.logger import logger
 from prowler.lib.scan_filters.scan_filters import is_resource_filtered
 from prowler.providers.aws.lib.service.service import AWSService
@@ -144,7 +145,9 @@ def __get_credential_report__(self):
                 if report_status["State"] == "COMPLETE":
                     report_is_completed = True
             # Convert credential report to list of dictionaries
-            credential = self.client.get_credential_report()["Content"].decode("utf-8")
+            credential = self.client.get_credential_report()["Content"].decode(
+                encoding_format_utf_8
+            )
             credential_lines = credential.split("\n")
             csv_reader = csv.DictReader(credential_lines, delimiter=",")
             credential_list = list(csv_reader)

tests/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration_test.py

@@ -287,3 +287,77 @@ def test_one_autoscaling_file_with_secrets_gzip(self):
             assert result[0].resource_id == launch_configuration_name
             assert result[0].resource_arn == launch_configuration_arn
             assert result[0].region == AWS_REGION_US_EAST_1
+
+    @mock_aws
+    def test_one_autoscaling_file_with_unicode_error(self):
+        # Include launch_configurations to check
+        invalid_utf8_bytes = b"\xc0\xaf"
+        launch_configuration_name = "tester"
+        autoscaling_client = client("autoscaling", region_name=AWS_REGION_US_EAST_1)
+        autoscaling_client.create_launch_configuration(
+            LaunchConfigurationName=launch_configuration_name,
+            ImageId="ami-12c6146b",
+            InstanceType="t1.micro",
+            KeyName="the_keys",
+            SecurityGroups=["default", "default2"],
+            UserData=invalid_utf8_bytes,
+        )
+
+        from prowler.providers.aws.services.autoscaling.autoscaling_service import (
+            AutoScaling,
+        )
+
+        current_audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
+            new=current_audit_info,
+        ), mock.patch(
+            "prowler.providers.aws.services.autoscaling.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_client",
+            new=AutoScaling(current_audit_info),
+        ):
+            from prowler.providers.aws.services.autoscaling.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_find_secrets_ec2_launch_configuration import (
+                autoscaling_find_secrets_ec2_launch_configuration,
+            )
+
+            check = autoscaling_find_secrets_ec2_launch_configuration()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    @mock_aws
+    def test_one_autoscaling_file_invalid_gzip_error(self):
+        # Include launch_configurations to check
+        invalid_gzip_bytes = b"\x1f\x8b\xc0\xaf"
+        launch_configuration_name = "tester"
+        autoscaling_client = client("autoscaling", region_name=AWS_REGION_US_EAST_1)
+        autoscaling_client.create_launch_configuration(
+            LaunchConfigurationName=launch_configuration_name,
+            ImageId="ami-12c6146b",
+            InstanceType="t1.micro",
+            KeyName="the_keys",
+            SecurityGroups=["default", "default2"],
+            UserData=invalid_gzip_bytes,
+        )
+
+        from prowler.providers.aws.services.autoscaling.autoscaling_service import (
+            AutoScaling,
+        )
+
+        current_audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
+            new=current_audit_info,
+        ), mock.patch(
+            "prowler.providers.aws.services.autoscaling.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_client",
+            new=AutoScaling(current_audit_info),
+        ):
+            from prowler.providers.aws.services.autoscaling.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_find_secrets_ec2_launch_configuration import (
+                autoscaling_find_secrets_ec2_launch_configuration,
+            )
+
+            check = autoscaling_find_secrets_ec2_launch_configuration()
+            result = check.execute()
+
+            assert len(result) == 0

tests/providers/aws/services/autoscaling/autoscaling_service_test.py

@@ -3,6 +3,7 @@
 from boto3 import client
 from moto import mock_aws
 
+from prowler.config.config import encoding_format_utf_8
 from prowler.providers.aws.services.autoscaling.autoscaling_service import AutoScaling
 from tests.providers.aws.audit_info_utils import (
     AWS_ACCOUNT_NUMBER,
@@ -72,7 +73,9 @@ def test__describe_launch_configurations__(self):
         assert len(autoscaling.launch_configurations) == 2
         assert autoscaling.launch_configurations[0].name == "tester1"
         assert (
-            b64decode(autoscaling.launch_configurations[0].user_data).decode("utf-8")
+            b64decode(autoscaling.launch_configurations[0].user_data).decode(
+                encoding_format_utf_8
+            )
             == "DB_PASSWORD=foobar123"
         )
         assert autoscaling.launch_configurations[0].image_id == "ami-12c6146b"

tests/providers/aws/services/ec2/ec2_service_test.py

@@ -8,6 +8,7 @@
 from freezegun import freeze_time
 from moto import mock_aws
 
+from prowler.config.config import encoding_format_utf_8
 from prowler.providers.aws.services.ec2.ec2_service import EC2
 from tests.providers.aws.audit_info_utils import (
     AWS_ACCOUNT_NUMBER,
@@ -316,7 +317,9 @@ def test__get_instance_user_data__(self):
             [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
         )
         ec2 = EC2(audit_info)
-        assert user_data == b64decode(ec2.instances[0].user_data).decode("utf-8")
+        assert user_data == b64decode(ec2.instances[0].user_data).decode(
+            encoding_format_utf_8
+        )
 
     # Test EC2 Get EBS Encryption by default
     @mock_aws