chore(sdk): add validation for invalid checks, services, and categories (#8971)

Co-authored-by: Andoni Alonso <14891798+andoniaf@users.noreply.github.com>

Author: MrCloudSec

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)

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,15 +59,48 @@ 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:
             for severity in severities:
                 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])
 

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 @@ def test_load_checks_to_execute_with_severities_and_services_multiple(self):
     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 @@ def test_threat_detection_single_check(self):
             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