Import GitHub OSV advisories incrementally

docs/source/reference_importer_overview.rst

@@ -33,3 +33,33 @@ model. The version ranges are required to be resolved into concrete ranges. Thes
 As of now, the following importers have been implemented in VulnerableCode:
 
 .. include:: ../../SOURCES.rst
+
+Incremental GitHub OSV imports
+-----------------------------------
+
+The ``github_osv_importer_v2`` pipeline uses a database ``GitImportCheckpoint`` keyed by
+pipeline ID and repository URL to remember its last successfully imported Git commit.
+The first run imports all JSON files in ``advisories/github-reviewed``. Later runs compare
+the checkpoint and current clone's HEAD with ``git diff --name-status`` and import only
+added or changed advisory JSON files. Unchanged files are not parsed or written again.
+Renames are treated as deletion plus addition, so the destination file is imported.
+
+If the previous commit is unavailable, including when a shallow clone omits it, or is no
+longer an ancestor of HEAD, the pipeline falls back to a full import. A run with no changed
+advisory files imports nothing. The existing cloning mechanism is retained; this prototype
+reduces parsing and database work rather than clone transfer size.
+
+The proposed deletion policy is to log deleted upstream paths for review and retain their
+stored advisories and history. A deleted file does not by itself establish that an advisory
+should be withdrawn. This also applies to a run containing only deletions.
+
+The checkpoint is saved as the final step, after all selected advisories have been imported
+and the temporary clone has been removed. Clone, parsing, storage, and cleanup failures
+leave the checkpoint unchanged. This importer opts into stopping on an individual storage
+failure; other importers keep their existing behavior. A retry reprocesses the delta, and
+already imported advisory content is deduplicated by the existing import mechanism.
+
+To force a full import, remove this pipeline and repository's ``GitImportCheckpoint`` row
+before the next run. Apply the database migration before running the updated importer.
+This prototype is limited to GitHub OSV. Maintainers should agree on the deletion policy
+and checkpoint timing before applying the approach to other pipelines.

vulnerabilities/migrations/0145_gitimportcheckpoint.py

@@ -0,0 +1,34 @@
+# Generated by Django 5.2.11 on 2026-11-12 16:43
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0144_remove_advisoryv2_advisory_latest_by_avid_idx_and_more"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="GitImportCheckpoint",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True, primary_key=True, serialize=False, verbose_name="ID"
+                    ),
+                ),
+                ("pipeline_id", models.CharField(max_length=600)),
+                ("repo_url", models.URLField(max_length=1024)),
+                ("commit_hash", models.CharField(max_length=64)),
+            ],
+            options={
+                "constraints": [
+                    models.UniqueConstraint(
+                        fields=("pipeline_id", "repo_url"), name="unique_git_import_checkpoint"
+                    )
+                ],
+            },
+        ),
+    ]

vulnerabilities/models.py

@@ -1970,6 +1970,21 @@ class CodeFixV2(CodeChangeV2):
     )
 
 
+class GitImportCheckpoint(models.Model):
+    """The last successfully imported Git snapshot for one pipeline and repository."""
+
+    pipeline_id = models.CharField(max_length=600)
+    repo_url = models.URLField(max_length=1024)
+    commit_hash = models.CharField(max_length=64)
+
+    class Meta:
+        constraints = [
+            models.UniqueConstraint(
+                fields=["pipeline_id", "repo_url"], name="unique_git_import_checkpoint"
+            ),
+        ]
+
+
 class PipelineRun(models.Model):
     """The Database representation of a pipeline execution."""
 

vulnerabilities/pipelines/__init__.py

@@ -279,6 +279,9 @@ class VulnerableCodeBaseImporterPipelineV2(VulnerableCodePipeline):
     ignorable_versions = []
     precedence = 0
 
+    # Incremental importers must retry failed advisories before advancing a checkpoint.
+    raise_on_import_error = False
+
     # Set this to True if computing fixed/affected package ToDo is not fruitful for this source.
     # An example of such advisory would be pipeline dedicated to collecting issues,
     # pull requests, commit messages, EPSS, exploits, etc.
@@ -351,6 +354,8 @@ def collect_and_store_advisories(self):
                     f"Failed to import advisory: {advisory!r} with error {e!r}:\n{traceback_format_exc()}",
                     level=logging.ERROR,
                 )
+                if self.raise_on_import_error:
+                    raise
                 continue
 
         self.log(f"Successfully collected {collected_advisory_count:,d} advisories")

vulnerabilities/pipelines/v2_importers/github_osv_importer.py

@@ -10,13 +10,16 @@
 import json
 from pathlib import Path
 from typing import Iterable
+from urllib.parse import quote
 
 from fetchcode.vcs import fetch_via_vcs
+from git import Repo
+from git.exc import GitError
 
 from vulnerabilities.importer import AdvisoryDataV2
+from vulnerabilities.models import GitImportCheckpoint
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
 from vulnerabilities.pipes.osv_v2 import parse_advisory_data_v3
