add: Add support for CleanStart OS vulnerability data provider

src/vunnel/providers/__init__.py

@@ -11,6 +11,7 @@
     arch,
     bitnami,
     chainguard,
+    cleanstart,
     chainguard_libraries,
     debian,
     echo,
@@ -41,6 +42,7 @@
 _providers: dict[str, type[provider.Provider]] = {
     # vulnerability providers
     alma.Provider.name(): alma.Provider,
+    cleanstart.Provider.name(): cleanstart.Provider,
     alpine.Provider.name(): alpine.Provider,
     amazon.Provider.name(): amazon.Provider,
     arch.Provider.name(): arch.Provider,

src/vunnel/providers/cleanstart/__init__.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+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
+
+PROVIDER_NAME = "cleanstart"
+SCHEMA = schema.OSVSchema()
+
+
+@dataclass
+class Config:
+    runtime: provider.RuntimeConfig = field(
+        default_factory=lambda: provider.RuntimeConfig(
+            result_store=result.StoreStrategy.SQLITE,
+            existing_results=provider.ResultStatePolicy.DELETE_BEFORE_WRITE,
+        ),
+    )
+    request_timeout: int = 125
+
+
+class Provider(provider.Provider):
+    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.logger.debug(f"config: {config}")
+
+        self.parser = Parser(
+            ws=self.workspace,
+            logger=self.logger,
+        )
+
+    @classmethod
+    def name(cls) -> str:
+        return PROVIDER_NAME
+
+    def update(self, last_updated: datetime.datetime | None) -> tuple[list[str], int]:
+        with self.results_writer() as writer:
+            for vuln_id, record in self.parser.get():
+                writer.write(
+                    identifier=vuln_id,
+                    schema=SCHEMA,
+                    payload=record,
+                )
+
+        return self.parser.urls, len(writer)

src/vunnel/providers/cleanstart/parser.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+
+from vunnel import workspace
+
+ADVISORIES_REPO = "https://github.com/cleanstart-dev/cleanstart-security-advisories.git"
+ADVISORIES_DIR = "advisories"
+
+
+class Parser:
+    def __init__(self, ws: workspace.Workspace, logger: logging.Logger | None = None):
+        self.workspace = ws
+        self.urls = [ADVISORIES_REPO]
+
+        if not logger:
+            logger = logging.getLogger(self.__class__.__name__)
+        self.logger = logger
+
+    def get(self):
+        yield from self._fetch_and_parse()
+
+    def _fetch_and_parse(self):
+        clone_dir = os.path.join(self.workspace.input_path, "cleanstart-security-advisories")
+
+        if os.path.exists(clone_dir):
+            self.logger.info("updating existing advisory repo")
+            subprocess.run(["git", "-C", clone_dir, "pull"], check=True, capture_output=True)
+        else:
+            self.logger.info(f"cloning advisory repo from {ADVISORIES_REPO}")
+            subprocess.run(["git", "clone", "--depth=1", ADVISORIES_REPO, clone_dir], check=True, capture_output=True)
+
+        advisories_path = os.path.join(clone_dir, ADVISORIES_DIR)
+
+        # walk all subdirectories (e.g. 2025/, 2026/)
+        json_files = []
+        for root, dirs, files in os.walk(advisories_path):
+            for filename in files:
+                if filename.endswith(".json"):
+                    json_files.append(os.path.join(root, filename))
+
+        self.logger.info(f"found {len(json_files)} advisories")
+
+        for filepath in json_files:
+            try:
+                with open(filepath, encoding="utf-8") as f:
+                    record = json.load(f)
+                vuln_id = record["id"]
+                yield vuln_id, record
+            except Exception as e:
+                self.logger.warning(f"skipping {filepath}: {e}")