diff --git a/docs/source/reference_importer_overview.rst b/docs/source/reference_importer_overview.rst index 04e56dd39..edec3b17e 100644 --- a/docs/source/reference_importer_overview.rst +++ b/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. diff --git a/vulnerabilities/migrations/0145_gitimportcheckpoint.py b/vulnerabilities/migrations/0145_gitimportcheckpoint.py new file mode 100644 index 000000000..d35a21b45 --- /dev/null +++ b/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" + ) + ], + }, + ), + ] diff --git a/vulnerabilities/models.py b/vulnerabilities/models.py index 521f0ca6a..68d5c40f2 100644 --- a/vulnerabilities/models.py +++ b/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.""" diff --git a/vulnerabilities/pipelines/__init__.py b/vulnerabilities/pipelines/__init__.py index 5940363b0..f7cc15aad 100644 --- a/vulnerabilities/pipelines/__init__.py +++ b/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") diff --git a/vulnerabilities/pipelines/v2_importers/github_osv_importer.py b/vulnerabilities/pipelines/v2_importers/github_osv_importer.py index 34775fc8c..ebbd3f348 100644 --- a/vulnerabilities/pipelines/v2_importers/github_osv_importer.py +++ b/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() diff --git a/vulnerabilities/tests/pipelines/v2_importers/test_github_osv_importer_v2.py b/vulnerabilities/tests/pipelines/v2_importers/test_github_osv_importer_v2.py index ae9b6f75f..3fced43fb 100644 --- a/vulnerabilities/tests/pipelines/v2_importers/test_github_osv_importer_v2.py +++ b/vulnerabilities/tests/pipelines/v2_importers/test_github_osv_importer_v2.py @@ -9,12 +9,359 @@ import json from pathlib import Path +from unittest.mock import patch import pytest +from fetchcode.vcs import VCSResponse +from git import Actor +from git import Repo -from vulnerabilities.importer import AdvisoryData from vulnerabilities.importer import AdvisoryDataV2 +from vulnerabilities.models import AdvisoryV2 +from vulnerabilities.models import GitImportCheckpoint from vulnerabilities.pipelines.v2_importers.github_osv_importer import GithubOSVImporterPipeline +from vulnerabilities.pipes.advisory import insert_advisory_v2 +from vulnerabilities.pipes.osv_v2 import parse_advisory_data_v3 + +CHANGED_ID = "GHSA-aaaa-bbbb-cccc" +DELETED_ID = "GHSA-dddd-eeee-ffff" +UNCHANGED_ID = "GHSA-gggg-hhhh-jjjj" +ADDED_ID = "GHSA-kkkk-mmmm-nnnn" + + +@pytest.fixture +def local_osv_repository(tmp_path): + """A real Git history containing reviewed advisories and unrelated files.""" + with Repo.init(tmp_path / "source", initial_branch="main") as repo: + actor = Actor("OSV Import Test", "osv-import@example.com") + + def write_advisory(advisory_id, summary="Initial summary", relative_path=None): + relative_path = relative_path or ( + f"advisories/github-reviewed/2026/{advisory_id}/{advisory_id}.json" + ) + file = Path(repo.working_dir) / relative_path + file.parent.mkdir(parents=True, exist_ok=True) + file.write_text(json.dumps({"id": advisory_id, "summary": summary, "affected": []})) + return file + + def commit(message): + repo.git.add("--all") + return repo.index.commit(message, author=actor, committer=actor).hexsha + + for advisory_id in (CHANGED_ID, DELETED_ID, UNCHANGED_ID): + write_advisory(advisory_id) + write_advisory("GHSA-pppp-qqqq-rrrr", relative_path="advisories/unreviewed/ignored.json") + (Path(repo.working_dir) / "README.md").write_text("Not an advisory") + initial_commit = commit("Initial advisories") + yield repo, write_advisory, commit, initial_commit + + +@pytest.fixture +def local_importer_factory(monkeypatch, tmp_path, local_osv_repository): + repo, _, _, _ = local_osv_repository + clone_paths = [] + repo_url = "git+" + Path(repo.working_dir).as_uri() + + def make_importer(**clone_options): + def fetch_local_repository(url): + assert url == repo_url + clone_path = tmp_path / f"checkout-{len(clone_paths)}" + clone_paths.append(clone_path) + with Repo.clone_from(repo.working_dir, clone_path, no_local=True, **clone_options): + pass + return VCSResponse(dest_dir=str(clone_path), vcs_type="git", domain="") + + monkeypatch.setattr( + "vulnerabilities.pipelines.v2_importers.github_osv_importer.fetch_via_vcs", + fetch_local_repository, + ) + importer = GithubOSVImporterPipeline() + importer.repo_url = repo_url + return importer + + return make_importer, clone_paths + + +def checkpoint_for(importer): + return GitImportCheckpoint.objects.get( + pipeline_id=importer.pipeline_id, repo_url=importer.repo_url + ) + + +@pytest.mark.django_db +def test_first_import_reads_all_reviewed_advisories(local_importer_factory, local_osv_repository): + make_importer, clone_paths = local_importer_factory + _, _, _, initial_commit = local_osv_repository + importer = make_importer() + + assert importer.execute() == (0, "") + + assert set(AdvisoryV2.objects.values_list("advisory_id", flat=True)) == { + CHANGED_ID, + DELETED_ID, + UNCHANGED_ID, + } + assert checkpoint_for(importer).commit_hash == initial_commit + assert importer.advisories_count() == 3 + assert all(not path.exists() for path in clone_paths) + + +@pytest.mark.django_db +def test_incremental_import_adds_updates_and_reports_deletions( + local_importer_factory, local_osv_repository +): + make_importer, _ = local_importer_factory + repo, write_advisory, commit, _ = local_osv_repository + assert make_importer().execute() == (0, "") + write_advisory(CHANGED_ID, "Updated summary") + write_advisory(ADDED_ID) + write_advisory(DELETED_ID).unlink() + (Path(repo.working_dir) / "README.md").write_text("Updated documentation") + write_advisory("GHSA-pppp-qqqq-rrrr", relative_path="advisories/unreviewed/new.json") + new_commit = commit("Add, update, and delete advisories") + + importer = make_importer() + with patch( + "vulnerabilities.pipelines.v2_importers.github_osv_importer.parse_advisory_data_v3", + wraps=parse_advisory_data_v3, + ) as parse: + assert importer.execute() == (0, "") + + assert {call.kwargs["raw_data"]["id"] for call in parse.call_args_list} == { + CHANGED_ID, + ADDED_ID, + } + assert importer.advisories_count() == 2 + assert checkpoint_for(importer).commit_hash == new_commit + latest = AdvisoryV2.objects.latest_for_avid(f"github_osv/{CHANGED_ID}") + assert latest.summary == "Updated summary" + assert AdvisoryV2.objects.filter(advisory_id=CHANGED_ID).count() == 2 + assert AdvisoryV2.objects.filter(advisory_id=UNCHANGED_ID).count() == 1 + assert AdvisoryV2.objects.filter(advisory_id=ADDED_ID).count() == 1 + assert AdvisoryV2.objects.filter(advisory_id=DELETED_ID, is_latest=True).exists() + assert any( + "Deleted upstream advisory retained for review" in log for log in importer.execution_log + ) + + +@pytest.mark.django_db +def test_unchanged_repository_parses_no_advisories(local_importer_factory): + make_importer, _ = local_importer_factory + assert make_importer().execute() == (0, "") + importer = make_importer() + with patch( + "vulnerabilities.pipelines.v2_importers.github_osv_importer.parse_advisory_data_v3" + ) as parse: + assert importer.execute() == (0, "") + parse.assert_not_called() + assert importer.advisories_count() == 0 + assert AdvisoryV2.objects.count() == 3 + + +@pytest.mark.django_db +def test_deletion_only_run_retains_advisory_and_saves_checkpoint( + local_importer_factory, local_osv_repository +): + make_importer, _ = local_importer_factory + _, write_advisory, commit, _ = local_osv_repository + assert make_importer().execute() == (0, "") + write_advisory(DELETED_ID).unlink() + new_commit = commit("Delete one advisory") + importer = make_importer() + + with patch( + "vulnerabilities.pipelines.v2_importers.github_osv_importer.parse_advisory_data_v3" + ) as parse: + assert importer.execute() == (0, "") + + parse.assert_not_called() + assert importer.advisories_count() == 0 + assert AdvisoryV2.objects.filter(advisory_id=DELETED_ID, is_latest=True).exists() + assert checkpoint_for(importer).commit_hash == new_commit + + +@pytest.mark.django_db +def test_missing_previous_commit_falls_back_to_full_import(local_importer_factory): + make_importer, _ = local_importer_factory + importer = make_importer() + assert importer.execute() == (0, "") + GitImportCheckpoint.objects.update(commit_hash="f" * 40) + importer = make_importer() + + assert importer.execute() == (0, "") + + assert importer.advisories_count() == 3 + assert AdvisoryV2.objects.count() == 3 + assert checkpoint_for(importer).commit_hash == importer.current_commit + assert any("checkpoint unavailable" in log for log in importer.execution_log) + + +@pytest.mark.django_db +def test_shallow_history_falls_back_to_full_import(local_importer_factory, local_osv_repository): + make_importer, _ = local_importer_factory + _, write_advisory, commit, _ = local_osv_repository + assert make_importer().execute() == (0, "") + write_advisory(ADDED_ID) + commit("Add advisory outside shallow history") + importer = make_importer(depth=1) + + assert importer.execute() == (0, "") + + assert importer.advisories_count() == 4 + assert AdvisoryV2.objects.count() == 4 + assert checkpoint_for(importer).commit_hash == importer.current_commit + + +@pytest.mark.django_db +def test_rewritten_history_falls_back_to_full_import(local_importer_factory, local_osv_repository): + make_importer, _ = local_importer_factory + repo, write_advisory, commit, _ = local_osv_repository + assert make_importer().execute() == (0, "") + repo.git.checkout("--orphan", "rewritten") + write_advisory(ADDED_ID) + commit("Replace repository history") + importer = make_importer() + + assert importer.execute() == (0, "") + + assert importer.advisories_count() == 4 + assert any("history has diverged" in log for log in importer.execution_log) + assert checkpoint_for(importer).commit_hash == importer.current_commit + + +@pytest.mark.django_db +def test_renamed_advisory_is_imported_from_new_path(local_importer_factory, local_osv_repository): + make_importer, _ = local_importer_factory + _, write_advisory, commit, _ = local_osv_repository + assert make_importer().execute() == (0, "") + old_file = write_advisory(CHANGED_ID) + new_file = old_file.parent / "renamed advisory\nfile.json" + old_file.rename(new_file) + commit("Move advisory to a new filename") + importer = make_importer() + + assert importer.execute() == (0, "") + + assert importer.advisories_count() == 1 + assert importer.advisory_files[0].name == new_file.name + assert AdvisoryV2.objects.latest_for_avid(f"github_osv/{CHANGED_ID}").url.endswith( + "renamed%20advisory%0Afile.json" + ) + + +@pytest.mark.django_db +def test_failed_storage_keeps_checkpoint_and_retries(local_importer_factory, local_osv_repository): + make_importer, clone_paths = local_importer_factory + _, write_advisory, commit, initial_commit = local_osv_repository + assert make_importer().execute() == (0, "") + write_advisory(CHANGED_ID, "Updated summary") + write_advisory(ADDED_ID) + new_commit = commit("Two advisories, one import will fail") + + def fail_one_advisory(**kwargs): + if kwargs["advisory"].advisory_id == ADDED_ID: + raise RuntimeError("Cannot store advisory") + return insert_advisory_v2(**kwargs) + + importer = make_importer() + with patch("vulnerabilities.pipelines.insert_advisory_v2", side_effect=fail_one_advisory): + exitcode, error = importer.execute() + assert exitcode == 1 + assert "Cannot store advisory" in error + assert checkpoint_for(importer).commit_hash == initial_commit + assert not AdvisoryV2.objects.filter(advisory_id=ADDED_ID).exists() + assert AdvisoryV2.objects.filter(advisory_id=CHANGED_ID).count() == 2 + assert all(not path.exists() for path in clone_paths) + + retry = make_importer() + assert retry.execute() == (0, "") + assert checkpoint_for(retry).commit_hash == new_commit + assert AdvisoryV2.objects.filter(advisory_id=ADDED_ID).count() == 1 + assert AdvisoryV2.objects.filter(advisory_id=CHANGED_ID).count() == 2 + + +@pytest.mark.django_db +def test_clone_failure_keeps_checkpoint(local_importer_factory, local_osv_repository): + make_importer, _ = local_importer_factory + _, _, _, initial_commit = local_osv_repository + assert make_importer().execute() == (0, "") + importer = make_importer() + + with patch( + "vulnerabilities.pipelines.v2_importers.github_osv_importer.fetch_via_vcs", + side_effect=OSError("Repository unavailable"), + ): + exitcode, error = importer.execute() + + assert exitcode == 1 + assert "Repository unavailable" in error + assert checkpoint_for(importer).commit_hash == initial_commit + + +@pytest.mark.django_db +def test_cleanup_failure_keeps_checkpoint(local_importer_factory, local_osv_repository): + make_importer, clone_paths = local_importer_factory + _, write_advisory, commit, initial_commit = local_osv_repository + assert make_importer().execute() == (0, "") + write_advisory(ADDED_ID) + commit("Add advisory before cleanup failure") + importer = make_importer() + delete = VCSResponse.delete + attempts = 0 + + def fail_first_cleanup(response): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise OSError("Cleanup failed") + delete(response) + + with patch.object(VCSResponse, "delete", autospec=True, side_effect=fail_first_cleanup): + exitcode, error = importer.execute() + + assert exitcode == 1 + assert "Cleanup failed" in error + assert checkpoint_for(importer).commit_hash == initial_commit + assert all(not path.exists() for path in clone_paths) + + +@pytest.mark.django_db +@pytest.mark.parametrize("invalid_content", ["{invalid json", "{}"]) +def test_parse_failure_does_not_create_checkpoint( + local_importer_factory, local_osv_repository, invalid_content +): + make_importer, clone_paths = local_importer_factory + _, write_advisory, commit, _ = local_osv_repository + write_advisory(CHANGED_ID).write_text(invalid_content) + commit("Malformed advisory") + importer = make_importer() + + exitcode, _ = importer.execute() + + assert exitcode == 1 + assert not GitImportCheckpoint.objects.exists() + assert all(not path.exists() for path in clone_paths) + + +@pytest.mark.django_db +def test_checkpoints_are_scoped_to_pipeline_and_repository(local_importer_factory): + make_importer, _ = local_importer_factory + GitImportCheckpoint.objects.create( + pipeline_id=GithubOSVImporterPipeline.pipeline_id, + repo_url="git+https://example.com/other-repository", + commit_hash="a" * 40, + ) + importer = make_importer() + GitImportCheckpoint.objects.create( + pipeline_id="another_pipeline", + repo_url=importer.repo_url, + commit_hash="b" * 40, + ) + + assert importer.execute() == (0, "") + + assert importer.advisories_count() == 3 + assert GitImportCheckpoint.objects.count() == 3 @pytest.fixture