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

Co-authored-by: Sergio <sergio@prowler.com>

prowler/config/config.py

@@ -64,7 +64,7 @@ default_config_file_path = (
 default_fixer_config_file_path = (
     f"{pathlib.Path(os.path.dirname(os.path.realpath(__file__)))}/fixer_config.yaml"
 )
-enconding_format_utf_8 = "utf-8"
+encoding_format_utf_8 = "utf-8"
 available_output_formats = ["csv", "json-asff", "json-ocsf", "html"]
 
 
@@ -110,7 +110,7 @@ def load_and_validate_config_file(provider: str, config_file_path: str) -> dict:
         dict: The configuration dictionary for the specified provider.
     """
     try:
-        with open(config_file_path, "r", encoding=enconding_format_utf_8) as f:
+        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, allow the old format config file without any provider keys
@@ -159,7 +159,7 @@ def load_and_validate_fixer_config_file(
         dict: The fixer configuration dictionary for the specified provider.
     """
     try:
-        with open(fixer_config_file_path, "r", encoding=enconding_format_utf_8) as f:
+        with open(fixer_config_file_path, "r", encoding=encoding_format_utf_8) as f:
             fixer_config_file = yaml.safe_load(f)
             return fixer_config_file.get(provider, {})
 

prowler/lib/utils/utils.py

@@ -22,14 +22,14 @@ from colorama import Style
 from detect_secrets import SecretsCollection
 from detect_secrets.settings import default_settings
 
-from prowler.config.config import enconding_format_utf_8
+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, encoding=enconding_format_utf_8)
+        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(
@@ -77,7 +77,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(enconding_format_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,8 +6,9 @@ from base64 import b64decode
 from detect_secrets import SecretsCollection
 from detect_secrets.settings import default_settings
 
-from prowler.config.config import enconding_format_utf_8
+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,
 )
@@ -26,12 +27,23 @@ class autoscaling_find_secrets_ec2_launch_configuration(Check):
                 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(
-                        enconding_format_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(enconding_format_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,7 +6,7 @@ from base64 import b64decode
 from detect_secrets import SecretsCollection
 from detect_secrets.settings import default_settings
 
-from prowler.config.config import enconding_format_utf_8
+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
 
@@ -27,9 +27,9 @@ class ec2_instance_secrets_user_data(Check):
                     if user_data[0:2] == b"\x1f\x8b":  # GZIP magic number
                         user_data = zlib.decompress(
                             user_data, zlib.MAX_WBITS | 32
-                        ).decode(enconding_format_utf_8)
+                        ).decode(encoding_format_utf_8)
                     else:
-                        user_data = user_data.decode(enconding_format_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/ec2/ec2_launch_template_no_secrets/ec2_launch_template_no_secrets.py

@@ -6,7 +6,7 @@ from base64 import b64decode
 from detect_secrets import SecretsCollection
 from detect_secrets.settings import default_settings
 
-from prowler.config.config import enconding_format_utf_8
+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
 
@@ -31,10 +31,10 @@ class ec2_launch_template_no_secrets(Check):
 
                 if user_data[0:2] == b"\x1f\x8b":  # GZIP magic number
                     user_data = zlib.decompress(user_data, zlib.MAX_WBITS | 32).decode(
-                        enconding_format_utf_8
+                        encoding_format_utf_8
                     )
                 else:
-                    user_data = user_data.decode(enconding_format_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,7 +5,7 @@ from typing import Optional
 from botocore.client import ClientError
 from pydantic import BaseModel
 
-from prowler.config.config import enconding_format_utf_8
+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
@@ -146,7 +146,7 @@ class IAM(AWSService):
                     report_is_completed = True
             # Convert credential report to list of dictionaries
             credential = self.client.get_credential_report()["Content"].decode(
-                enconding_format_utf_8
+                encoding_format_utf_8
             )
             credential_lines = credential.split("\n")
             csv_reader = csv.DictReader(credential_lines, delimiter=",")

tests/providers/aws/lib/mutelist/aws_mutelist_test.py

@@ -7,7 +7,7 @@ from boto3 import client, resource
 from mock import MagicMock, patch
 from moto import mock_aws
 
-from prowler.config.config import enconding_format_utf_8
+from prowler.config.config import encoding_format_utf_8
 from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist
 from tests.providers.aws.services.awslambda.awslambda_service_test import (
     create_zip_file,
@@ -46,7 +46,7 @@ def mock_make_api_call(self, operation_name, kwarg):
                             }
                         }
                     }
-                ).encode(enconding_format_utf_8)
+                ).encode(encoding_format_utf_8)
             )
         }
 

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

@@ -284,3 +284,77 @@ class Test_autoscaling_find_secrets_ec2_launch_configuration:
             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,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.autoscaling.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_client",
+            new=AutoScaling(aws_provider),
+        ):
+            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,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.autoscaling.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_find_secrets_ec2_launch_configuration.autoscaling_client",
+            new=AutoScaling(aws_provider),
+        ):
+            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,7 +3,7 @@ from base64 import b64decode
 from boto3 import client
 from moto import mock_aws
 
