Merge branch 'master' into prowler-inventory

docs/tutorials/scan-inventory.md

@@ -0,0 +1,39 @@
+# Scan Inventory
+
+The scan-inventory feature is a tool that generates a JSON report within the `/output/inventory/<provider>` directory and the scanned service. This feature allows you to perform a inventory of the resources existing in your provider that are scanned by Prowler.
+
+## Usage
+
+To use the scan-inventory feature, run Prowler with the `--scan-inventory` option. For example:
+
+```
+prowler <provider> --scan-inventory
+```
+
+This will generate a JSON report within the `/output/inventory/<provider>` directory and the scanned service.
+
+## Output Directory Contents
+
+The contents of the `/output/<provider>` directory and the scanned service depend on the Prowler execution. This directory contains all the information gathered during scanning, including a JSON report containing all the gathered information.
+
+## Limitations
+
+The scan-inventory feature has some limitations. For example:
+
+* It is only available for the AWS provider.
+* It only contains the information retrieved by Prowler during the execution.
+
+## Example
+
+Here's an example of how to use the scan-inventory feature and the contents of the `/output/inventory/<provider>` directory and the scanned service:
+
+`prowler aws -s ec2 --scan-inventory`
+
+```
+/output/inventory/aws directory
+   |
+   |-- ec2
+   |    |
+   |    |-- ec2_output.json
+```
+In this example, Prowler is run with the `-s ec2` and `--scan-inventory` options for the AWS provider. The `/output/inventory/aws` directory contains a JSON report showing all the information gathered during scanning.

mkdocs.yml

@@ -55,6 +55,7 @@ nav:
       - Dashboard: tutorials/dashboard.md
       - Fixer (remediations): tutorials/fixer.md
       - Quick Inventory: tutorials/quick-inventory.md
+      - Scan Inventory: tutorials/scan-inventory.md
       - Slack Integration: tutorials/integrations.md
       - Configuration File: tutorials/configuration_file.md
       - Logging: tutorials/logging.md

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand.
 
 [[package]]
 name = "about-time"

prowler/__main__.py

@@ -73,6 +73,7 @@ from prowler.providers.aws.models import AWSOutputOptions
 from prowler.providers.azure.models import AzureOutputOptions
 from prowler.providers.common.provider import Provider
 from prowler.providers.common.quick_inventory import run_provider_quick_inventory
+from prowler.providers.common.scan_inventory import run_prowler_scan_inventory
 from prowler.providers.gcp.models import GCPOutputOptions
 from prowler.providers.kubernetes.models import KubernetesOutputOptions
 
@@ -688,6 +689,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_scan_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/providers/aws/lib/arguments/arguments.py

@@ -100,6 +100,13 @@ def init_parser(self):
         action="store_true",
         help="Run Prowler Quick Inventory. The inventory will be stored in an output csv by default",
     )
+    # AWS Scan Inventory
+    aws_scan_inventory_subparser = aws_parser.add_argument_group("Scan Inventory")
+    aws_scan_inventory_subparser.add_argument(
+        "--scan-inventory",
+        action="store_true",
+        help="Run Prowler Scan 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")
     aws_outputs_bucket_parser = aws_outputs_subparser.add_mutually_exclusive_group()

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 @@ class AWSService:
             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/scan_inventory.py

@@ -0,0 +1,104 @@
+import importlib
+import json
+import os
+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_scan_inventory(checks_to_execute, provider):
+    output_folder_path = f"./output/inventory/{provider}"
+
+    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}")
+
+                # Convert to JSON
+                output_file = service_client.split("_client")[0]
+
+                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)
+    print(
+        f"\n{Style.BRIGHT}{Fore.GREEN}Scan inventory for {provider} results: {orange_color}{output_folder_path}"
+    )

tests/lib/cli/parser_test.py

@@ -979,6 +979,12 @@ class Test_Parser:
         parsed = self.parser.parse(command)
         assert parsed.quick_inventory
 
+    def test_aws_parser_scan_inventory_long(self):
+        argument = "--scan-inventory"
+        command = [prowler_command, argument]
+        parsed = self.parser.parse(command)
+        assert parsed.scan_inventory
+
     def test_aws_parser_output_bucket_short(self):
         argument = "-B"
         bucket = "test-bucket"