feat(inventory): add prowler scanner-inventory(auditor mode)

docs/tutorials/quick-inventory.md

@@ -6,7 +6,7 @@ Prowler allows you to execute a quick inventory to extract the number of resourc
     Currently, it is only available for AWS provider.
 
 
-- You can use option `-i`/`--quick-inventory` to execute it:
+- You can use option `-i`/`--scan-inventory` to execute it:
 ```sh
 prowler <provider> -i
 ```

prowler/__main__.py

@@ -71,8 +71,8 @@
 from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
 from prowler.providers.aws.models import AWSOutputOptions
 from prowler.providers.azure.models import AzureOutputOptions
+from prowler.providers.common.inventory import run_prowler_inventory
 from prowler.providers.common.provider import Provider
-from prowler.providers.common.quick_inventory import run_provider_quick_inventory
 from prowler.providers.gcp.models import GCPOutputOptions
 from prowler.providers.kubernetes.models import KubernetesOutputOptions
 
@@ -257,11 +257,6 @@ def prowler():
             args, bulk_checks_metadata, global_provider.identity
         )
 
-    # Run the quick inventory for the provider if available
-    if hasattr(args, "quick_inventory") and args.quick_inventory:
-        run_provider_quick_inventory(global_provider, args)
-        sys.exit()
-
     # Execute checks
     findings = []
 
@@ -688,6 +683,11 @@ def prowler():
     if checks_folder:
         remove_custom_checks_module(checks_folder, provider)
 
+    # Run the quick inventory for the provider if available
+    if hasattr(args, "scan_inventory") and args.scan_inventory:
+        run_prowler_inventory(checks_to_execute, args.provider)
+        sys.exit()
+
     # If there are failed findings exit code 3, except if -z is input
     if (
         not args.ignore_exit_code_3

prowler/lib/cli/parser.py

@@ -378,3 +378,18 @@ def __init_third_party_integrations_parser__(self):
             action="store_true",
             help="Send a summary of the execution with a Slack APP in your channel. Environment variables SLACK_API_TOKEN and SLACK_CHANNEL_NAME are required (see more in https://docs.prowler.cloud/en/latest/tutorials/integrations/#slack).",
         )
+
+    def __init_inventory_parser__(self):
+        inventory_parser = self.common_providers_parser.add_argument_group(
+            "ScanInventory"
+        )
+        inventory_parser.add_argument(
+            "--scan-inventory",
+            action="store_true",
+            help="Run Prowler in inventory mode",
+        )
+        inventory_parser.add_argument(
+            "-i",
+            nargs="?",
+            help="Output folder path for the inventory",
+        )

prowler/providers/aws/lib/arguments/arguments.py

@@ -93,12 +93,12 @@ def init_parser(self):
         help="Send only Prowler failed findings to SecurityHub",
     )
     # AWS Quick Inventory
-    aws_quick_inventory_subparser = aws_parser.add_argument_group("Quick Inventory")
+    aws_quick_inventory_subparser = aws_parser.add_argument_group("Inventory")
     aws_quick_inventory_subparser.add_argument(
-        "--quick-inventory",
+        "--scan-inventory",
         "-i",
         action="store_true",
-        help="Run Prowler Quick Inventory. The inventory will be stored in an output csv by default",
+        help="Run Prowler Inventory. The inventory will be stored in an output json file.",
     )
     # AWS Outputs
     aws_outputs_subparser = aws_parser.add_argument_group("AWS Outputs to S3")

prowler/providers/aws/lib/service/service.py

@@ -1,4 +1,7 @@
+from collections import deque
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime
+from typing import Any, Dict
 
 from prowler.lib.logger import logger
 from prowler.providers.aws.aws_provider import AwsProvider
@@ -101,3 +104,47 @@ def __threading_call__(self, call, iterator=None):
             except Exception:
                 # Handle exceptions if necessary
                 pass  # Replace 'pass' with any additional exception handling logic. Currently handled within the called function
+
+    def __to_dict__(self, seen=None) -> Dict[str, Any]:
+        if seen is None:
+            seen = set()
+
+        def convert_value(value):
+            if isinstance(value, (AwsProvider,)):
+                return {}
+            if isinstance(value, datetime):
+                return value.isoformat()  # Convert datetime to ISO 8601 string
+            elif isinstance(value, deque):
+                return [convert_value(item) for item in value]
+            elif isinstance(value, list):
+                return [convert_value(item) for item in value]
+            elif isinstance(value, tuple):
+                return tuple(convert_value(item) for item in value)
+            elif isinstance(value, dict):
+                # Ensure keys are strings and values are processed
+                return {
+                    convert_value(str(k)): convert_value(v) for k, v in value.items()
+                }
+            elif hasattr(value, "__dict__"):
+                obj_id = id(value)
+                if obj_id in seen:
+                    return None  # Avoid infinite recursion
+                seen.add(obj_id)
+                return {key: convert_value(val) for key, val in value.__dict__.items()}
+            else:
+                return value  # Handle basic types and non-serializable objects
+
+        return {
+            key: convert_value(value)
+            for key, value in self.__dict__.items()
+            if key
+            not in [
+                "audit_config",
+                "provider",
+                "session",
+                "regional_clients",
+                "client",
+                "thread_pool",
+                "fixer_config",
+            ]
+        }

