feat(sbom): symlink-aware SBOM filesystem graph (fs_tree)

.gitignore

@@ -279,4 +279,4 @@ $RECYCLE.BIN/
 .Trash-*
 
 # .nfs files are created when an open file is removed but is still being accessed
-.nfs*
+.nfs*

surfactant/cmd/generate.py

@@ -7,10 +7,11 @@
 import queue
 import re
 import sys
-from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 import click
 from loguru import logger
+from networkx.exception import NetworkXError
 
 from surfactant import ContextEntry
 from surfactant.cmd.internal.generate_utils import SpecimenContextParamType
@@ -324,7 +325,7 @@ def sbom(
         while not contextQ.empty():
             entry: ContextEntry = contextQ.get()
             if entry.archive:
-                logger.info("Processing parent container " + str(entry.archive))
+                logger.info(f"Processing parent container {entry.archive}")
                 # TODO: if the parent archive has an info extractor that does unpacking interally, should the children be added to the SBOM?
                 # current thoughts are (Syft) doesn't provide hash information for a proper SBOM software entry, so exclude these
                 # extractor plugins meant to unpack files could be okay when used on an "archive", but then extractPaths should be empty
@@ -419,17 +420,40 @@ def sbom(
                             user_institution_name=recorded_institution,
                             container_prefix=entry.containerPrefix,
                         )
-                    except Exception as e:
+                    except Exception as e:  # pylint: disable=broad-exception-caught
                         raise RuntimeError(f"Unable to process: {filepath}") from e
                     entries.append(sw_parent)
                     entries.extend(sw_children if sw_children else [])
+                    # ------------------------------------------------------------------------
+                    # (Optional - Early Injection) Inject symlink paths into each Software entry so SBOM helper handles them
+                    # ------------------------------------------------------------------------
+                    # Early injection: add symlinks gathered so far so fs_tree sees them
+                    for sw in entries:
+                        if sw.fileName is None:
+                            sw.fileName = []
+                        if sw.installPath is None:
+                            sw.installPath = []
+                        # Filename symlinks
+                        for link in filename_symlinks.get(sw.sha256, []):
+                            if link not in sw.fileName:
+                                logger.debug(
+                                    f"Injecting filename symlink '{link}' for SHA {sw.sha256}"
+                                )
+                                sw.fileName.append(link)
+                        # Install-path symlinks
+                        for link in file_symlinks.get(sw.sha256, []):
+                            if link not in sw.installPath:
+                                logger.debug(
+                                    f"Injecting install-path symlink '{link}' for SHA {sw.sha256}"
+                                )
+                                sw.installPath.append(link)
                     new_sbom.add_software_entries(entries, parent_entry=parent_entry)
                     # epath was a file, no need to walk the directory tree
                     continue
 
                 # epath is a directory, walk it
                 for cdir, dirs, files in os.walk(epath):
-                    logger.info("Processing " + str(cdir))
+                    logger.info(f"Processing {cdir}")
 
                     if entry.installPrefix:
                         for dir_ in dirs:
@@ -446,8 +470,20 @@ def sbom(
                                         epath.as_posix(), entry.installPrefix, dest
                                     )
                                     dir_symlinks.append((install_source, install_dest))
-
-                    entries = []
+                                    # Reflect in fs_tree immediately
+                                    try:
+                                        new_sbom.record_symlink(
+                                            install_source, install_dest, subtype="directory"
+                                        )
+                                        logger.debug(
+                                            f"[fs_tree] (dir) {install_source} → {install_dest}"
+                                        )
+                                    except (NetworkXError, ValueError) as e:
+                                        logger.warning(
+                                            f"Failed to record directory symlink in fs_tree: {install_source} → {install_dest}: {e}"
+                                        )
+
+                    entries: List[Software] = []
                     for file in files:
                         # os.path.join will insert an OS specific separator between cdir and f
                         # need to make sure that separator is a / and not a \ on windows
@@ -463,6 +499,7 @@ def sbom(
                             # Dead/infinite links will error so skip them
                             if true_filepath is None:
                                 continue
+
                             # Compute sha256 hash of the file; skip if the file pointed by the symlink can't be opened
                             try:
                                 true_file_sha256 = sha256sum(true_filepath)
@@ -471,6 +508,26 @@ def sbom(
                                     f"Unable to open symlink {filepath} pointing to {true_filepath}"
                                 )
                                 continue
+
+                            # Record both source and target paths under the same hash node
+                            install_filepath = real_path_to_install_path(
+                                epath.as_posix(), entry.installPrefix, filepath
+                            )
+                            install_dest = real_path_to_install_path(
+                                epath.as_posix(), entry.installPrefix, true_filepath
+                            )
+
+                            try:
+                                new_sbom.record_hash_node(install_filepath, true_file_sha256)
+                                new_sbom.record_hash_node(install_dest, true_file_sha256)
+                                logger.debug(
+                                    f"[fs_tree] Linked symlink + target by hash: {install_filepath} ↔ {install_dest}"
+                                )
+                            except Exception as e:  # pylint: disable=broad-exception-caught
+                                logger.warning(
+                                    f"[fs_tree] Failed to link symlink + target by hash for {filepath}: {e}"
+                                )
+
                             # Record the symlink name to be added as a file name
                             # Dead links would appear as a file, so need to check the true path to see
                             # if the thing pointed to is a file or a directory
@@ -496,6 +553,22 @@ def sbom(
                                     file_symlinks[true_file_sha256].append(install_filepath)
                                 else:
                                     dir_symlinks.append((install_filepath, install_dest))
+
+                                # Reflect this symlink in fs_tree immediately
+                                try:
+                                    subtype = (
+                                        "file" if os.path.isfile(true_filepath) else "directory"
+                                    )
+                                    new_sbom.record_symlink(
+                                        install_filepath, install_dest, subtype=subtype
+                                    )
+                                    logger.debug(
+                                        f"[fs_tree] ({subtype}) {install_filepath} → {install_dest}"
+                                    )
+                                except (NetworkXError, ValueError) as e:
+                                    logger.warning(
+                                        f"Failed to record symlink in fs_tree: {install_filepath} → {install_dest}: {e}"
+                                    )
                             # NOTE Two cases that don't get recorded (but maybe should?) are:
                             # 1. If the file pointed to is outside the extract paths, it won't
                             # appear in the SBOM at all -- is that desirable? If it were included,
@@ -509,6 +582,8 @@ def sbom(
                                 entry.includeFileExts = []
                             if not entry.excludeFileExts:
                                 entry.excludeFileExts = []
+
+                            # file-type identification and SBOM entry creation
                             if (
                                 (
                                     ftype := pm.hook.identify_file_type(
@@ -539,69 +614,52 @@ def sbom(
                                         or entry.omitUnrecognizedTypes,
                                         container_prefix=entry.containerPrefix,
                                     )
-                                except Exception as e:
+                                except Exception as e:  # pylint: disable=broad-exception-caught
                                     raise RuntimeError(f"Unable to process: {filepath}") from e
 
                                 entries.append(sw_parent)
                                 entries.extend(sw_children if sw_children else [])
+                    # ------------------------------------------------------------------------
+                    # (Optional - Early Injection) Inject symlink paths into each Software entry so SBOM helper handles them
+                    # ------------------------------------------------------------------------
+                    # Early injection for batch (so fs_tree captures aliases)
+                    for sw in entries:
+                        if sw.fileName is None:
+                            sw.fileName = []
+                        if sw.installPath is None:
+                            sw.installPath = []
+                        # Filename symlinks
+                        for link in filename_symlinks.get(sw.sha256, []):
+                            if link not in sw.fileName:
+                                logger.debug(
+                                    f"Injecting filename symlink '{link}' for SHA {sw.sha256}"
+                                )
+                                sw.fileName.append(link)
+                        # Install-path symlinks
+                        for link in file_symlinks.get(sw.sha256, []):
+                            if link not in sw.installPath:
+                                logger.debug(
+                                    f"Injecting install-path symlink '{link}' for SHA {sw.sha256}"
+                                )
+                                sw.installPath.append(link)
                     new_sbom.add_software_entries(entries, parent_entry=parent_entry)
 
-        # Add symlinks to install paths and file names
-        for software in new_sbom.software:
-            # ensure fileName, installPath, and metadata lists for the software entry have been created
-            # for a user supplied input SBOM, there are no guarantees
-            if software.fileName is None:
-                software.fileName = []
-            if software.installPath is None:
-                software.installPath = []
-            if software.metadata is None:
-                software.metadata = []
-            if software.sha256 in filename_symlinks:
-                filename_symlinks_added = []
-                for filename in filename_symlinks[software.sha256]:
-                    if filename not in software.fileName:
-                        software.fileName.append(filename)
-                        filename_symlinks_added.append(filename)
-                if filename_symlinks_added:
-                    # Store information on which file names are symlinks
-                    software.metadata.append({"fileNameSymlinks": filename_symlinks_added})
-            if software.sha256 in file_symlinks:
-                symlinks_added = []
-                for full_path in file_symlinks[software.sha256]:
-                    if full_path not in software.installPath:
-                        software.installPath.append(full_path)
-                        symlinks_added.append(full_path)
-                if symlinks_added:
-                    # Store information on which install paths are symlinks
-                    software.metadata.append({"installPathSymlinks": symlinks_added})
-
-        # Add directory symlink destinations to extract/install paths
-        for software in new_sbom.software:
-            # NOTE: this probably doesn't actually add any containerPath symlinks
-            for paths in (software.containerPath, software.installPath):
-                if paths is None:
-                    continue
-                paths_to_add = []
-                for path in paths:
-                    for link_source, link_dest in dir_symlinks:
-                        if path.startswith(link_dest):
-                            # Replace the matching start with the symlink instead
-                            # We can't use os.path.join here because we end up with absolute paths after
-                            # removing the common start.
-                            paths_to_add.append(path.replace(link_dest, link_source, 1))
-                if paths_to_add:
-                    found_md_installpathsymlinks = False
-                    # make sure software.metadata list has been initialized
-                    if software.metadata is None:
-                        software.metadata = []
-                    if isinstance(software.metadata, Iterable):
-                        for md in software.metadata:
-                            if isinstance(md, Dict) and "installPathSymlinks" in md:
-                                found_md_installpathsymlinks = True
-                                md["installPathSymlinks"] += paths_to_add
-                    if not found_md_installpathsymlinks:
-                        software.metadata.append({"installPathSymlinks": paths_to_add})
-                    paths += paths_to_add
+        # ------------------------------------------------------------------
+        # Expand deferred directory symlinks once fs_tree is fully populated
+        # ------------------------------------------------------------------
+        new_sbom.expand_pending_dir_symlinks()
+
+        # ------------------------------------------------------------------
+        # Expand deferred file symlinks after all installPath nodes are added
+        # ------------------------------------------------------------------
+        new_sbom.expand_pending_file_symlinks()
+
+        # ------------------------------------------------------------------
+        # Inject legacy-style symlink metadata (fileNameSymlinks and
+        # installPathSymlinks) derived from fs_tree relationships
+        # ------------------------------------------------------------------
+        new_sbom.inject_symlink_metadata()
+
     else:
         logger.info("Skipping gathering file metadata and adding software entries")
 

surfactant/cmd/merge.py

@@ -98,7 +98,11 @@ def merge(
         merged_sbom.merge(sbom_m)
 
     # Find root nodes: those with zero incoming edges
-    roots = [n for n, deg in merged_sbom.graph.in_degree() if deg == 0]
+    roots = [
+        n
+        for n, deg in merged_sbom.graph.in_degree()
+        if deg == 0 and merged_sbom.graph.nodes.get(n, {}).get("type") != "Path"
+    ]
     logger.info(f"ROOT NODES: {roots}")
 
     # Detect any directed cycles
@@ -122,6 +126,8 @@ def merge(
         if config and "systemRelationship" in config:
             system_relationship = config["systemRelationship"]
         for root_uuid in roots:
+            if merged_sbom.graph.nodes.get(root_uuid, {}).get("type") == "Path":
+                continue
             merged_sbom.create_relationship(system_obj.UUID, root_uuid, system_relationship)
     else:
         logger.warning(

surfactant/output/cytrics_writer.py

@@ -4,13 +4,16 @@
 # SPDX-License-Identifier: MIT
 from typing import Optional
 
+from loguru import logger
+
 import surfactant.plugin
 from surfactant.sbomtypes import SBOM
 
 
 @surfactant.plugin.hookimpl
 def write_sbom(sbom: SBOM, outfile) -> None:
     # outfile is a file pointer, not a file name
+    logger.debug("writing SBOM")
     outfile.write(sbom.to_json(indent=2))
 
 

surfactant/relationships/_internal/windows_utils.py

@@ -37,3 +37,47 @@ def find_installed_software(
                             # matching probe directory and filename, add software to list
                             possible_matches.append(e)
     return possible_matches
+
+
+# construct a list of directories to probe for establishing dotnet relationships
+def get_dotnet_probedirs(software: Software, refCulture, refName, dnProbingPaths):
+    probedirs = []
+    # probe for the referenced assemblies
+    if isinstance(software.installPath, Iterable):
+        for install_filepath in software.installPath:
+            install_basepath = pathlib.PureWindowsPath(install_filepath).parent.as_posix()
+            if refCulture is None or refCulture == "":
+                # [application base] / [assembly name].dll
+                # [application base] / [assembly name] / [assembly name].dll
+                probedirs.append(pathlib.PureWindowsPath(install_basepath).as_posix())
+                probedirs.append(pathlib.PureWindowsPath(install_basepath, refName).as_posix())
+                if dnProbingPaths is not None:
+                    # add probing private paths
+                    for path in dnProbingPaths:
+                        # [application base] / [binpath] / [assembly name].dll
+                        # [application base] / [binpath] / [assembly name] / [assembly name].dll
+                        probedirs.append(pathlib.PureWindowsPath(install_basepath, path).as_posix())
+                        probedirs.append(
+                            pathlib.PureWindowsPath(install_basepath, path, refName).as_posix()
+                        )
+            else:
+                # [application base] / [culture] / [assembly name].dll
+                # [application base] / [culture] / [assembly name] / [assembly name].dll
+                probedirs.append(pathlib.PureWindowsPath(install_basepath, refCulture).as_posix())
+                probedirs.append(
+                    pathlib.PureWindowsPath(install_basepath, refName, refCulture).as_posix()
+                )
+                if dnProbingPaths is not None:
+                    # add probing private paths
+                    for path in dnProbingPaths:
+                        # [application base] / [binpath] / [culture] / [assembly name].dll
+                        # [application base] / [binpath] / [culture] / [assembly name] / [assembly name].dll
+                        probedirs.append(
+                            pathlib.PureWindowsPath(install_basepath, path, refCulture).as_posix()
+                        )
+                        probedirs.append(
+                            pathlib.PureWindowsPath(
+                                install_basepath, path, refName, refCulture
+                            ).as_posix()
+                        )
+    return probedirs