feat(outputs): add AWS inventory connectivity graph output format

prowler/CHANGELOG.md

@@ -6,6 +6,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 
 ### 🚀 Added
 
+- `inventory-graph` output format: produces a `.inventory.json` machine-readable graph and an interactive `.inventory.html` D3.js force-directed connectivity map of all scanned AWS resources and their relationships (VPC, IAM trust, event triggers, KMS encryption, etc.) [(#10382)](https://github.com/prowler-cloud/prowler/pull/10382)
 - `misconfig` scanner as default for Image provider scans [(#10167)](https://github.com/prowler-cloud/prowler/pull/10167)
 - `entra_conditional_access_policy_device_code_flow_blocked` check for M365 provider [(#10218)](https://github.com/prowler-cloud/prowler/pull/10218)
 - RBI compliance for the Azure provider [(#10339)](https://github.com/prowler-cloud/prowler/pull/10339)

prowler/__main__.py

@@ -15,6 +15,7 @@
     csv_file_suffix,
     get_available_compliance_frameworks,
     html_file_suffix,
+    inventory_graph_file_suffix,
     json_asff_file_suffix,
     json_ocsf_file_suffix,
     orange_color,
@@ -528,6 +529,14 @@ def streaming_callback(findings_batch):
                 html_output.batch_write_data_to_file(
                     provider=global_provider, stats=stats
                 )
+            if mode == "inventory-graph":
+                from prowler.lib.outputs.inventory.inventory_output import (
+                    generate_inventory_outputs,
+                )
+
+                generate_inventory_outputs(
+                    f"{filename}{inventory_graph_file_suffix}"
+                )
 
     if getattr(args, "push_to_cloud", False):
         if not ocsf_output or not getattr(ocsf_output, "file_path", None):

prowler/config/config.py

@@ -109,6 +109,7 @@ def get_available_compliance_frameworks(provider=None):
 json_asff_file_suffix = ".asff.json"
 json_ocsf_file_suffix = ".ocsf.json"
 html_file_suffix = ".html"
+inventory_graph_file_suffix = ".inventory"
 default_config_file_path = (
     f"{pathlib.Path(os.path.dirname(os.path.realpath(__file__)))}/config.yaml"
 )
@@ -119,7 +120,7 @@ def get_available_compliance_frameworks(provider=None):
     f"{pathlib.Path(os.path.dirname(os.path.realpath(__file__)))}/llm_config.yaml"
 )
 encoding_format_utf_8 = "utf-8"
-available_output_formats = ["csv", "json-asff", "json-ocsf", "html"]
+available_output_formats = ["csv", "json-asff", "json-ocsf", "html", "inventory-graph"]
 
 # Prowler Cloud API settings
 cloud_api_base_url = os.getenv("PROWLER_CLOUD_API_BASE_URL", "https://api.prowler.com")

prowler/lib/outputs/inventory/extractors/ec2_extractor.py

@@ -0,0 +1,92 @@
+from typing import List, Tuple
+
+from prowler.lib.outputs.inventory.models import ResourceEdge, ResourceNode
+
+
+def extract(client) -> Tuple[List[ResourceNode], List[ResourceEdge]]:
+    """
+    Extract EC2 instance and security-group nodes with their edges.
+
+    Edges produced:
+      - instance → security-group  [network]
+      - instance → subnet          [network]
+      - security-group → VPC       [network]
+    """
+    nodes: List[ResourceNode] = []
+    edges: List[ResourceEdge] = []
+
+    # EC2 Instances
+    for instance in client.instances:
+        name = instance.id
+        for tag in instance.tags or []:
+            if tag.get("Key") == "Name":
+                name = tag["Value"]
+                break
+
+        props = {
+            "instance_type": getattr(instance, "type", None),
+            "state": getattr(instance, "state", None),
+            "vpc_id": getattr(instance, "vpc_id", None),
+            "subnet_id": getattr(instance, "subnet_id", None),
+            "public_ip": getattr(instance, "public_ip_address", None),
+            "private_ip": getattr(instance, "private_ip_address", None),
+        }
+
+        nodes.append(
+            ResourceNode(
+                id=instance.arn,
+                type="ec2_instance",
+                name=name,
+                service="ec2",
+                region=instance.region,
+                account_id=client.audited_account,
+                properties={k: v for k, v in props.items() if v is not None},
+            )
+        )
+
+        for sg_id in instance.security_groups or []:
+            edges.append(
+                ResourceEdge(
+                    source_id=instance.arn,
+                    target_id=sg_id,
+                    edge_type="network",
+                    label="sg",
+                )
+            )
+
+        if instance.subnet_id:
+            edges.append(
+                ResourceEdge(
+                    source_id=instance.arn,
+                    target_id=instance.subnet_id,
+                    edge_type="network",
+                    label="subnet",
+                )
+            )
+
+    # Security Groups
+    for sg in client.security_groups.values():
+        name = sg.name if hasattr(sg, "name") else sg.id if hasattr(sg, "id") else sg.arn
+        nodes.append(
+            ResourceNode(
+                id=sg.arn,
+                type="security_group",
+                name=name,
+                service="ec2",
+                region=sg.region,
+                account_id=client.audited_account,
+                properties={"vpc_id": sg.vpc_id},
+            )
+        )
+
+        if sg.vpc_id:
+            edges.append(
+                ResourceEdge(
+                    source_id=sg.arn,
+                    target_id=sg.vpc_id,
+                    edge_type="network",
+                    label="in-vpc",
+                )
+            )
+
+    return nodes, edges

prowler/lib/outputs/inventory/extractors/elbv2_extractor.py

@@ -0,0 +1,60 @@
+from typing import List, Tuple
+
+from prowler.lib.outputs.inventory.models import ResourceEdge, ResourceNode
+
+
+def extract(client) -> Tuple[List[ResourceNode], List[ResourceEdge]]:
+    """
+    Extract ELBv2 (ALB/NLB) load balancer nodes and their edges.
+
+    Edges produced:
+      - load_balancer → security-group  [network]
+      - load_balancer → VPC             [network]
+    """
+    nodes: List[ResourceNode] = []
+    edges: List[ResourceEdge] = []
+
+    for lb in client.loadbalancersv2.values():
+        props = {
+            "type": getattr(lb, "type", None),
+            "scheme": getattr(lb, "scheme", None),
+            "dns_name": getattr(lb, "dns", None),
+            "vpc_id": getattr(lb, "vpc_id", None),
+        }
+
+        name = getattr(lb, "name", lb.arn.split("/")[-2] if "/" in lb.arn else lb.arn)
+
+        nodes.append(
+            ResourceNode(
+                id=lb.arn,
+                type="load_balancer",
+                name=name,
+                service="elbv2",
+                region=lb.region,
+                account_id=client.audited_account,
+                properties={k: v for k, v in props.items() if v is not None},
+            )
+        )
+
+        for sg_id in lb.security_groups or []:
+            edges.append(
+                ResourceEdge(
+                    source_id=lb.arn,
+                    target_id=sg_id,
+                    edge_type="network",
+                    label="sg",
+                )
+            )
+
+        vpc_id = getattr(lb, "vpc_id", None)
+        if vpc_id:
+            edges.append(
+                ResourceEdge(
+                    source_id=lb.arn,
+                    target_id=vpc_id,
+                    edge_type="network",
+                    label="in-vpc",
+                )
+            )
+
+    return nodes, edges

prowler/lib/outputs/inventory/extractors/iam_extractor.py

@@ -0,0 +1,82 @@
+import json
+from typing import Any, Dict, List, Tuple
+
+from prowler.lib.logger import logger
+from prowler.lib.outputs.inventory.models import ResourceEdge, ResourceNode
+
+
+def _parse_trust_principals(assume_role_policy: Any) -> List[str]:
+    """
+    Return a flat list of principal strings from an IAM assume-role policy document.
+    The policy may be a dict already or a JSON string.
+    """
+    if not assume_role_policy:
+        return []
+
+    if isinstance(assume_role_policy, str):
+        try:
+            assume_role_policy = json.loads(assume_role_policy)
+        except (json.JSONDecodeError, ValueError):
+            return []
+
+    principals = []
+    for statement in assume_role_policy.get("Statement", []):
+        principal = statement.get("Principal", {})
+        if isinstance(principal, str):
+            principals.append(principal)
+        elif isinstance(principal, dict):
+            for v in principal.values():
+                if isinstance(v, list):
+                    principals.extend(v)
+                else:
+                    principals.append(v)
+        elif isinstance(principal, list):
+            principals.extend(principal)
+
+    return principals
+
+
+def extract(client) -> Tuple[List[ResourceNode], List[ResourceEdge]]:
+    """
+    Extract IAM role nodes and their trust-relationship edges.
+
+    Edges produced:
+      - trusted-principal → role  [iam]  (who can assume this role)
+    """
+    nodes: List[ResourceNode] = []
+    edges: List[ResourceEdge] = []
+
+    for role in client.roles:
+        props: Dict[str, Any] = {
+            "path": getattr(role, "path", None),
+            "create_date": str(getattr(role, "create_date", "") or ""),
+        }
+
+        nodes.append(
+            ResourceNode(
+                id=role.arn,
+                type="iam_role",
+                name=role.name,
+                service="iam",
+                region="global",
+                account_id=client.audited_account,
+                properties={k: v for k, v in props.items() if v},
+            )
+        )
+
+        # Trust-relationship edges: principal → role (principal CAN assume role)
+        try:
+            for principal in _parse_trust_principals(role.assume_role_policy):
+                if principal and principal != "*":
+                    edges.append(
+                        ResourceEdge(
+                            source_id=principal,
+                            target_id=role.arn,
+                            edge_type="iam",
+                            label="can-assume",
+                        )
+                    )
+        except Exception as e:
+            logger.debug(f"inventory iam_extractor: could not parse trust policy for {role.arn}: {e}")
+
+    return nodes, edges