Rank OSV.dev findings by severity (#61)

Author: ikretz

pyproject.toml

@@ -2,6 +2,7 @@
 name = "scfw"
 dynamic = ["version"]
 dependencies = [
+    "cvss",
     "datadog-api-client",
     "inquirer",
     "packaging",
@@ -26,7 +27,9 @@ requires = ["setuptools"]
 build-backend = "setuptools.build_meta"
 
 [tool.setuptools]
-packages = ["scfw", "scfw.commands", "scfw.configure", "scfw.loggers", "scfw.verifiers"]
+packages = [
+    "scfw", "scfw.commands", "scfw.configure", "scfw.loggers", "scfw.verifiers", "scfw.verifiers.osv_verifier"
+]
 
 [tool.setuptools.dynamic]
 version = {attr = "scfw.__version__"}

requirements.txt

@@ -97,6 +97,9 @@ charset-normalizer==3.4.1 ; python_version >= "3.10" and python_version < "4" \
     --hash=sha256:fc54db6c8593ef7d4b2a331b58653356cf04f67c960f584edb7c3d8c97e8f39e \
     --hash=sha256:fd4ec41f914fa74ad1b8304bbc634b3de73d2a0889bd32076342a573e0779e00 \
     --hash=sha256:ffc9202a29ab3920fa812879e95a9e78b2465fd10be7fcbd042899695d75e616
+cvss==3.4 ; python_version >= "3.10" and python_version < "4" \
+    --hash=sha256:632353244ba3c58b53355466677edc968b9d7143c317b66271f9fd7939951ee8 \
+    --hash=sha256:d9950613758e60820f7fac37ca5f35158712f8f2ea4f6629858a60c4984fe4ef
 datadog-api-client==2.33.1 ; python_version >= "3.10" and python_version < "4" \
     --hash=sha256:26e6b44fccff5d7bc9eaeaae1d3ce4fc4c0587400d869a53c81e4cc272dc63b4 \
     --hash=sha256:9b56521fa8ea7b743024759c624804cf0435dca110f111f31f08dd68d640737a
@@ -115,9 +118,9 @@ packaging==24.2 ; python_version >= "3.10" and python_version < "4" \
 python-dateutil==2.9.0.post0 ; python_version >= "3.10" and python_version < "4" \
     --hash=sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3 \
     --hash=sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427
-python-dotenv==1.0.1 ; python_version >= "3.10" and python_version < "4" \
-    --hash=sha256:e324ee90a023d808f1959c46bcbc04446a10ced277783dc6ee09987c37ec10ca \
-    --hash=sha256:f7b63ef50f1b690dddf550d03497b66d609393b40b564ed0d674909a68ebf16a
+python-dotenv==1.1.0 ; python_version >= "3.10" and python_version < "4" \
+    --hash=sha256:41f90bc6f5f177fb41f53e87666db362025010eb28f60a01c9143bfa33a2b2d5 \
+    --hash=sha256:d7c01d9e2293916c18baf562d95698754b0dbbb5e74d457c45d4f6561fb9d55d
 readchar==4.2.1 ; python_version >= "3.10" and python_version < "4" \
     --hash=sha256:91ce3faf07688de14d800592951e5575e9c7a3213738ed01d394dcc949b79adb \
     --hash=sha256:a769305cd3994bb5fa2764aa4073452dc105a4ec39068ffe6efd3c20c60acc77
@@ -130,9 +133,9 @@ runs==1.2.2 ; python_version >= "3.10" and python_version < "4" \
 six==1.17.0 ; python_version >= "3.10" and python_version < "4" \
     --hash=sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274 \
     --hash=sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81
-typing-extensions==4.12.2 ; python_version >= "3.10" and python_version < "4" \
-    --hash=sha256:04e5ca0351e0f3f85c6853954072df659d0d13fac324d0072316b67d7794700d \
-    --hash=sha256:1a7ead55c7e559dd4dee8856e3a88b41225abfe1ce8df57b7c13915fe121ffb8
+typing-extensions==4.13.0 ; python_version >= "3.10" and python_version < "4" \
+    --hash=sha256:0a4ac55a5820789d87e297727d229866c9650f6521b64206413c4fbada24d95b \
+    --hash=sha256:c8dd92cc0d6425a97c18fbb9d1954e5ff92c1ca881a309c45f06ebc0b79058e5
 urllib3==2.3.0 ; python_version >= "3.10" and python_version < "4" \
     --hash=sha256:1cee9ad369867bfdbbb48b7dd50374c0967a0bb7710050facf0dd6911440e3df \
     --hash=sha256:f8c5449b3cf0861679ce7e0503c7b44b5ec981bec0d1d3795a07f1ba96f0204d

scfw/logger.py

@@ -5,6 +5,7 @@
 
 from abc import (ABCMeta, abstractmethod)
 from enum import Enum
+from typing_extensions import Self
 
 from scfw.ecosystem import ECOSYSTEM
 from scfw.target import InstallTarget
@@ -15,9 +16,9 @@ class FirewallAction(Enum):
     The various actions the firewall may take in response to inspecting a
     package manager command.
     """
-    ALLOW = "ALLOW"
-    ABORT = "ABORT"
-    BLOCK = "BLOCK"
+    ALLOW = 0
+    ABORT = 1
+    BLOCK = 2
 
     def __lt__(self, other) -> bool:
         """
@@ -38,15 +39,7 @@ def __lt__(self, other) -> bool:
                 f"'<' not supported between instances of '{self.__class__}' and '{other.__class__}'"
             )
 
-        match self.name, other.name:
-            case "ALLOW", "ABORT":
-                return True
-            case "ALLOW", "BLOCK":
-                return True
-            case "ABORT", "BLOCK":
-                return True
-            case _:
-                return False
+        return self.value < other.value
 
     def __str__(self) -> str:
         """
@@ -55,7 +48,26 @@ def __str__(self) -> str:
         Returns:
             A `str` representing the given `FirewallAction` suitable for printing.
         """
-        return self.value
+        return self.name
+
+    @classmethod
+    def from_string(cls, s: str) -> Self:
+        """
+        Convert a string into a `FirewallAction`.
+
+        Args:
+            s: The `str` to be converted.
+
+        Returns:
+            The `FirewallAction` referred to by the given string.
+
+        Raises:
+            ValueError: The given string does not refer to a valid `FirewallAction`.
+        """
+        mappings = {f"{action}".lower(): action for action in cls}
+        if (action := mappings.get(s.lower())):
+            return action
+        raise ValueError(f"Invalid firewall action '{s}'")
 
 
 class FirewallLogger(metaclass=ABCMeta):

scfw/loggers/dd_logger.py

@@ -66,12 +66,13 @@ def __init__(self, logger: logging.Logger):
             logger: A configured log handle to which logs will be written.
         """
         self._logger = logger
+        self._level = _DD_LOG_LEVEL_DEFAULT
 
         try:
-            self._level = FirewallAction(os.getenv(DD_LOG_LEVEL_VAR))
+            if (dd_log_level := os.getenv(DD_LOG_LEVEL_VAR)) is not None:
+                self._level = FirewallAction.from_string(dd_log_level)
         except ValueError:
             _log.warning(f"Undefined or invalid Datadog log level: using default level {_DD_LOG_LEVEL_DEFAULT}")
-            self._level = _DD_LOG_LEVEL_DEFAULT
 
     def log(
         self,

scfw/main.py

@@ -49,7 +49,7 @@ def _configure_logging(level: int):
     """
     handler = logging.StreamHandler()
     handler.addFilter(logging.Filter(name="scfw"))
-    handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
+    handler.setFormatter(logging.Formatter("[SCFW] %(levelname)s: %(message)s"))
 
     log = logging.getLogger()
     log.addHandler(handler)

scfw/verifiers/osv_verifier.py scfw/verifiers/osv_verifier/__init__.pyscfw/verifiers/osv_verifier.py renamed to scfw/verifiers/osv_verifier/__init__.py

@@ -3,13 +3,15 @@
 and malicious open source software packages.
 """
 
+import functools
 import logging
 
 import requests
 
 from scfw.ecosystem import ECOSYSTEM
 from scfw.target import InstallTarget
 from scfw.verifier import FindingSeverity, InstallTargetVerifier
+from scfw.verifiers.osv_verifier.osv_advisory import OsvAdvisory
 
 _log = logging.getLogger(__name__)
 
@@ -53,16 +55,12 @@ def verify(self, target: InstallTarget) -> list[tuple[FindingSeverity, str]]:
             requests.HTTPError:
                 An error occurred while querying an installation target against the OSV.dev API.
         """
-        def mal_finding(id: str) -> str:
+        def finding(osv: OsvAdvisory) -> str:
+            kind = "malicious package " if osv.id.startswith("MAL") else ""
+            severity_tag = f"[{osv.severity}] " if osv.severity else ""
             return (
-                f"An OSV.dev malicious package disclosure exists for package {target}:\n"
-                f"  * {_OSV_DEV_VULN_URL_PREFIX}/{id}"
-            )
-
-        def non_mal_finding(id: str) -> str:
-            return (
-                f"An OSV.dev disclosure exists for package {target}:\n"
-                f"  * {_OSV_DEV_VULN_URL_PREFIX}/{id}"
+                f"An OSV.dev {kind}disclosure exists for package {target}:\n"
+                f"  * {severity_tag}{_OSV_DEV_VULN_URL_PREFIX}/{osv.id}"
             )
 
         def error_message(e: str) -> str:
@@ -102,19 +100,27 @@ def error_message(e: str) -> str:
             if not vulns:
                 return []
 
-            osv_ids = set(filter(lambda id: id is not None, map(lambda vuln: vuln.get("id"), vulns)))
-            mal_ids = set(filter(lambda id: id.startswith("MAL"), osv_ids))
-            non_mal_ids = osv_ids - mal_ids
+            osvs = set(map(OsvAdvisory.from_json, filter(lambda vuln: vuln.get("id"), vulns)))
+            mal_osvs = set(filter(lambda osv: osv.id.startswith("MAL"), osvs))
+            non_mal_osvs = osvs - mal_osvs
+
+            osv_sort_key = functools.cmp_to_key(OsvAdvisory.compare_severities)
+            sorted_mal_osvs = sorted(mal_osvs, reverse=True, key=osv_sort_key)
+            sorted_non_mal_osvs = sorted(non_mal_osvs, reverse=True, key=osv_sort_key)
 
             return (
-                [(FindingSeverity.CRITICAL, mal_finding(id)) for id in mal_ids]
-                + [(FindingSeverity.WARNING, non_mal_finding(id)) for id in non_mal_ids]
+                [(FindingSeverity.CRITICAL, finding(osv)) for osv in sorted_mal_osvs]
+                + [(FindingSeverity.WARNING, finding(osv)) for osv in sorted_non_mal_osvs]
             )
 
         except requests.exceptions.RequestException as e:
             _log.warning(f"Failed to query OSV.dev API: returning WARNING finding for target {target}")
             return [(FindingSeverity.WARNING, error_message(str(e)))]
 
+        except Exception as e:
+            _log.warning(f"Target verification failed: returning WARNING finding for target {target}")
+            return [(FindingSeverity.WARNING, error_message(str(e)))]
+
 
 def load_verifier() -> InstallTargetVerifier:
     """