prowler/providers/common/inventory.py

@@ -0,0 +1,123 @@
+import importlib
+import json
+import os
+import shutil
+from collections import deque
+from datetime import datetime
+
+from colorama import Fore, Style
+from pydantic import BaseModel
+
+from prowler.config.config import orange_color
+
+
+def run_prowler_inventory(checks_to_execute, provider):
+    output_folder_path = f"./output/inventory/{provider}"
+    meta_json_file = {}
+
+    os.makedirs(output_folder_path, exist_ok=True)
+
+    # Recursive function to handle serialization
+    def class_to_dict(obj, seen=None):
+        if seen is None:
+            seen = set()
+
+        if isinstance(obj, dict):
+            new_dict = {}
+            for key, value in obj.items():
+                if isinstance(key, tuple):
+                    key = str(key)  # Convert tuple to string
+                new_dict[key] = class_to_dict(value)
+            return new_dict
+        if isinstance(obj, datetime):
+            return obj.isoformat()
+        elif isinstance(obj, deque):
+            return list(class_to_dict(item, seen) for item in obj)
+        elif isinstance(obj, BaseModel):
+            return obj.dict()
+        elif isinstance(obj, (list, tuple)):
+            return [class_to_dict(item, seen) for item in obj]
+        elif hasattr(obj, "__dict__") and id(obj) not in seen:
+            seen.add(id(obj))
+            return {
+                key: class_to_dict(value, seen) for key, value in obj.__dict__.items()
+            }
+        else:
+            return obj
+
+    service_set = set()
+
+    for check_name in checks_to_execute:
+        try:
+            service = check_name.split("_")[0]
+
+            if service in service_set:
+                continue
+
+            service_set.add(service)
+
+            service_path = f"./prowler/providers/{provider}/services/{service}"
+
+            # List to store all _client filenames
+            client_files = []
+
+            # Walk through the directory and find all files
+            for root, dirs, files in os.walk(service_path):
+                for file in files:
+                    if file.endswith("_client.py"):
+                        # Append only the filename to the list (not the full path)
+                        client_files.append(file)
+
+            service_output_folder = f"{output_folder_path}/{service}"
+
+            os.makedirs(service_output_folder, exist_ok=True)
+
+            for service_client in client_files:
+
+                service_client = service_client.split(".py")[0]
+                check_module_path = (
+                    f"prowler.providers.{provider}.services.{service}.{service_client}"
+                )
+
+                try:
+                    lib = importlib.import_module(f"{check_module_path}")
+                except ModuleNotFoundError:
+                    print(f"Module not found: {check_module_path}")
+                    break
+                except Exception as e:
+                    print(f"Error while importing module {check_module_path}: {e}")
+                    break
+
+                client_path = getattr(lib, f"{service_client}")
+
+                if not meta_json_file.get(f"{service}"):
+                    meta_json_file[f"{service}"] = []
+
+                # Convert to JSON
+                output_file = service_client.split("_client")[0]
+
+                meta_json_file[f"{service}"].append(
+                    f"./{service}/{output_file}_output.json"
+                )
+
+                with open(
+                    f"{service_output_folder}/{output_file}_output.json", "w+"
+                ) as fp:
+                    output = client_path.__to_dict__()
+                    json.dump(output, fp=fp, default=str, indent=4)
+
+        except Exception as e:
+            print("Exception: ", e)
+
+    with open(f"{output_folder_path}/output_metadata.json", "w+") as fp:
+        json.dump(meta_json_file, fp=fp, default=str, indent=4)
+
+    # end of all things
+    folder_to_compress = f"{output_folder_path}"
+    output_zip_file = f"{output_folder_path}/rmfx-scan-compressed"  # The output file (without extension)
+
+    # Compress the folder into a zip file
+    shutil.make_archive(f"{output_zip_file}", "zip", folder_to_compress)
+    print(
+        f"\n{Style.BRIGHT}{Fore.GREEN}Scan inventory for {provider} results: {orange_color}{output_folder_path}"
+    )

tests/lib/cli/parser_test.py

@@ -974,7 +974,7 @@ def test_aws_parser_quick_inventory_short(self):
         assert parsed.quick_inventory
 
     def test_aws_parser_quick_inventory_long(self):
-        argument = "--quick-inventory"
+        argument = "--scan-inventory"
         command = [prowler_command, argument]
         parsed = self.parser.parse(command)
         assert parsed.quick_inventory