Add rootio provider for Root CVE feed

README.md

@@ -21,6 +21,7 @@ Supported data sources:
 - SLES (https://ftp.suse.com/pub/projects/security/oval)
 - Ubuntu (https://launchpad.net/ubuntu-cve-tracker)
 - Wolfi (https://packages.wolfi.dev)
+- Root (https://api.root.io/external/cve_feed)
 
 
 ## Installation
@@ -56,17 +57,23 @@ List the available vulnerability data providers:
 ```
 $ vunnel list
 
+alma
 alpine
 amazon
+bitnami
 chainguard
 debian
 echo
+epss
 github
+kev
 mariner
 minimos
 nvd
 oracle
 rhel
+rocky
+rootio
 sles
 ubuntu
 wolfi

src/vunnel/cli/config.py

@@ -58,6 +58,7 @@ class Providers:
     oracle: providers.oracle.Config = field(default_factory=providers.oracle.Config)
     rhel: providers.rhel.Config = field(default_factory=providers.rhel.Config)
     rocky: providers.rocky.Config = field(default_factory=providers.rocky.Config)
+    rootio: providers.rootio.Config = field(default_factory=providers.rootio.Config)
     sles: providers.sles.Config = field(default_factory=providers.sles.Config)
     ubuntu: providers.ubuntu.Config = field(default_factory=providers.ubuntu.Config)
     wolfi: providers.wolfi.Config = field(default_factory=providers.wolfi.Config)

src/vunnel/providers/__init__.py

@@ -21,6 +21,7 @@
     oracle,
     rhel,
     rocky,
+    rootio,
     sles,
     ubuntu,
     wolfi,
@@ -43,6 +44,7 @@
     oracle.Provider.name(): oracle.Provider,
     rhel.Provider.name(): rhel.Provider,
     rocky.Provider.name(): rocky.Provider,
+    rootio.Provider.name(): rootio.Provider,
     sles.Provider.name(): sles.Provider,
     ubuntu.Provider.name(): ubuntu.Provider,
     wolfi.Provider.name(): wolfi.Provider,

src/vunnel/providers/rootio/__init__.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+from vunnel import provider, result, schema
+
+from .parser import Parser
+
+if TYPE_CHECKING:
+    import datetime
+
+
+@dataclass
+class Config:
+    runtime: provider.RuntimeConfig = field(
+        default_factory=lambda: provider.RuntimeConfig(
+            result_store=result.StoreStrategy.SQLITE,
+            existing_results=result.ResultStatePolicy.DELETE_BEFORE_WRITE,
+        ),
+    )
+    request_timeout: int = 125
+
+
+class Provider(provider.Provider):
+    __schema__ = schema.OSSchema()
+    __distribution_version__ = int(__schema__.major_version)
+
+    _url = "https://api.root.io/external/cve_feed"
+
+    def __init__(self, root: str, config: Config | None = None):
+        if not config:
+            config = Config()
+        super().__init__(root, runtime_cfg=config.runtime)
+        self.config = config
+
+        self.parser = Parser(
+            workspace=self.workspace,
+            url=self._url,
+            download_timeout=self.config.request_timeout,
+            logger=self.logger,
+        )
+
+    @classmethod
+    def name(cls) -> str:
+        return "rootio"
+
+    def update(self, last_updated: datetime.datetime | None) -> tuple[list[str], int]:
+        with self.results_writer() as writer:
+            for namespace, vuln_id, record in self.parser.get():
+                writer.write(
+                    identifier=os.path.join(namespace, vuln_id.lower()),
+                    schema=self.__schema__,
+                    payload=record,
+                )
+
+        return [self._url], len(writer)

src/vunnel/providers/rootio/parser.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import copy
+import logging
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+import orjson
+
+from vunnel.utils import http_wrapper as http
+from vunnel.utils import vulnerability
+
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
+    from vunnel import workspace
+
+
+class Parser:
+    _data_dir = "rootio-data"
+    _data_filename = "cve_feed.json"
+
+    def __init__(
+        self,
+        workspace: workspace.Workspace,
+        url: str,
+        download_timeout: int = 125,
+        logger: logging.Logger | None = None,
+    ):
+        self.download_timeout = download_timeout
+        self.data_dir_path = Path(workspace.input_path) / self._data_dir
+        self.url = url
+
+        if not logger:
+            logger = logging.getLogger(self.__class__.__name__)
+        self.logger = logger
+
+    def _download(self) -> None:
+        if not os.path.exists(self.data_dir_path):
+            os.makedirs(self.data_dir_path, exist_ok=True)
+
+        try:
+            self.logger.info(f"downloading Root.io CVE feed from {self.url}")
+            r = http.get(self.url, self.logger, stream=True, timeout=self.download_timeout)
+            file_path = self.data_dir_path / self._data_filename
+            with open(file_path, "wb") as fp:
+                for chunk in r.iter_content():
+                    fp.write(chunk)
+        except Exception:
+            self.logger.exception(f"Error downloading Root.io data from {self.url}")
+            raise
+
+    def _normalize(self, distro_name: str, distro_data: dict[str, Any]) -> dict[str, dict[str, Any]]:
+        """Transform Root.io data into OS schema format"""
+        vuln_dict = {}
+
+        distro_version = distro_data.get("distroversion", "unknown")
+        namespace = f"rootio:distro:{distro_name}:{distro_version}"
+
+        for package_data in distro_data.get("packages", []):
+            pkg_info = package_data.get("pkg", {})
+            package_name = pkg_info.get("name", "")
+
+            for cve_id, cve_info in pkg_info.get("cves", {}).items():
+                if cve_id not in vuln_dict:
+                    record = copy.deepcopy(vulnerability.vulnerability_element)
+                    record["Vulnerability"]["Name"] = cve_id
+                    record["Vulnerability"]["NamespaceName"] = namespace
+
+                    # Build reference links
+                    reference_links = vulnerability.build_reference_links(cve_id)
+                    record["Vulnerability"]["Link"] = reference_links[0] if reference_links else ""
+
+                    record["Vulnerability"]["Severity"] = "Unknown"
+                    record["Vulnerability"]["Description"] = f"Vulnerability {cve_id} in {package_name}"
+                    record["Vulnerability"]["FixedIn"] = []
+                    record["Vulnerability"]["Metadata"] = {
+                        "CVE": [{"Name": cve_id, "Link": reference_links[0] if reference_links else ""}],
+                    }
+                    vuln_dict[cve_id] = record
+
+                # Add fixed version info
+                cve_record = vuln_dict[cve_id]
+                fixed_versions = cve_info.get("fixed_versions", [])
+
+                # Determine version format based on distro
+                version_format = "dpkg"  # default
+                if distro_name == "alpine":
+                    version_format = "apk"
+                elif distro_name in ["rhel", "centos", "rocky", "alma"]:
+                    version_format = "rpm"
+
+                for fixed_version in fixed_versions:
+                    cve_record["Vulnerability"]["FixedIn"].append({
+                        "Name": package_name,
+                        "Version": fixed_version,
+                        "VersionFormat": version_format,
+                        "NamespaceName": namespace,
+                        "VendorAdvisory": {"NoAdvisory": True},
+                    })
+
+                # If no fixed versions, add unfixed entry
+                if not fixed_versions:
+                    cve_record["Vulnerability"]["FixedIn"].append({
+                        "Name": package_name,
+                        "Version": "",  # Empty version indicates no fix available
+                        "VersionFormat": version_format,
+                        "NamespaceName": namespace,
+                        "VendorAdvisory": {"NoAdvisory": True},
+                    })
+
+        return vuln_dict
+
+    def get(self) -> Generator[tuple[str, str, dict[str, Any]], None, None]:
+        """Download, parse and yield Root.io vulnerability records"""
+        # Download the data
+        self._download()
+
+        # Load the JSON data
+        with open(self.data_dir_path / self._data_filename) as fh:
+            feed_data = orjson.loads(fh.read())
+
+        # Process each distribution
+        for distro_name, distro_list in feed_data.items():
+            for distro_data in distro_list:
+                distro_version = distro_data.get("distroversion", "unknown")
+                namespace = f"rootio:distro:{distro_name}:{distro_version}"
+
+                vuln_records = self._normalize(distro_name, distro_data)
+
+                for vuln_id, record in vuln_records.items():
+                    yield namespace, vuln_id, record

tests/quality/config.yaml

@@ -345,3 +345,21 @@ tests:
       - wolfi:distro:wolfi:rolling
     validations:
       - *default-validations
+
+  - provider: rootio
+    additional_providers:
+      - name: nvd
+        use_cache: true
+    images:
+      - docker.io/alpine:3.17@sha256:f271e74b17ced29b915d351685fd4644785c6d1559dd1f2d4189a5e851ef753a
+      - docker.io/debian:11@sha256:e538a2f0566efc44db21503277c7312a142f4d0dedc5d2886932b92626104bff
+    expected_namespaces:
+      - rootio:distro:alpine:3.17
+      - rootio:distro:alpine:3.18
+      - rootio:distro:alpine:3.19
+      - rootio:distro:debian:11
+      - rootio:distro:debian:12
+      - rootio:distro:ubuntu:20.04
+      - rootio:distro:ubuntu:22.04
+    validations:
+      - *default-validations

tests/unit/cli/test_cli.py

@@ -473,6 +473,23 @@ def test_config(monkeypatch) -> None:
       result_store: sqlite
       skip_download: false
       skip_newer_archive_check: false
+  rootio:
+    request_timeout: 125
+    runtime:
+      existing_input: keep
+      existing_results: delete-before-write
+      import_results_enabled: false
+      import_results_host: ''
+      import_results_path: providers/{provider_name}/listing.json
+      on_error:
+        action: fail
+        input: keep
+        results: keep
+        retry_count: 3
+        retry_delay: 5
+      result_store: sqlite
+      skip_download: false
+      skip_newer_archive_check: false
   sles:
     allow_versions:
       - '11'

tests/unit/test_provider.py

@@ -980,6 +980,7 @@ def test_provider_versions(tmpdir):
         "oracle": 1,
         "rhel": 1,
         "rocky": 1,
+        "rootio": 1,
         "sles": 1,
         "ubuntu": 3,
         "wolfi": 1,
@@ -1015,6 +1016,7 @@ def test_provider_distribution_versions(tmpdir):
         "oracle": 1,
         "rhel": 1,
         "rocky": 1,
+        "rootio": 1,
         "sles": 1,
         "ubuntu": 1,
         "wolfi": 1,