Add Neo4j output writer for Surfactant SBOM exports

scripts/load_sbom_to_neo4j.py

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+# /// script
+# dependencies = [
+#   "neo4j>=5",
+# ]
+# ///
+"""Standalone loader for importing an existing CyTRICS SBOM into Neo4j.
+
+Usage:
+
+    export NEO4J_URI=neo4j://localhost:7687
+    export NEO4J_USER=neo4j
+    export NEO4J_PASSWORD=your-password
+    python load_sbom_to_neo4j.py existing-sbom.json
+
+This script assumes it is run inside a Surfactant checkout or environment where
+`surfactant` is importable. It imports `export_sbom_to_neo4j` from the
+surfactant.output.neo4j_writer module.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+
+from neo4j import GraphDatabase
+
+from surfactant.input_readers.cytrics_reader import read_sbom
+from surfactant.output.neo4j_writer import export_sbom_to_neo4j
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Import an existing Surfactant/CyTRICS SBOM into Neo4j"
+    )
+    parser.add_argument("sbom_json", help="Path to an existing CyTRICS SBOM JSON file")
+    parser.add_argument("--database", default=os.environ.get("NEO4J_DATABASE", "neo4j"))
+    parser.add_argument(
+        "--batch-size", type=int, default=int(os.environ.get("NEO4J_BATCH_SIZE", "1000"))
+    )
+    args = parser.parse_args()
+
+    if args.batch_size <= 0:
+        raise SystemExit("--batch-size must be greater than zero")
+
+    uri = os.environ.get("NEO4J_URI")
+    user = os.environ.get("NEO4J_USER", "neo4j")
+    password = os.environ.get("NEO4J_PASSWORD")
+    if not uri:
+        raise SystemExit("NEO4J_URI is required, for example neo4j://localhost:7687")
+    if password is None:
+        raise SystemExit("NEO4J_PASSWORD is required")
+
+    with open(args.sbom_json, encoding="utf-8") as infile:
+        sbom = read_sbom(infile)
+
+    with GraphDatabase.driver(uri, auth=(user, password)) as driver:
+        summary = export_sbom_to_neo4j(
+            sbom,
+            driver=driver,
+            database=args.database,
+            batch_size=args.batch_size,
+        )
+
+    print(json.dumps(summary, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

surfactant/cmd/generate.py

@@ -7,7 +7,7 @@
 import queue
 import re
 import sys
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Mapping, Optional, Tuple, Union
 
 import click
 from loguru import logger
@@ -159,6 +159,33 @@ def _normalize_author_value(value: Optional[str], option_name: str) -> Optional[
     return value or None
 
 
+def _is_path_graph_node(attrs: Mapping[str, Any]) -> bool:
+    return str((attrs or {}).get("type") or "").lower() == "path"
+
+
+def _is_logical_graph_edge(graph: Any, u: Any, v: Any, key: Any) -> bool:
+    if str(key).lower() == "symlink":
+        return False
+
+    if _is_path_graph_node(graph.nodes.get(u, {})):
+        return False
+    if _is_path_graph_node(graph.nodes.get(v, {})):
+        return False
+
+    return True
+
+
+def _count_logical_graph_edges(graph: Any) -> int:
+    if graph is None:
+        return 0
+
+    return sum(
+        1
+        for u, v, key in graph.edges(keys=True)
+        if _is_logical_graph_edge(graph, u, v, key)
+    )
+
+
 # pylint: disable-next=redefined-outer-name
 def _set_sbom_author(sbom: SBOM, author_name: Optional[str], author_type: Optional[str]) -> None:
     """Set a generated SBOM author from CLI/config values."""
@@ -518,6 +545,15 @@ def sbom(
     output_writer = find_io_plugin(pm, output_format, "write_sbom")
     input_reader = find_io_plugin(pm, input_format, "read_sbom")
 
+    logger.info(
+        f'Using SBOM output writer "{pm.get_canonical_name(output_writer)}" '
+        f'for output_format="{output_format}"'
+    )
+    logger.info(
+        f'Using SBOM input reader "{pm.get_canonical_name(input_reader)}" '
+        f'for input_format="{input_format}"'
+    )
+
     contextQ: queue.Queue[ContextEntry] = queue.Queue()
 
     for cfg_entry in specimen_context:
@@ -527,10 +563,25 @@ def sbom(
     new_sbom: SBOM
     # Click has Sentinel.UNSET type that doesn't have READ attribute, which may appear when running regression test script
     if not input_sbom or not hasattr(input_sbom, "read"):
+        logger.info("No input SBOM supplied; creating a new empty SBOM")
         new_sbom = SBOM()
     else:
+        logger.info(f"Reading input SBOM from {getattr(input_sbom, 'name', '<stream>')}")
         new_sbom = input_reader.read_sbom(input_sbom)
 
+        graph = getattr(new_sbom, "graph", None)
+        fs_tree = getattr(new_sbom, "fs_tree", None)
+
+        logger.info(
+            "Loaded input SBOM: "
+            f"software={len(getattr(new_sbom, 'software', []) or [])}, "
+            f"logical_relationships={_count_logical_graph_edges(graph)}, "
+            f"graph_nodes={graph.number_of_nodes() if graph is not None else 'None'}, "
+            f"graph_edges_total={graph.number_of_edges() if graph is not None else 'None'}, "
+            f"fs_tree_nodes={fs_tree.number_of_nodes() if fs_tree is not None else 'None'}, "
+            f"fs_tree_edges={fs_tree.number_of_edges() if fs_tree is not None else 'None'}"
+        )
+
     _set_sbom_author(new_sbom, author_name, author_type)
 
     # gather metadata for files and add/augment software entries in the sbom
@@ -886,7 +937,24 @@ def sbom(
         logger.info("Skipping relationships based on imports metadata")
 
     # TODO should contents from different containers go in different SBOM files, so new portions can be added bit-by-bit with a final merge?
+    graph = getattr(new_sbom, "graph", None)
+    fs_tree = getattr(new_sbom, "fs_tree", None)
+
+    logger.info(
+        "Prepared SBOM for output: "
+        f"software={len(getattr(new_sbom, 'software', []) or [])}, "
+        f"logical_relationships={_count_logical_graph_edges(graph)}, "
+        f"graph_nodes={graph.number_of_nodes() if graph is not None else 'None'}, "
+        f"graph_edges_total={graph.number_of_edges() if graph is not None else 'None'}, "
+        f"fs_tree_nodes={fs_tree.number_of_nodes() if fs_tree is not None else 'None'}, "
+        f"fs_tree_edges={fs_tree.number_of_edges() if fs_tree is not None else 'None'}"
+    )
+    logger.info(
+        f"Calling output writer {pm.get_canonical_name(output_writer)} "
+        f"for {getattr(sbom_outfile, 'name', '<stream>')}"
+    )
     output_writer.write_sbom(new_sbom, sbom_outfile)
+    logger.info(f"Finished writing SBOM output to {getattr(sbom_outfile, 'name', '<stream>')}")
 
 
 def resolve_link(