-from prowler.config.config import enconding_format_utf_8
+from prowler.config.config import encoding_format_utf_8
 from prowler.providers.aws.services.autoscaling.autoscaling_service import AutoScaling
 from tests.providers.aws.utils import (
     AWS_ACCOUNT_NUMBER,
@@ -74,7 +74,7 @@ class Test_AutoScaling_Service:
         assert autoscaling.launch_configurations[0].name == "tester1"
         assert (
             b64decode(autoscaling.launch_configurations[0].user_data).decode(
-                enconding_format_utf_8
+                encoding_format_utf_8
             )
             == "DB_PASSWORD=foobar123"
         )

tests/providers/aws/services/ec2/ec2_launch_template_no_secrets/ec2_launch_template_no_secrets_test.py

@@ -6,7 +6,7 @@ from unittest import mock
 from boto3 import client
 from moto import mock_aws
 
-from prowler.config.config import enconding_format_utf_8
+from prowler.config.config import encoding_format_utf_8
 from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
 
 ACTUAL_DIRECTORY = Path(path.dirname(path.realpath(__file__)))
@@ -51,8 +51,8 @@ class Test_ec2_launch_template_no_secrets:
             LaunchTemplateData={
                 "InstanceType": "t1.micro",
                 "UserData": b64encode(
-                    "This is some user_data".encode(enconding_format_utf_8)
-                ).decode(enconding_format_utf_8),
+                    "This is some user_data".encode(encoding_format_utf_8)
+                ).decode(encoding_format_utf_8),
             },
         )
 
@@ -104,7 +104,7 @@ class Test_ec2_launch_template_no_secrets:
             VersionDescription="Launch Template with secrets",
             LaunchTemplateData={
                 "InstanceType": "t1.micro",
-                "UserData": b64encode(secrets).decode(enconding_format_utf_8),
+                "UserData": b64encode(secrets).decode(encoding_format_utf_8),
             },
         )
 
@@ -160,7 +160,7 @@ class Test_ec2_launch_template_no_secrets:
             VersionDescription="Launch Template with secrets",
             LaunchTemplateData={
                 "InstanceType": "t1.micro",
-                "UserData": b64encode(secrets).decode(enconding_format_utf_8),
+                "UserData": b64encode(secrets).decode(encoding_format_utf_8),
             },
         )
 
@@ -169,7 +169,7 @@ class Test_ec2_launch_template_no_secrets:
             VersionDescription="Second Launch Template version with secrets",
             LaunchTemplateData={
                 "InstanceType": "t1.micro",
-                "UserData": b64encode(secrets).decode(enconding_format_utf_8),
+                "UserData": b64encode(secrets).decode(encoding_format_utf_8),
             },
         )
 
@@ -228,7 +228,7 @@ class Test_ec2_launch_template_no_secrets:
             VersionDescription="Launch Template with secrets",
             LaunchTemplateData={
                 "InstanceType": "t1.micro",
-                "UserData": b64encode(secrets).decode(enconding_format_utf_8),
+                "UserData": b64encode(secrets).decode(encoding_format_utf_8),
             },
         )
 
@@ -292,7 +292,7 @@ class Test_ec2_launch_template_no_secrets:
             VersionDescription="Launch Template with secrets",
             LaunchTemplateData={
                 "InstanceType": "t1.micro",
-                "UserData": b64encode(secrets).decode(enconding_format_utf_8),
+                "UserData": b64encode(secrets).decode(encoding_format_utf_8),
             },
         )
 
@@ -394,7 +394,7 @@ class Test_ec2_launch_template_no_secrets:
             VersionDescription="Launch Template with secrets",
             LaunchTemplateData={
                 "InstanceType": "t1.micro",
-                "UserData": b64encode(secrets).decode(enconding_format_utf_8),
+                "UserData": b64encode(secrets).decode(encoding_format_utf_8),
             },
         )
 
@@ -411,7 +411,7 @@ class Test_ec2_launch_template_no_secrets:
             VersionDescription="Launch Template without secrets",
             LaunchTemplateData={
                 "InstanceType": "t1.micro",
-                "UserData": b64encode(b"Test").decode(enconding_format_utf_8),
+                "UserData": b64encode(b"Test").decode(encoding_format_utf_8),
             },
         )
 

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

@@ -10,7 +10,7 @@ from dateutil.tz import tzutc
 from freezegun import freeze_time
 from moto import mock_aws
 
-from prowler.config.config import enconding_format_utf_8
+from prowler.config.config import encoding_format_utf_8
 from prowler.providers.aws.services.ec2.ec2_service import EC2
 from tests.providers.aws.utils import (
     AWS_ACCOUNT_NUMBER,
@@ -322,7 +322,7 @@ class Test_EC2_Service:
         )
         ec2 = EC2(aws_provider)
         assert user_data == b64decode(ec2.instances[0].user_data).decode(
-            enconding_format_utf_8
+            encoding_format_utf_8
         )
 
     # Test EC2 Get EBS Encryption by default
@@ -636,8 +636,8 @@ class Test_EC2_Service:
             LaunchTemplateData={
                 "InstanceType": TEMPLATE_INSTANCE_TYPE,
                 "UserData": b64encode(
-                    KNOWN_SECRET_USER_DATA.encode(enconding_format_utf_8)
-                ).decode(enconding_format_utf_8),
+                    KNOWN_SECRET_USER_DATA.encode(encoding_format_utf_8)
+                ).decode(encoding_format_utf_8),
             },
         )
 
@@ -681,8 +681,8 @@ class Test_EC2_Service:
             LaunchTemplateData={
                 "InstanceType": TEMPLATE_INSTANCE_TYPE,
                 "UserData": b64encode(
-                    KNOWN_SECRET_USER_DATA.encode(enconding_format_utf_8)
-                ).decode(enconding_format_utf_8),
+                    KNOWN_SECRET_USER_DATA.encode(encoding_format_utf_8)
+                ).decode(encoding_format_utf_8),
             },
         )
 
@@ -704,6 +704,6 @@ class Test_EC2_Service:
 
         assert version2.template_data["InstanceType"] == TEMPLATE_INSTANCE_TYPE
         assert (
-            b64decode(version2.template_data["UserData"]).decode(enconding_format_utf_8)
+            b64decode(version2.template_data["UserData"]).decode(encoding_format_utf_8)
             == KNOWN_SECRET_USER_DATA
         )