refactor: make bomreader a factory

By making a the BomReader a factory, we can unify the SBOM type
detection based on the input channel (e.g. file, stream, json). By that,
this complex and inconsistently implemented logic can be re-used across
the project.

This commit only provides the factory and does some minimal refactorings
to make the tests pass. In later commits more use of the infrastructure
will be made.

Signed-off-by: Felix Moessbauer <felix.moessbauer@siemens.com>

Author: fmoessbauer

src/debsbom/bomreader/bomreader.py

@@ -3,27 +3,67 @@
 # SPDX-License-Identifier: MIT
 
 from abc import abstractmethod
+from io import IOBase
 from pathlib import Path
-from io import TextIOBase
 
+from ..sbom import SBOMType
+from ..util.sbom_processor import SbomProcessor
 
-class BomReader:
+
+class BomReader(SbomProcessor):
     """Base class for SBOM importers"""
 
     @classmethod
-    @abstractmethod
-    def read_file(cls, filename: Path):
-        """Parse and return a BOM instance from the file"""
-        raise NotImplementedError()
+    def create(cls, filename: Path, bomtype: SBOMType | None = None):
+        if bomtype is SBOMType.SPDX or filename.name.endswith("spdx.json"):
+            SBOMType.SPDX.validate_dependency_availability()
+            from .spdxbomreader import SpdxBomFileReader
+
+            reader_cls = SpdxBomFileReader
+        elif bomtype is SBOMType.CycloneDX or filename.name.endswith("cdx.json"):
+            SBOMType.CycloneDX.validate_dependency_availability()
+            from .cdxbomreader import CdxBomFileReader
+
+            reader_cls = CdxBomFileReader
+        else:
+            raise RuntimeError("SBOM type cannot be detected based on filename")
+
+        return reader_cls(filename)
 
     @classmethod
-    @abstractmethod
-    def read_stream(cls, stream: TextIOBase):
-        """Parse and return a BOM instance from the stream"""
-        raise NotImplementedError()
+    def from_stream(cls, stream: IOBase, bomtype: SBOMType):
+        if bomtype is SBOMType.SPDX:
+            SBOMType.SPDX.validate_dependency_availability()
+            from .spdxbomreader import SpdxBomFileReader
+
+            reader_cls = SpdxBomFileReader
+        elif bomtype is SBOMType.CycloneDX:
+            SBOMType.CycloneDX.validate_dependency_availability()
+            from .cdxbomreader import CdxBomFileReader
+
+            reader_cls = CdxBomFileReader
+        else:
+            raise NotImplementedError("Unsupported SBOM type")
+
+        return reader_cls(stream)
 
     @classmethod
+    def from_json(cls, json_obj, bomtype: SBOMType):
+        if bomtype is SBOMType.SPDX:
+            SBOMType.SPDX.validate_dependency_availability()
+            from .spdxbomreader import SpdxBomJsonReader
+
+            reader_cls = SpdxBomJsonReader
+        elif bomtype is SBOMType.CycloneDX:
+            SBOMType.CycloneDX.validate_dependency_availability()
+            from .cdxbomreader import CdxBomJsonReader
+
+            reader_cls = CdxBomJsonReader
+        else:
+            raise NotImplementedError("Unsupported SBOM type")
+
+        return reader_cls(json_obj)
+
     @abstractmethod
-    def from_json(cls, json_obj):
-        """Parse and return a BOM instance from a Json object"""
-        raise NotImplementedError
+    def read(self):
+        return NotImplementedError()

src/debsbom/bomreader/cdxbomreader.py

@@ -12,18 +12,32 @@
 from cyclonedx.model.bom import Bom
 
 
-class CdxBomReader(BomReader, CDXType):
-    """Import an CycloneDX SBOM"""
+class CdxBomFileReader(BomReader, CDXType):
+    """Import a CycloneDX SBOM from a file"""
 
-    @classmethod
-    def read_file(cls, filename: Path) -> Bom:
-        with open(filename, "r") as f:
-            return cls.read_stream(f)
+    def __init__(self, filename: Path):
+        self.filename = filename
 
-    @classmethod
-    def read_stream(cls, stream: TextIOBase) -> Bom:
-        return cls.from_json(json.load(stream))
+    def read(self) -> Bom:
+        with open(self.filename, "r") as f:
+            return CdxBomStreamReader(f).read()
 