-from vulnerabilities.utils import get_advisory_url
 
 
 class GithubOSVImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
@@ -33,22 +36,76 @@ class GithubOSVImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     repo_url = "git+https://github.com/github/advisory-database/"
 
     precedence = 200
+    raise_on_import_error = True
+    advisory_directory = "advisories/github-reviewed"
 
     @classmethod
     def steps(cls):
         return (
             cls.clone,
+            cls.select_advisory_files,
             cls.collect_and_store_advisories,
             cls.clean_downloads,
+            cls.save_checkpoint,
         )
 
     def clone(self):
-        self.log(f"Cloning `{self.repo_url}`")
+        self.advisory_files = None
+        self.log(f"Cloning {self.repo_url}")
         self.vcs_response = fetch_via_vcs(self.repo_url)
 
+    def select_advisory_files(self):
+        """Select added and changed JSON files, or use a full scan without usable history."""
+        self.advisory_files = None
+        checkpoint = GitImportCheckpoint.objects.filter(
+            pipeline_id=self.pipeline_id, repo_url=self.repo_url
+        ).first()
+        with Repo(self.vcs_response.dest_dir) as repo:
+            self.current_commit = repo.head.commit.hexsha
+            if not checkpoint:
+                self.log("No Git import checkpoint; importing all reviewed advisories")
+                return
+
+            try:
+                previous_commit = repo.commit(checkpoint.commit_hash)
+                if not repo.is_ancestor(previous_commit, self.current_commit):
+                    self.log("Git import history has diverged; importing all reviewed advisories")
+                    return
+                changes = repo.git.diff(
+                    "--name-status",
+                    "-z",
+                    "--no-renames",
+                    previous_commit.hexsha,
+                    self.current_commit,
+                    "--",
+                    self.advisory_directory,
+                ).split("\0")
+            except (GitError, ValueError) as error:
+                self.log(f"Git import checkpoint unavailable. Importing all advisories: {error}")
+                return
+
+        base_path = Path(self.vcs_response.dest_dir)
+        files = []
+        # --no-renames represents moves as deletion plus addition. NUL separators
+        # preserve spaces and newlines in filenames.
+        for change_type, relative_path in zip(changes[::2], changes[1::2]):
+            if not relative_path.endswith(".json"):
+                continue
+            if change_type == "D":
+                self.log(f"Deleted upstream advisory retained for review: {relative_path}")
+            else:
+                files.append(base_path / relative_path)
+        self.advisory_files = tuple(sorted(files))
+        self.log(f"Selected {len(files)} added or changed advisory files")
+
+    def get_advisory_files(self):
+        if getattr(self, "advisory_files", None) is None:
+            advisory_dir = Path(self.vcs_response.dest_dir) / self.advisory_directory
+            self.advisory_files = tuple(sorted(advisory_dir.rglob("*.json")))
+        return self.advisory_files
+
     def advisories_count(self):
-        advisory_dir = Path(self.vcs_response.dest_dir) / "advisories/github-reviewed"
-        return sum(1 for _ in advisory_dir.rglob("*.json"))
+        return len(self.get_advisory_files())
 
     def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
         supported_ecosystems = [
@@ -63,28 +120,37 @@ def collect_advisories(self) -> Iterable[AdvisoryDataV2]:
             "cargo",
         ]
         base_path = Path(self.vcs_response.dest_dir)
-        advisory_dir = base_path / "advisories/github-reviewed"
 
-        for file in advisory_dir.rglob("*.json"):
-            advisory_url = get_advisory_url(
-                file=file,
-                base_path=base_path,
-                url="https://github.com/github/advisory-database/blob/main/",
+        for file in self.get_advisory_files():
+            advisory_url = "https://github.com/github/advisory-database/blob/main/" + quote(
+                file.relative_to(base_path).as_posix()
             )
             with open(file) as f:
                 raw_data = json.load(f)
             advisory_text = file.read_text()
-            yield parse_advisory_data_v3(
+            advisory = parse_advisory_data_v3(
                 raw_data=raw_data,
                 supported_ecosystems=supported_ecosystems,
                 advisory_url=advisory_url,
                 advisory_text=advisory_text,
             )
+            if advisory is None:
+                raise ValueError(f"Could not parse advisory: {file}")
+            yield advisory
+
+    def save_checkpoint(self):
+        """Advance only after every selected advisory and cleanup succeeded."""
+        GitImportCheckpoint.objects.update_or_create(
+            pipeline_id=self.pipeline_id,
+            repo_url=self.repo_url,
+            defaults={"commit_hash": self.current_commit},
+        )
 
     def clean_downloads(self):
-        if self.vcs_response:
+        if getattr(self, "vcs_response", None):
             self.log("Removing cloned repository")
             self.vcs_response.delete()
+            self.vcs_response = None
 
     def on_failure(self):
         self.clean_downloads()