Cyt 814 docker scout

surfactant/cmd/generate.py

@@ -186,7 +186,7 @@ def warn_if_hash_collision(soft1: Optional[Software], soft2: Optional[Software])
     help="SBOM output format, see --list-output-formats for list of options; default is CyTRICS",
 )
 @click.option(
-    "--list-output-formats",
+    "--list_output_formats",
     is_flag=True,
     callback=print_output_formats,
     expose_value=False,
@@ -200,7 +200,7 @@ def warn_if_hash_collision(soft1: Optional[Software], soft2: Optional[Software])
     help="Input SBOM format, see --list-input-formats for list of options; default is CyTRICS",
 )
 @click.option(
-    "--list-input-formats",
+    "--list_input_formats",
     is_flag=True,
     callback=print_input_formats,
     expose_value=False,

surfactant/filetypeid/id_magic.py

@@ -2,7 +2,9 @@
 # See the top-level LICENSE file for details.
 #
 # SPDX-License-Identifier: MIT
+import json
 import pathlib
+import tarfile
 from enum import Enum, auto
 from typing import Optional
 
@@ -19,6 +21,30 @@ class ExeType(Enum):
     MACHO64 = auto()
 
 
+def is_docker_archive(filepath: str) -> bool:
+    # pylint: disable=too-many-return-statements
+    with tarfile.open(filepath) as tar:
+        try:
+            manifest_info = tar.getmember("manifest.json")
+            if not manifest_info.isfile():
+                return False
+            with tar.extractfile(manifest_info) as manifest_file:
+                manifest = json.load(manifest_file)
+                # There's one entry in the list for each image
+                if not isinstance(manifest, list):
+                    return False
+                for data in manifest:
+                    # Just check if this data member exists
+                    _ = tar.getmember(data["Config"])
+                    # Now check that each of the layers exist
+                    for layer in data["Layers"]:
+                        _ = tar.getmember(layer)
+                # Everything seems to exist and be in order; this is most likely a Docker archive
+                return True
+        except KeyError:
+            return False
+
+
 @surfactant.plugin.hookimpl(tryfirst=True)
 def identify_file_type(filepath: str) -> Optional[str]:
     # pylint: disable=too-many-return-statements
@@ -76,8 +102,12 @@ def identify_file_type(filepath: str) -> Optional[str]:
                 ".tar.gz",
                 ".cab.gz",
             ]:
+                if is_docker_archive(filepath):
+                    return "DOCKER_GZIP"
                 return "GZIP"
             if magic_bytes[257:265] == b"ustar\x0000" or magic_bytes[257:265] == b"ustar  \x00":
+                if is_docker_archive(filepath):
+                    return "DOCKER_TAR"
                 return "TAR"
             if magic_bytes[:4] in [b"PK\x03\x04", b"PK\x05\x06", b"PK\x07\x08"]:
                 suffix = pathlib.Path(filepath).suffix.lower()

surfactant/infoextractors/docker_image.py

@@ -0,0 +1,64 @@
+# Copyright 2023 Lawrence Livermore National Security, LLC
+# See the top-level LICENSE file for details.
+#
+# SPDX-License-Identifier: MIT
+import gzip
+import json
+import subprocess
+import tempfile
+
+from loguru import logger
+
+import surfactant.plugin
+from surfactant.sbomtypes import SBOM, Software
+
+
+def is_docker_scout_installed():
+    # Check that Docker Scout can be run
+    try:
+        result = subprocess.run(["docker", "scout"], capture_output=True, check=False)
+        if result.returncode != 0:
+            logger.warning("Install Docker Scout to scan containers for additional information")
+            return False
+        return True
+    except FileNotFoundError:
+        return False
+
+
+# Check if Docker Scout is installed when this Python module gets loaded
+disable_docker_scout = not is_docker_scout_installed()
+
+
+def supports_file(filetype: str) -> bool:
+    return filetype in ("DOCKER_TAR", "DOCKER_GZIP")
+
+
+@surfactant.plugin.hookimpl
+def extract_file_info(sbom: SBOM, software: Software, filename: str, filetype: str) -> object:
+    if disable_docker_scout or not supports_file(filetype):
+        return None
+    return extract_docker_info(filetype, filename)
+
+
+def extract_docker_info(filetype: str, filename: str) -> object:
+    if filetype == "DOCKER_GZIP":
+        with open(filename, "rb") as gzip_in:
+            gzip_data = gzip_in.read()
+        with tempfile.NamedTemporaryFile() as gzip_out:
+            gzip_out.write(gzip.decompress(gzip_data))
+            return run_docker_scout(gzip_out.name)
+    return run_docker_scout(filename)
+
+
+# Function that extract_docker_info delegates to to actually run Docker scout
+def run_docker_scout(filename: str) -> object:
+    result = subprocess.run(
+        ["docker", "scout", "sbom", "--format", "spdx", f"fs://{filename}"],
+        capture_output=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        logger.warning(f"Running Docker Scout on {filename} failed")
+        return {}
+    spdx_out = json.loads(result.stdout)
+    return {"dockerSPDX": spdx_out}

surfactant/plugin/manager.py

@@ -17,6 +17,7 @@ def _register_plugins(pm: pluggy.PluginManager) -> None:
     from surfactant.infoextractors import (
         a_out_file,
         coff_file,
+        docker_image,
         elf_file,
         java_file,
         ole_file,
@@ -42,6 +43,7 @@ def _register_plugins(pm: pluggy.PluginManager) -> None:
         id_extension,
         a_out_file,
         coff_file,
+        docker_image,
         elf_file,
         java_file,
         pe_file,