-    @classmethod
-    def from_json(cls, json_obj) -> Bom:
-        return Bom.from_json(json_obj)
+
+class CdxBomStreamReader(BomReader, CDXType):
+    """Import a CycloneDX SBOM from a file stream"""
+
+    def __init__(self, stream: TextIOBase):
+        self.stream = stream
+
+    def read(self) -> Bom:
+        return CdxBomJsonReader(json.load(self.stream)).read()
+
+
+class CdxBomJsonReader(BomReader, CDXType):
+    """Import a CycloneDX SBOM from a json object"""
+
+    def __init__(self, json_obj):
+        self.json_obj = json_obj
+
+    def read(self) -> Bom:
+        return Bom.from_json(self.json_obj)

src/debsbom/bomreader/spdxbomreader.py

@@ -6,7 +6,6 @@
 from pathlib import Path
 from io import TextIOBase
 
-from pathlib import Path
 from spdx_tools.spdx.parser.parse_anything import parse_file as spdx_parse_file
 from spdx_tools.spdx.parser.jsonlikedict.json_like_dict_parser import JsonLikeDictParser
 from spdx_tools.spdx.model.document import Document
@@ -15,17 +14,31 @@
 from ..sbom import SPDXType
 
 
-class SpdxBomReader(BomReader, SPDXType):
-    """Import an SPDX SBOM"""
+class SpdxBomFileReader(BomReader, SPDXType):
+    """Import a CycloneDX SBOM from a file"""
+
+    def __init__(self, filename: Path):
+        self.filename = filename
+
+    def read(self) -> Document:
+        return spdx_parse_file(str(self.filename))
+
+
+class SpdxBomStreamReader(BomReader, SPDXType):
+    """Import a CycloneDX SBOM from a file stream"""
+
+    def __init__(self, stream: TextIOBase):
+        self.stream = stream
+
+    def read(self) -> Document:
+        return SpdxBomJsonReader(json.load(self.stream)).read()
+
 
-    @classmethod
-    def read_file(cls, filename: Path) -> Document:
-        return spdx_parse_file(str(filename))
+class SpdxBomJsonReader(BomReader, SPDXType):
+    """Import a CycloneDX SBOM from a json object"""
 
-    @classmethod
-    def read_stream(cls, stream: TextIOBase) -> Document:
-        return cls.from_json(json.load(stream))
+    def __init__(self, json_obj):
+        self.json_obj = json_obj
 
-    @classmethod
-    def from_json(cls, json_obj) -> Document:
-        return JsonLikeDictParser().parse(json_obj)
+    def read(self) -> Document:
+        return JsonLikeDictParser().parse(self.json_obj)

src/debsbom/commands/merge.py

@@ -47,14 +47,14 @@ def run(args):
             raise ValueError("can not merge mixed SPDX and CycloneDX documents")
         elif len(spdx_paths) > 0 or args.sbom_type == "spdx":
             SBOMType.SPDX.validate_dependency_availability()
-            from ..bomreader.spdxbomreader import SpdxBomReader
+            from ..bomreader.spdxbomreader import SpdxBomFileReader, SpdxBomJsonReader
             from ..merge.spdx import SpdxSbomMerger
 
             if json_sboms:
                 for obj in json_sboms:
-                    docs.append(SpdxBomReader.from_json(obj))
+                    docs.append(SpdxBomJsonReader(obj).read())
             for path in spdx_paths:
