""" inventory_output.py ------------------- Writes the ConnectivityGraph produced by graph_builder to two files: .inventory.json – machine-readable graph (nodes + edges) .inventory.html – interactive D3.js force-directed graph """ import json import os from dataclasses import asdict from datetime import datetime from typing import Optional from prowler.lib.logger import logger from lib.models import ConnectivityGraph # --------------------------------------------------------------------------- # JSON output # --------------------------------------------------------------------------- def write_json(graph: ConnectivityGraph, file_path: str) -> None: """Serialise the graph to a JSON file.""" try: os.makedirs(os.path.dirname(file_path), exist_ok=True) data = { "generated_at": datetime.utcnow().isoformat() + "Z", "nodes": [asdict(n) for n in graph.nodes], "edges": [asdict(e) for e in graph.edges], "stats": { "node_count": len(graph.nodes), "edge_count": len(graph.edges), }, } with open(file_path, "w", encoding="utf-8") as fh: json.dump(data, fh, indent=2, default=str) logger.info(f"Inventory graph JSON written to {file_path}") except Exception as e: logger.error( f"inventory_output.write_json: {e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}" ) # --------------------------------------------------------------------------- # HTML output (self-contained, D3.js CDN) # --------------------------------------------------------------------------- # Colour palette per node type _NODE_COLOURS = { "lambda_function": "#f59e0b", "ec2_instance": "#3b82f6", "security_group": "#6366f1", "vpc": "#10b981", "subnet": "#34d399", "rds_instance": "#ef4444", "load_balancer": "#8b5cf6", "s3_bucket": "#06b6d4", "iam_role": "#f97316", "default": "#94a3b8", } # Edge stroke colours per edge type _EDGE_COLOURS = { "network": "#64748b", "iam": "#f97316", "triggers": "#a855f7", "data_flow": "#0ea5e9", "depends_on": "#94a3b8", "routes_to": "#22c55e", "replicates_to": "#ec4899", "encrypts": "#eab308", "logs_to": "#78716c", } _HTML_TEMPLATE = """\ Prowler – AWS Connectivity Graph

Node types

{legend_nodes_html}

Edge types

{legend_edges_html}
""" def _build_legend_html(colours: dict, shape: str) -> str: rows = [] for key, colour in sorted(colours.items()): if shape == "dot": rows.append( f'
' f'
' f"{key}
" ) else: rows.append( f'
' f'
' f"{key}
" ) return "\n".join(rows) def write_html(graph: ConnectivityGraph, file_path: str) -> None: """Render the graph as a self-contained interactive HTML page.""" try: os.makedirs(os.path.dirname(file_path), exist_ok=True) nodes_json = json.dumps( [ { "id": n.id, "type": n.type, "name": n.name, "service": n.service, "region": n.region, "account_id": n.account_id, "properties": n.properties, } for n in graph.nodes ], indent=None, default=str, ) edges_json = json.dumps( [ { "source_id": e.source_id, "target_id": e.target_id, "edge_type": e.edge_type, "label": e.label or "", } for e in graph.edges ], indent=None, default=str, ) html = _HTML_TEMPLATE.format( generated_at=datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC"), nodes_json=nodes_json, edges_json=edges_json, node_colours_json=json.dumps(_NODE_COLOURS), edge_colours_json=json.dumps(_EDGE_COLOURS), legend_nodes_html=_build_legend_html(_NODE_COLOURS, "dot"), legend_edges_html=_build_legend_html(_EDGE_COLOURS, "line"), ) with open(file_path, "w", encoding="utf-8") as fh: fh.write(html) logger.info(f"Inventory graph HTML written to {file_path}") except Exception as e: logger.error( f"inventory_output.write_html: {e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}" ) # --------------------------------------------------------------------------- # Convenience entry-point called from __main__.py # --------------------------------------------------------------------------- def generate_inventory_outputs(output_path: str) -> None: """ Build the connectivity graph from currently-loaded service clients and write both JSON and HTML outputs. Args: output_path: base file path WITHOUT extension, e.g. "output/prowler-output-20240101120000". The function appends .inventory.json and .inventory.html. """ from lib.graph_builder import build_graph graph = build_graph() if not graph.nodes: logger.warning( "Inventory graph: no nodes discovered. " "Make sure at least one AWS service was scanned before generating the inventory." ) write_json(graph, f"{output_path}.inventory.json") write_html(graph, f"{output_path}.inventory.html")