From e038b2fd111fd11c58dc16c2c73900d1b694fb47 Mon Sep 17 00:00:00 2001 From: Sergio Garcia Date: Thu, 6 Nov 2025 05:46:21 -0500 Subject: [PATCH] chore(sdk): add validation for invalid checks, services, and categories (#8971) Co-authored-by: Andoni Alonso <14891798+andoniaf@users.noreply.github.com> --- prowler/CHANGELOG.md | 1 + prowler/lib/check/checks_loader.py | 66 +++++++++++- tests/lib/check/check_loader_test.py | 154 +++++++++++++++++++++++++-- 3 files changed, 214 insertions(+), 7 deletions(-) diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 334aea97cd..fd34a5e1af 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -14,6 +14,7 @@ All notable changes to the **Prowler SDK** are documented in this file. - C5 compliance framework for Azure provider [(#9081)](https://github.com/prowler-cloud/prowler/pull/9081) - C5 compliance framework for the GCP provider [(#9097)](https://github.com/prowler-cloud/prowler/pull/9097) - HIPAA compliance framework for the GCP provider [(#8955)](https://github.com/prowler-cloud/prowler/pull/8955) +- Added validation for invalid checks, services, and categories in `load_checks_to_execute` function [(#8971)](https://github.com/prowler-cloud/prowler/pull/8971) ### Changed - Update AWS Direct Connect service metadata to new format [(#8855)](https://github.com/prowler-cloud/prowler/pull/8855) diff --git a/prowler/lib/check/checks_loader.py b/prowler/lib/check/checks_loader.py index c03034b0e4..dab5535d77 100644 --- a/prowler/lib/check/checks_loader.py +++ b/prowler/lib/check/checks_loader.py @@ -1,3 +1,5 @@ +import sys + from colorama import Fore, Style from prowler.lib.check.check import parse_checks_from_file @@ -57,8 +59,24 @@ def load_checks_to_execute( # Handle if there are checks passed using -c/--checks if check_list: + # Validate that all checks exist + available_checks = set(bulk_checks_metadata.keys()) + available_checks.update(check_aliases.keys()) + invalid_checks = [] for check_name in check_list: - checks_to_execute.add(check_name) + if check_name not in available_checks: + invalid_checks.append(check_name) + else: + checks_to_execute.add(check_name) + + if invalid_checks: + logger.critical( + f"Invalid check(s) specified: {', '.join(invalid_checks)}" + ) + logger.critical( + f"Please provide valid check names. Use 'prowler {provider} --list-checks' to see available checks." + ) + sys.exit(1) # Handle if there are some severities passed using --severity elif severities: @@ -66,6 +84,23 @@ def load_checks_to_execute( checks_to_execute.update(check_severities[severity]) if service_list: + # Validate that all services exist + available_services = set() + for metadata in bulk_checks_metadata.values(): + available_services.add(metadata.ServiceName) + + invalid_services = [ + s for s in service_list if s not in available_services + ] + if invalid_services: + logger.critical( + f"Invalid service(s) specified: {', '.join(invalid_services)}" + ) + logger.critical( + f"Please provide valid service names. Use 'prowler {provider} --list-services' to see available services." + ) + sys.exit(1) + checks_from_services = set() for service in service_list: service_checks = CheckMetadata.list( @@ -81,6 +116,21 @@ def load_checks_to_execute( # Handle if there are services passed using -s/--services elif service_list: + # Validate that all services exist + available_services = set() + for metadata in bulk_checks_metadata.values(): + available_services.add(metadata.ServiceName) + + invalid_services = [s for s in service_list if s not in available_services] + if invalid_services: + logger.critical( + f"Invalid service(s) specified: {', '.join(invalid_services)}" + ) + logger.critical( + f"Please provide valid service names. Use 'prowler {provider} --list-services' to see available services." + ) + sys.exit(1) + for service in service_list: checks_to_execute.update( CheckMetadata.list( @@ -103,6 +153,20 @@ def load_checks_to_execute( # Handle if there are categories passed using --categories elif categories: + # Validate that all categories exist + available_categories = set(check_categories.keys()) + invalid_categories = [ + c for c in categories if c not in available_categories + ] + if invalid_categories: + logger.critical( + f"Invalid category(ies) specified: {', '.join(invalid_categories)}" + ) + logger.critical( + f"Please provide valid category names. Use 'prowler {provider} --list-categories' to see available categories." + ) + sys.exit(1) + for category in categories: checks_to_execute.update(check_categories[category]) diff --git a/tests/lib/check/check_loader_test.py b/tests/lib/check/check_loader_test.py index 12e9b3e547..740ed67844 100644 --- a/tests/lib/check/check_loader_test.py +++ b/tests/lib/check/check_loader_test.py @@ -1,3 +1,4 @@ +import pytest from mock import patch from prowler.lib.check.checks_loader import ( @@ -190,18 +191,22 @@ class TestCheckLoader: def test_load_checks_to_execute_with_severities_and_services_not_within_severity( self, ): + """Test that service not in metadata causes sys.exit(1) when used with severities""" bulk_checks_metatada = { S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() } service_list = ["ec2"] severities = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY] - assert set() == load_checks_to_execute( - bulk_checks_metadata=bulk_checks_metatada, - service_list=service_list, - severities=severities, - provider=self.provider, - ) + # ec2 service doesn't exist in the metadata, so it should exit with error + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + service_list=service_list, + severities=severities, + provider=self.provider, + ) + assert exc_info.value.code == 1 def test_load_checks_to_execute_with_checks_file( self, @@ -382,3 +387,140 @@ class TestCheckLoader: categories=categories, provider=self.provider, ) + + def test_load_checks_to_execute_with_invalid_check(self): + """Test that invalid check names cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + check_list = ["invalid_check_name"] + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + check_list=check_list, + provider=self.provider, + ) + assert exc_info.value.code == 1 + + def test_load_checks_to_execute_with_multiple_invalid_checks(self): + """Test that multiple invalid check names cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + check_list = ["invalid_check_1", "invalid_check_2", "invalid_check_3"] + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + check_list=check_list, + provider=self.provider, + ) + assert exc_info.value.code == 1 + + def test_load_checks_to_execute_with_mixed_valid_invalid_checks(self): + """Test that mix of valid and invalid checks cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + check_list = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME, "invalid_check"] + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + check_list=check_list, + provider=self.provider, + ) + assert exc_info.value.code == 1 + + def test_load_checks_to_execute_with_invalid_service(self): + """Test that invalid service names cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + service_list = ["invalid_service"] + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + service_list=service_list, + provider=self.provider, + ) + assert exc_info.value.code == 1 + + def test_load_checks_to_execute_with_invalid_service_and_severity(self): + """Test that invalid service names with severity cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + service_list = ["invalid_service"] + severities = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY] + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + service_list=service_list, + severities=severities, + provider=self.provider, + ) + assert exc_info.value.code == 1 + + def test_load_checks_to_execute_with_multiple_invalid_services(self): + """Test that multiple invalid service names cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + service_list = ["invalid_service_1", "invalid_service_2"] + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + service_list=service_list, + provider=self.provider, + ) + assert exc_info.value.code == 1 + + def test_load_checks_to_execute_with_invalid_category(self): + """Test that invalid category names cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + categories = {"invalid_category"} + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + categories=categories, + provider=self.provider, + ) + assert exc_info.value.code == 1 + + def test_load_checks_to_execute_with_multiple_invalid_categories(self): + """Test that multiple invalid category names cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + categories = {"invalid_category_1", "invalid_category_2"} + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + categories=categories, + provider=self.provider, + ) + assert exc_info.value.code == 1 + + def test_load_checks_to_execute_with_mixed_valid_invalid_categories(self): + """Test that mix of valid and invalid categories cause sys.exit(1)""" + bulk_checks_metatada = { + S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata() + } + categories = {"internet-exposed", "invalid_category"} + + with pytest.raises(SystemExit) as exc_info: + load_checks_to_execute( + bulk_checks_metadata=bulk_checks_metatada, + categories=categories, + provider=self.provider, + ) + assert exc_info.value.code == 1