chore(parser): add --subservices tag

prowler/__main__.py

@@ -18,6 +18,7 @@ from prowler.lib.check.check import (
     list_checks_json,
     list_fixers,
     list_services,
+    list_subservices,
     parse_checks_from_folder,
     print_categories,
     print_checks,
@@ -25,6 +26,7 @@ from prowler.lib.check.check import (
     print_compliance_requirements,
     print_fixers,
     print_services,
+    print_subservices,
     remove_custom_checks_module,
     run_fixer,
 )
@@ -69,6 +71,7 @@ def prowler():
     excluded_checks = args.excluded_check
     excluded_services = args.excluded_service
     services = args.service
+    subservices = args.subservice
     categories = args.category
     checks_file = args.checks_file
     checks_folder = args.checks_folder
@@ -78,6 +81,7 @@ def prowler():
     default_execution = (
         not checks
         and not services
+        and not subservices
         and not categories
         and not excluded_checks
         and not excluded_services
@@ -102,7 +106,9 @@ def prowler():
     if args.list_services:
         print_services(list_services(provider))
         sys.exit()
-
+    if args.list_subservices:
+        print_subservices(list_subservices(provider))
+        sys.exit()
     if args.list_fixer:
         print_fixers(list_fixers(provider))
         sys.exit()
@@ -151,6 +157,7 @@ def prowler():
         checks_file,
         checks,
         services,
+        subservices,
         severities,
         compliance_framework,
         categories,

prowler/lib/check/check.py

@@ -204,6 +204,18 @@ def list_services(provider: str) -> set:
     return sorted(available_services)
 
 
+def list_subservices(provider: str) -> set:
+    available_subservices = set()
+    checks_tuple = recover_checks_from_provider(provider)
+    for _, check_path in checks_tuple:
+        check_name = check_path.split("/")[-1]
+        check_path = f"{check_path}/{check_name}.metadata.json"
+        check_metadata = load_check_metadata(check_path)
+        if check_metadata.SubServiceName:
+            available_subservices.add(check_metadata.SubServiceName)
+    return sorted(available_subservices)
+
+
 def list_fixers(provider: str) -> set:
     available_fixers = set()
     checks = recover_checks_from_provider(provider, include_fixers=True)
@@ -256,6 +268,19 @@ def print_services(service_list: set):
     print(message)
 
 
+def print_subservices(sub_service_list: set):
+    subservices_num = len(sub_service_list)
+    plural_string = f"\nThere are {Fore.YELLOW}{subservices_num}{Style.RESET_ALL} available subservices.\n"
+    singular_string = f"\nThere is {Fore.YELLOW}{subservices_num}{Style.RESET_ALL} available subservice.\n"
+
+    message = plural_string if subservices_num > 1 else singular_string
+
+    for subservice in sub_service_list:
+        print(f"- {subservice}")
+
+    print(message)
+
+
 def print_fixers(fixers_list: set):
     fixers_num = len(fixers_list)
     plural_string = (
@@ -794,3 +819,33 @@ def recover_checks_from_service(service_list: list, provider: str) -> set:
         logger.error(
             f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
         )
+
+
+def recover_checks_from_subservice(sub_service_list: list, provider: str) -> set:
+    """
+    Recover all checks from the selected provider and subservices
+
+    Returns a set of checks from the given subservices
+    """
+    # Get all the services available for the provider
+    try:
+        checks = set()
+        available_services = list_services(provider)
+        available_subservices = list_subservices(provider)
+        # Check if the subservice is valid
+        for sub_service in sub_service_list:
+            if sub_service not in available_subservices:
+                logger.error(f"SubService '{sub_service}' does not have checks.")
+                continue
+        checks_names = recover_checks_from_service(available_services, provider)
+        for sub_service in sub_service_list:
+            for check in checks_names:
+                check_name = check.split("_")[1]
+                if check_name == sub_service:
+                    checks.add(check)
+        return checks
+
+    except Exception as error:
+        logger.error(
+            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+        )

prowler/lib/check/checks_loader.py

@@ -6,6 +6,7 @@ from prowler.lib.check.check import (
     parse_checks_from_file,
     recover_checks_from_provider,
     recover_checks_from_service,
+    recover_checks_from_subservice,
 )
 from prowler.lib.logger import logger
 
@@ -17,6 +18,7 @@ def load_checks_to_execute(
     checks_file: str,
     check_list: list,
     service_list: list,
+    sub_service_list: list,
     severities: list,
     compliance_frameworks: list,
     categories: set,
@@ -77,6 +79,12 @@ def load_checks_to_execute(
         elif service_list:
             checks_to_execute = recover_checks_from_service(service_list, provider)
 
+        # Handle if there are sub-services passed using -u/--sub-services
+        elif sub_service_list:
+            checks_to_execute = recover_checks_from_subservice(
+                sub_service_list, provider
+            )
+
         # Handle if there are compliance frameworks passed using --compliance
         elif compliance_frameworks:
             checks_to_execute = parse_checks_from_compliance_framework(

prowler/lib/cli/parser.py

@@ -252,6 +252,13 @@ Detailed documentation at https://docs.prowler.com
             nargs="+",
             help="List of services to be executed.",
         )
+        group.add_argument(
+            "--subservice",
+            "--subservices",
+            "-u",
+            nargs="+",
+            help="List of subservices to be executed.",
+        )
         common_checks_parser.add_argument(
             "--severity",
             "--severities",
@@ -299,6 +306,11 @@ Detailed documentation at https://docs.prowler.com
             action="store_true",
             help="List covered services by given provider",
         )
+        list_group.add_argument(
+            "--list-subservices",
+            action="store_true",
+            help="List covered subservices by given provider",
+        )
         list_group.add_argument(
             "--list-compliance",
             "--list-compliances",

prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306.metadata.json

@@ -6,7 +6,7 @@
     "Infrastructure Security"
   ],
   "ServiceName": "ec2",
-  "SubServiceName": "securitygroups",
+  "SubServiceName": "securitygroup",
   "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
   "Severity": "high",
   "ResourceType": "AwsEc2SecurityGroup",

tests/lib/check/check_test.py

@@ -20,6 +20,7 @@ from prowler.lib.check.check import (
     parse_checks_from_folder,
     recover_checks_from_provider,
     recover_checks_from_service,
+    recover_checks_from_subservice,
     remove_custom_checks_module,
     update_audit_metadata,
 )
@@ -645,6 +646,20 @@ class TestCheck:
         recovered_checks = recover_checks_from_service(service_list, provider)
         assert recovered_checks == expected_checks
 
+    @patch(
+        "prowler.lib.check.check.recover_checks_from_provider",
+        new=mock_recover_checks_from_aws_provider,
+    )
+    def test_recover_checks_from_subservice(self):
+        subservice_list = ["securitygroup"]
+        provider = "aws"
+        expected_checks = {
+            "ec2_securitygroup_allow_ingress_from_internet_to_any_port",
+        }
+
+        recovered_checks = recover_checks_from_subservice(subservice_list, provider)
+        assert recovered_checks == expected_checks
+
     # def test_parse_checks_from_compliance_framework_two(self):
     #     test_case = {
     #         "input": {"compliance_frameworks": ["cis_v1.4_aws", "ens_v3_aws"]},

tests/lib/cli/parser_test.py

@@ -693,6 +693,12 @@ class Test_Parser:
         parsed = self.parser.parse(command)
         assert parsed.list_services
 
+    def test_list_checks_parser_list_subservices(self):
+        argument = "--list-subservices"
+        command = [prowler_command, argument]
+        parsed = self.parser.parse(command)
+        assert parsed.list_subservices
+
     def test_list_checks_parser_list_compliance(self):
         argument = "--list-compliance"
         command = [prowler_command, argument]