-                docs.append(SpdxBomReader.read_file(path))
+                docs.append(SpdxBomFileReader(path).read())
             sbom_merger = SpdxSbomMerger(
                 distro_name=args.distro_name,
                 distro_supplier=args.distro_supplier,
@@ -74,14 +74,14 @@ def run(args):
                 BomWriter.write_to_file(bom, SBOMType.SPDX, Path(out), args.validate)
         elif len(cdx_paths) > 0 or args.sbom_type == "cdx":
             SBOMType.CycloneDX.validate_dependency_availability()
-            from ..bomreader.cdxbomreader import CdxBomReader
+            from ..bomreader.cdxbomreader import CdxBomFileReader, CdxBomJsonReader
             from ..merge.cdx import CdxSbomMerger
 
             if json_sboms:
                 for obj in json_sboms:
-                    docs.append(CdxBomReader.from_json(obj))
+                    docs.append(CdxBomJsonReader(obj).read())
             for path in cdx_paths:
-                docs.append(CdxBomReader.read_file(path))
+                docs.append(CdxBomFileReader(path).read())
             sbom_merger = CdxSbomMerger(
                 distro_name=args.distro_name,
                 distro_supplier=args.distro_supplier,

src/debsbom/export/exporter.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 from io import IOBase
 
+from ..bomreader.bomreader import BomReader
 from ..util.sbom_processor import SbomProcessor
 from ..sbom import SBOMType
 
@@ -33,24 +34,22 @@ def create(filename: Path, format: GraphOutputFormat) -> "GraphExporter":
         """
         Factory to create a GraphExporter for the given SBOM type (based on the filename extension).
         """
-        if filename.name.endswith("spdx.json"):
-            SBOMType.SPDX.validate_dependency_availability()
-            from ..bomreader.spdxbomreader import SpdxBomReader
-            from .spdx import SpdxGraphMLExporter
+        reader = BomReader.create(filename)
+        if format == GraphOutputFormat.GRAPHML:
+            if reader.sbom_type() == SBOMType.SPDX:
+                from .spdx import SpdxGraphMLExporter
 
-            bom = SpdxBomReader.read_file(filename)
-            if format == GraphOutputFormat.GRAPHML:
-                return SpdxGraphMLExporter(bom)
-        elif filename.name.endswith("cdx.json"):
-            SBOMType.CycloneDX.validate_dependency_availability()
-            from ..bomreader.cdxbomreader import CdxBomReader
-            from .cdx import CdxGraphMLExporter
+                exporter_cls = SpdxGraphMLExporter
+            elif reader.sbom_type() == SBOMType.CycloneDX:
+                from .cdx import CdxGraphMLExporter
 
-            bom = CdxBomReader.read_file(filename)
-            if format == GraphOutputFormat.GRAPHML:
-                return CdxGraphMLExporter(bom)
+                exporter_cls = CdxGraphMLExporter
+            else:
+                raise NotImplementedError("unreachable")
         else:
-            raise RuntimeError("Cannot determine file format")
+            raise NotImplementedError("unreachable")
+
+        return exporter_cls(reader.read())
 
     @staticmethod
     def from_stream(

src/debsbom/resolver/resolver.py

@@ -34,15 +34,15 @@ def create(filename: Path, bomtype: SBOMType | None = None) -> "PackageResolver"
         if filename.name.endswith("spdx.json"):
             SBOMType.SPDX.validate_dependency_availability()
             from .spdx import SpdxPackageResolver
-            from ..bomreader.spdxbomreader import SpdxBomReader
+            from ..bomreader.spdxbomreader import SpdxBomFileReader
 
-            return SpdxPackageResolver(SpdxBomReader.read_file(filename))
+            return SpdxPackageResolver(SpdxBomFileReader(filename).read())
         elif filename.name.endswith("cdx.json"):
             SBOMType.CycloneDX.validate_dependency_availability()
             from .cdx import CdxPackageResolver
-            from ..bomreader.cdxbomreader import CdxBomReader
+            from ..bomreader.cdxbomreader import CdxBomFileReader
 
-            return CdxPackageResolver(CdxBomReader.read_file(filename))
+            return CdxPackageResolver(CdxBomFileReader(filename).read())
         else:
             raise RuntimeError("Cannot determine file format")
 

tests/test_merge.py

@@ -12,7 +12,7 @@ def test_spdx_merge():
     _spdx_tools = pytest.importorskip("spdx_tools")
 
     from spdx_tools.spdx.model.relationship import Relationship, RelationshipType
-    from debsbom.bomreader.spdxbomreader import SpdxBomReader
+    from debsbom.bomreader.spdxbomreader import SpdxBomFileReader
     from debsbom.merge.spdx import SpdxSbomMerger
 
     # The test files are created with
@@ -24,7 +24,7 @@ def test_spdx_merge():
     merger = SpdxSbomMerger(distro_name=distro_name)
     docs = []
     for sbom in ["tests/data/merge-full.spdx.json", "tests/data/merge-minimal.spdx.json"]:
-        docs.append(SpdxBomReader.read_file(Path(sbom)))
+        docs.append(SpdxBomFileReader(Path(sbom)).read())
     bom = merger.merge(docs)
 
     assert (
@@ -73,7 +73,7 @@ def test_cdx_merge():
     _cyclonedx = pytest.importorskip("cyclonedx")
 
     from cyclonedx.model.dependency import Dependency
-    from debsbom.bomreader.cdxbomreader import CdxBomReader
+    from debsbom.bomreader.cdxbomreader import CdxBomFileReader
     from debsbom.merge.cdx import CdxSbomMerger
 
     # The test files are created with
@@ -85,7 +85,7 @@ def test_cdx_merge():
     merger = CdxSbomMerger(distro_name=distro_name)
     docs = []
     for sbom in ["tests/data/merge-full.cdx.json", "tests/data/merge-minimal.cdx.json"]:
-        docs.append(CdxBomReader.read_file(Path(sbom)))
+        docs.append(CdxBomFileReader(Path(sbom)).read())
     bom = merger.merge(docs)
 
     distro_bom_ref = bom.metadata.component.bom_ref
@@ -121,7 +121,7 @@ def test_cdx_merge():
 def test_cdx_hash_merge():
     _cyclonedx = pytest.importorskip("cyclonedx")
 
-    from debsbom.bomreader.cdxbomreader import CdxBomReader
+    from debsbom.bomreader.cdxbomreader import CdxBomFileReader
     from debsbom.merge.cdx import CdxSbomMerger
 
     distro_name = "cdx-merge-hash-merge"
@@ -131,7 +131,7 @@ def test_cdx_hash_merge():
         "tests/data/checksum-merge-md5.cdx.json",
         "tests/data/checksum-merge-sha256.cdx.json",
     ]:
-        docs.append(CdxBomReader.read_file(Path(sbom)))
+        docs.append(CdxBomFileReader(Path(sbom)).read())
     bom = merger.merge(docs)
 
     component = next(iter(bom.components))
@@ -141,7 +141,7 @@ def test_cdx_hash_merge():
 def test_spdx_checksum_merge():
     _spdx_tools = pytest.importorskip("spdx_tools")
 
-    from debsbom.bomreader.spdxbomreader import SpdxBomReader
+    from debsbom.bomreader.spdxbomreader import SpdxBomFileReader
     from debsbom.merge.spdx import SpdxSbomMerger
 
     distro_name = "spx-merge-checksum-merge"
@@ -151,7 +151,7 @@ def test_spdx_checksum_merge():
         "tests/data/checksum-merge-md5.spdx.json",
         "tests/data/checksum-merge-sha256.spdx.json",
     ]:
-        docs.append(SpdxBomReader.read_file(Path(sbom)))
+        docs.append(SpdxBomFileReader(Path(sbom)).read())
     bom = merger.merge(docs)
 
     package = next(iter(filter(lambda p: p.name == "example-pkg", bom.packages)))
@@ -161,7 +161,7 @@ def test_spdx_checksum_merge():
 def test_cdx_bad_checksum():
     _cyclonedx = pytest.importorskip("cyclonedx")
 
-    from debsbom.bomreader.cdxbomreader import CdxBomReader
+    from debsbom.bomreader.cdxbomreader import CdxBomFileReader
     from debsbom.merge.cdx import CdxSbomMerger
 
     distro_name = "cdx-merge-hash-merge"
@@ -171,15 +171,15 @@ def test_cdx_bad_checksum():
         "tests/data/checksum-merge-md5.cdx.json",
         "tests/data/checksum-merge-bad.cdx.json",
     ]:
-        docs.append(CdxBomReader.read_file(Path(sbom)))
+        docs.append(CdxBomFileReader(Path(sbom)).read())
     with pytest.raises(ChecksumMismatchError):
         _bom = merger.merge(docs)
 
 
 def test_spdx_bad_checksum():
     _spdx_tools = pytest.importorskip("spdx_tools")
 
-    from debsbom.bomreader.spdxbomreader import SpdxBomReader
+    from debsbom.bomreader.spdxbomreader import SpdxBomFileReader
     from debsbom.merge.spdx import SpdxSbomMerger
 
     distro_name = "cdx-merge-checksum-bad"
@@ -189,7 +189,7 @@ def test_spdx_bad_checksum():
         "tests/data/checksum-merge-md5.spdx.json",
         "tests/data/checksum-merge-bad.spdx.json",
     ]:
-        docs.append(SpdxBomReader.read_file(Path(sbom)))
+        docs.append(SpdxBomFileReader(Path(sbom)).read())
 
     with pytest.raises(ChecksumMismatchError):
         _bom = merger.merge(docs)