feat(cli): add --resource-group flag to filter checks by resource group (#10479)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

prowler/CHANGELOG.md

@@ -12,6 +12,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - `awslambda_function_no_dead_letter_queue`, `awslambda_function_using_cross_account_layers`, and `awslambda_function_env_vars_not_encrypted_with_cmk` checks for AWS Lambda [(#10381)](https://github.com/prowler-cloud/prowler/pull/10381)
 - `entra_conditional_access_policy_mdm_compliant_device_required` check for M365 provider [(#10220)](https://github.com/prowler-cloud/prowler/pull/10220)
 - `ec2_securitygroup_allow_ingress_from_internet_to_any_port_from_ip` check for AWS provider using `ipaddress.is_global` for accurate public IP detection [(#10335)](https://github.com/prowler-cloud/prowler/pull/10335)
+- `--resource-group` and `--list-resource-groups` CLI flags to filter checks by resource group across all providers [(#10479)](https://github.com/prowler-cloud/prowler/pull/10479)
 
 ### 🔄 Changed
 

prowler/__main__.py

@@ -27,6 +27,7 @@ from prowler.lib.check.check import (
     list_categories,
     list_checks_json,
     list_fixers,
+    list_resource_groups,
     list_services,
     load_custom_checks_metadata,
     parse_checks_from_file,
@@ -36,6 +37,7 @@ from prowler.lib.check.check import (
     print_compliance_frameworks,
     print_compliance_requirements,
     print_fixers,
+    print_resource_groups,
     print_services,
     remove_custom_checks_module,
     run_fixer,
@@ -161,6 +163,7 @@ def prowler():
     excluded_services = args.excluded_service
     services = args.service
     categories = args.category
+    resource_groups = args.resource_group
     checks_file = args.checks_file
     checks_folder = args.checks_folder
     severities = args.severity
@@ -170,6 +173,7 @@ def prowler():
         not checks
         and not services
         and not categories
+        and not resource_groups
         and not excluded_checks
         and not excluded_services
         and not severities
@@ -215,6 +219,10 @@ def prowler():
         print_categories(list_categories(bulk_checks_metadata))
         sys.exit()
 
+    if args.list_resource_groups:
+        print_resource_groups(list_resource_groups(bulk_checks_metadata))
+        sys.exit()
+
     bulk_compliance_frameworks = {}
     # Load compliance frameworks
     logger.debug("Loading compliance frameworks from .json files")
@@ -256,6 +264,7 @@ def prowler():
         severities=severities,
         compliance_frameworks=compliance_framework,
         categories=categories,
+        resource_groups=resource_groups,
         provider=provider,
     )
 

prowler/lib/check/check.py

@@ -228,6 +228,28 @@ def print_categories(categories: set):
     print(message)
 
 
+def list_resource_groups(bulk_checks_metadata: dict) -> set:
+    available_resource_groups = set()
+    for check in bulk_checks_metadata.values():
+        if check.ResourceGroup:
+            available_resource_groups.add(check.ResourceGroup)
+    return available_resource_groups
+
+
+def print_resource_groups(resource_groups: set):
+    rg_num = len(resource_groups)
+    plural_string = f"\nThere are {Fore.YELLOW}{rg_num}{Style.RESET_ALL} available resource groups.\n"
+    singular_string = (
+        f"\nThere is {Fore.YELLOW}{rg_num}{Style.RESET_ALL} available resource group.\n"
+    )
+
+    message = plural_string if rg_num > 1 else singular_string
+    for rg in sorted(resource_groups):
+        print(f"- {rg}")
+
+    print(message)
+
+
 def print_services(service_list: set):
     services_num = len(service_list)
     plural_string = f"\nThere are {Fore.YELLOW}{services_num}{Style.RESET_ALL} available services.\n"

prowler/lib/check/checks_loader.py

@@ -19,6 +19,7 @@ def load_checks_to_execute(
     severities: list = None,
     compliance_frameworks: list = None,
     categories: set = None,
+    resource_groups: set = None,
 ) -> set:
     """Generate the list of checks to execute based on the cloud provider and the input arguments given"""
     try:
@@ -30,6 +31,7 @@ def load_checks_to_execute(
         checks_to_execute = set()
         check_aliases = {}
         check_categories = {}
+        check_resource_groups = {}
         check_severities = {severity.value: [] for severity in Severity}
 
         if not bulk_checks_metadata:
@@ -52,6 +54,13 @@ def load_checks_to_execute(
                     if category not in check_categories:
                         check_categories[category] = []
                     check_categories[category].append(check)
+
+                # Resource Groups (stored lowercase for case-insensitive matching)
+                if metadata.ResourceGroup:
+                    rg_key = metadata.ResourceGroup.lower()
+                    if rg_key not in check_resource_groups:
+                        check_resource_groups[rg_key] = []
+                    check_resource_groups[rg_key].append(check)
             except Exception as error:
                 logger.error(
                     f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
@@ -170,6 +179,28 @@ def load_checks_to_execute(
             for category in categories:
                 checks_to_execute.update(check_categories[category])
 
+        # Handle if there are resource groups passed using --resource-group
+        elif resource_groups:
+            # Validate that all resource groups exist (case-insensitive)
+            available_resource_groups = set(check_resource_groups.keys())
+            normalized_resource_groups = [rg.lower() for rg in resource_groups]
+            invalid_resource_groups = [
+                rg
+                for rg in normalized_resource_groups
+                if rg not in available_resource_groups
+            ]
+            if invalid_resource_groups:
+                logger.critical(
+                    f"Invalid resource group(s) specified: {', '.join(invalid_resource_groups)}"
+                )
+                logger.critical(
+                    f"Please provide valid resource group names. Use 'prowler {provider} --list-resource-groups' to see available resource groups."
+                )
+                sys.exit(1)
+
+            for resource_group in normalized_resource_groups:
+                checks_to_execute.update(check_resource_groups[resource_group])
+
         # If there are no checks passed as argument
         else:
             # get all checks

prowler/lib/cli/parser.py

@@ -322,6 +322,13 @@ Detailed documentation at https://docs.prowler.com
             default=[],
             # TODO: Pending validate choices
         )
+        group.add_argument(
+            "--resource-group",
+            "--resource-groups",
+            nargs="+",
+            help="List of resource groups to be executed.",
+            default=[],
+        )
         common_checks_parser.add_argument(
             "--checks-folder",
             "-x",
@@ -332,7 +339,7 @@ Detailed documentation at https://docs.prowler.com
     def __init_list_checks_parser__(self):
         # List checks options
         list_checks_parser = self.common_providers_parser.add_argument_group(
-            "List checks/services/categories/compliance-framework checks"
+            "List checks/services/categories/resource-groups/compliance-framework checks"
         )
         list_group = list_checks_parser.add_mutually_exclusive_group()
         list_group.add_argument(
@@ -365,6 +372,11 @@ Detailed documentation at https://docs.prowler.com
             action="store_true",
             help="List the available check's categories",
         )
+        list_group.add_argument(
+            "--list-resource-groups",
+            action="store_true",
+            help="List the available check's resource groups",
+        )
         list_group.add_argument(
             "--list-fixer",
             "--list-fixers",

tests/lib/check/check_loader_test.py

@@ -40,6 +40,7 @@ class TestCheckLoader:
             ResourceIdTemplate="arn:partition:s3:::bucket_name",
             Severity=S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY,
             ResourceType="AwsS3Bucket",
+            ResourceGroup="storage",
             Description="Check S3 Bucket Level Public Access Block.",
             Risk="Public access policies may be applied to sensitive data buckets.",
             RelatedUrl="",
@@ -76,6 +77,7 @@ class TestCheckLoader:
             ResourceIdTemplate="arn:partition:iam::account-id:user/user_name",
             Severity=IAM_USER_NO_MFA_SEVERITY,
             ResourceType="AwsIamUser",
+            ResourceGroup="IAM",
             Description="Check IAM User No MFA.",
             Risk="IAM users should have Multi-Factor Authentication (MFA) enabled.",
             RelatedUrl="",
@@ -533,3 +535,97 @@ class TestCheckLoader:
                 provider=self.provider,
             )
         assert exc_info.value.code == 1
+
+    def test_load_checks_to_execute_with_resource_groups(self):
+        """Test that checks are filtered by resource group"""
+        bulk_checks_metadata = {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata(),
+            IAM_USER_NO_MFA_NAME: self.get_custom_check_iam_metadata(),
+        }
+        resource_groups = {"storage"}
+
+        assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
+            bulk_checks_metadata=bulk_checks_metadata,
+            resource_groups=resource_groups,
+            provider=self.provider,
+        )
+
+    def test_load_checks_to_execute_with_multiple_resource_groups(self):
+        """Test that checks are filtered by multiple resource groups"""
+        bulk_checks_metadata = {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata(),
+            IAM_USER_NO_MFA_NAME: self.get_custom_check_iam_metadata(),
+        }
+        resource_groups = {"storage", "IAM"}
+
+        assert {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME,
+            IAM_USER_NO_MFA_NAME,
+        } == load_checks_to_execute(
+            bulk_checks_metadata=bulk_checks_metadata,
+            resource_groups=resource_groups,
+            provider=self.provider,
+        )
+
+    def test_load_checks_to_execute_with_resource_group_case_insensitive(self):
+        """Test that resource group matching is case-insensitive"""
+        bulk_checks_metadata = {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata(),
+            IAM_USER_NO_MFA_NAME: self.get_custom_check_iam_metadata(),
+        }
+        # "iam" lowercase should match metadata "IAM", "Storage" mixed case should match "storage"
+        resource_groups = {"iam", "Storage"}
+
+        assert {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME,
+            IAM_USER_NO_MFA_NAME,
+        } == load_checks_to_execute(
+            bulk_checks_metadata=bulk_checks_metadata,
+            resource_groups=resource_groups,
+            provider=self.provider,
+        )
+
+    def test_load_checks_to_execute_with_invalid_resource_group(self):
+        """Test that invalid resource group names cause sys.exit(1)"""
+        bulk_checks_metadata = {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
+        }
+        resource_groups = {"invalid_resource_group"}
+
+        with pytest.raises(SystemExit) as exc_info:
+            load_checks_to_execute(
+                bulk_checks_metadata=bulk_checks_metadata,
+                resource_groups=resource_groups,
+                provider=self.provider,
+            )
+        assert exc_info.value.code == 1
+
+    def test_load_checks_to_execute_with_multiple_invalid_resource_groups(self):
+        """Test that multiple invalid resource group names cause sys.exit(1)"""
+        bulk_checks_metadata = {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
+        }
+        resource_groups = {"invalid_rg_1", "invalid_rg_2"}
+
+        with pytest.raises(SystemExit) as exc_info:
+            load_checks_to_execute(
+                bulk_checks_metadata=bulk_checks_metadata,
+                resource_groups=resource_groups,
+                provider=self.provider,
+            )
+        assert exc_info.value.code == 1
+
+    def test_load_checks_to_execute_with_mixed_valid_invalid_resource_groups(self):
+        """Test that mix of valid and invalid resource groups cause sys.exit(1)"""
+        bulk_checks_metadata = {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
+        }
+        resource_groups = {"storage", "invalid_resource_group"}
+
+        with pytest.raises(SystemExit) as exc_info:
+            load_checks_to_execute(
+                bulk_checks_metadata=bulk_checks_metadata,
+                resource_groups=resource_groups,
+                provider=self.provider,
+            )
+        assert exc_info.value.code == 1

tests/lib/cli/parser_test.py

@@ -81,6 +81,7 @@ class Test_Parser:
         assert not parsed.severity
         assert not parsed.compliance
         assert len(parsed.category) == 0
+        assert len(parsed.resource_group) == 0
         assert not parsed.excluded_check
         assert not parsed.excluded_service
         assert not parsed.excluded_checks_file
@@ -89,6 +90,7 @@ class Test_Parser:
         assert not parsed.list_compliance
         assert not parsed.list_compliance_requirements
         assert not parsed.list_categories
+        assert not parsed.list_resource_groups
         assert not parsed.profile
         assert not parsed.role
         assert parsed.session_duration == 3600
@@ -131,6 +133,7 @@ class Test_Parser:
         assert not parsed.severity
         assert not parsed.compliance
         assert len(parsed.category) == 0
+        assert len(parsed.resource_group) == 0
         assert not parsed.excluded_check
         assert not parsed.excluded_service
         assert not parsed.excluded_checks_file
@@ -139,6 +142,7 @@ class Test_Parser:
         assert not parsed.list_compliance
         assert not parsed.list_compliance_requirements
         assert not parsed.list_categories
+        assert not parsed.list_resource_groups
         assert len(parsed.subscription_id) == 0
         assert not parsed.az_cli_auth
         assert parsed.sp_env_auth
@@ -173,6 +177,7 @@ class Test_Parser:
         assert not parsed.severity
         assert not parsed.compliance
         assert len(parsed.category) == 0
+        assert len(parsed.resource_group) == 0
         assert not parsed.excluded_check
         assert not parsed.excluded_service
         assert not parsed.excluded_checks_file
@@ -181,6 +186,7 @@ class Test_Parser:
         assert not parsed.list_compliance
         assert not parsed.list_compliance_requirements
         assert not parsed.list_categories
+        assert not parsed.list_resource_groups
         assert not parsed.credentials_file
 
     def test_default_parser_no_arguments_kubernetes(self):
@@ -210,6 +216,7 @@ class Test_Parser:
         assert not parsed.severity
         assert not parsed.compliance
         assert len(parsed.category) == 0
+        assert len(parsed.resource_group) == 0
         assert not parsed.excluded_check
         assert not parsed.excluded_service
         assert not parsed.excluded_checks_file
@@ -218,6 +225,7 @@ class Test_Parser:
         assert not parsed.list_compliance
         assert not parsed.list_compliance_requirements
         assert not parsed.list_categories
+        assert not parsed.list_resource_groups
         assert parsed.kubeconfig_file == "~/.kube/config"
         assert not parsed.context
         assert not parsed.namespace
@@ -723,6 +731,32 @@ class Test_Parser:
         assert category_1 in parsed.category
         assert category_2 in parsed.category
 
+    def test_checks_parser_resource_group(self):
+        argument = "--resource-group"
+        resource_group = "storage"
+        command = [prowler_command, argument, resource_group]
+        parsed = self.parser.parse(command)
+        assert len(parsed.resource_group) == 1
+        assert resource_group in parsed.resource_group
+
+    def test_checks_parser_resource_groups_alias(self):
+        argument = "--resource-groups"
+        resource_group = "storage"
+        command = [prowler_command, argument, resource_group]
+        parsed = self.parser.parse(command)
+        assert len(parsed.resource_group) == 1
+        assert resource_group in parsed.resource_group
+
+    def test_checks_parser_resource_groups_two(self):
+        argument = "--resource-group"
+        resource_group_1 = "storage"
+        resource_group_2 = "compute"
+        command = [prowler_command, argument, resource_group_1, resource_group_2]
+        parsed = self.parser.parse(command)
+        assert len(parsed.resource_group) == 2
+        assert resource_group_1 in parsed.resource_group
+        assert resource_group_2 in parsed.resource_group
+
     def test_list_checks_parser_list_checks_short(self):
         argument = "-l"
         command = [prowler_command, argument]
@@ -759,6 +793,12 @@ class Test_Parser:
         parsed = self.parser.parse(command)
         assert parsed.list_categories
 
+    def test_list_checks_parser_list_resource_groups(self):
+        argument = "--list-resource-groups"
+        command = [prowler_command, argument]
+        parsed = self.parser.parse(command)
+        assert parsed.list_resource_groups
+
     def test_list_checks_parser_list_fixers(self):
         argument = "--list-fixers"
         command = [prowler_command, argument]