From b7c22d18ab2fb66d491df786744832df0fa47ac3 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 7 Aug 2024 17:47:48 -0400 Subject: [PATCH] fix(autoscaling): Add exception manage while decoding UserData (#4675) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Rubén De la Torre Vico Co-authored-by: Pepe Fagoaga Co-authored-by: Sergio --- prowler/config/config.py | 4 +- prowler/lib/utils/utils.py | 5 +- ...g_find_secrets_ec2_launch_configuration.py | 23 ++++-- .../ec2_instance_secrets_user_data.py | 5 +- .../providers/aws/services/iam/iam_service.py | 5 +- ...d_secrets_ec2_launch_configuration_test.py | 74 +++++++++++++++++++ .../autoscaling/autoscaling_service_test.py | 5 +- .../aws/services/ec2/ec2_service_test.py | 5 +- 8 files changed, 112 insertions(+), 14 deletions(-) diff --git a/prowler/config/config.py b/prowler/config/config.py index 1d17c43d15..3191cae670 100644 --- a/prowler/config/config.py +++ b/prowler/config/config.py @@ -61,6 +61,7 @@ html_file_suffix = ".html" 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 diff --git a/prowler/lib/utils/utils.py b/prowler/lib/utils/utils.py index 687f6b2092..883d694f71 100644 --- a/prowler/lib/utils/utils.py +++ b/prowler/lib/utils/utils.py @@ -12,13 +12,14 @@ from time import mktime 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): diff --git a/prowler/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration.py b/prowler/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration.py index ed6c94e099..f496b1472d 100644 --- a/prowler/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration.py +++ b/prowler/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration.py @@ -6,7 +6,9 @@ from base64 import b64decode 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 @@ 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( - "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") diff --git a/prowler/providers/aws/services/ec2/ec2_instance_secrets_user_data/ec2_instance_secrets_user_data.py b/prowler/providers/aws/services/ec2/ec2_instance_secrets_user_data/ec2_instance_secrets_user_data.py index 9dc5c6a1a0..5911c97f8f 100644 --- a/prowler/providers/aws/services/ec2/ec2_instance_secrets_user_data/ec2_instance_secrets_user_data.py +++ b/prowler/providers/aws/services/ec2/ec2_instance_secrets_user_data/ec2_instance_secrets_user_data.py @@ -6,6 +6,7 @@ from base64 import b64decode 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 @@ 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("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") diff --git a/prowler/providers/aws/services/iam/iam_service.py b/prowler/providers/aws/services/iam/iam_service.py index b0cb5f112d..cdec499b63 100644 --- a/prowler/providers/aws/services/iam/iam_service.py +++ b/prowler/providers/aws/services/iam/iam_service.py @@ -5,6 +5,7 @@ from typing import Optional 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 @@ class IAM(AWSService): 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) diff --git a/tests/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration_test.py b/tests/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration_test.py index 6a40b3416e..5f290f64cd 100644 --- a/tests/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration_test.py +++ b/tests/providers/aws/services/autoscaling/autoscaling_find_secrets_ec2_launch_configuration/autoscaling_find_secrets_ec2_launch_configuration_test.py @@ -287,3 +287,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, + ) + + 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 diff --git a/tests/providers/aws/services/autoscaling/autoscaling_service_test.py b/tests/providers/aws/services/autoscaling/autoscaling_service_test.py index 230181436e..b48764d4a3 100644 --- a/tests/providers/aws/services/autoscaling/autoscaling_service_test.py +++ b/tests/providers/aws/services/autoscaling/autoscaling_service_test.py @@ -3,6 +3,7 @@ from base64 import b64decode 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 @@ class Test_AutoScaling_Service: 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" diff --git a/tests/providers/aws/services/ec2/ec2_service_test.py b/tests/providers/aws/services/ec2/ec2_service_test.py index 4321ef4eda..6bcd0c769e 100644 --- a/tests/providers/aws/services/ec2/ec2_service_test.py +++ b/tests/providers/aws/services/ec2/ec2_service_test.py @@ -8,6 +8,7 @@ from dateutil.tz import tzutc 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 @@ class Test_EC2_Service: [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