diff --git a/vulnerabilities/importers/debian.py b/vulnerabilities/importers/debian.py index b9cd3d5ec..655781994 100644 --- a/vulnerabilities/importers/debian.py +++ b/vulnerabilities/importers/debian.py @@ -1,5 +1,5 @@ # -# Copyright (c) 2017 nexB Inc. and others. All rights reserved. +# Copyright (c) nexB Inc. and others. All rights reserved. # http://nexb.com and https://github.com/nexB/vulnerablecode/ # The VulnerableCode software is licensed under the Apache License version 2.0. # Data generated with VulnerableCode require an acknowledgment. @@ -18,16 +18,17 @@ # OR CONDITIONS OF ANY KIND, either express or implied. No content created from # VulnerableCode should be considered or used as legal advice. Consult an Attorney # for any legal advice. -# VulnerableCode is a free software code scanning tool from nexB Inc. and others. +# VulnerableCode is a free software tool from nexB Inc. and others. # Visit https://github.com/nexB/vulnerablecode/ for support and download. + import dataclasses -import json +from dateutil import parser as dateparser from typing import Any from typing import List from typing import Mapping from typing import Set -from urllib.request import urlopen +import requests from packageurl import PackageURL from schema import Optional from schema import Or @@ -84,8 +85,12 @@ class DebianDataSource(DataSource): CONFIG_CLASS = DebianConfiguration def __enter__(self): - self._api_response = self._fetch() - validate_schema(self._api_response) + if self.response_is_new(): + self._api_response = self._fetch() + validate_schema(self._api_response) + + else: + self._api_response = {} def updated_advisories(self) -> Set[Advisory]: advisories = [] @@ -96,8 +101,7 @@ def updated_advisories(self) -> Set[Advisory]: return self.batch_advisories(advisories) def _fetch(self) -> Mapping[str, Any]: - with urlopen(self.config.debian_tracker_url) as response: - return json.load(response) + return requests.get(self.config.debian_tracker_url).json() def _parse(self, pkg_name: str, records: Mapping[str, Any]) -> List[Advisory]: advisories = [] @@ -146,3 +150,8 @@ def _parse(self, pkg_name: str, records: Mapping[str, Any]) -> List[Advisory]: )) return advisories + + def response_is_new(self): + date_str = requests.head(self.config.debian_tracker_url).headers.get('last-modified') + last_modified_date = dateparser.parse(date_str) + return self.config.last_run_date < last_modified_date diff --git a/vulnerabilities/tests/test_debian.py b/vulnerabilities/tests/test_debian.py index 43cc5f160..f86285544 100644 --- a/vulnerabilities/tests/test_debian.py +++ b/vulnerabilities/tests/test_debian.py @@ -23,32 +23,31 @@ import json import os from unittest.mock import patch +from unittest.mock import MagicMock from django.test import TestCase from vulnerabilities import models from vulnerabilities.import_runner import ImportRunner +from vulnerabilities.importers import DebianDataSource BASE_DIR = os.path.dirname(os.path.abspath(__file__)) -TEST_DATA = os.path.join(BASE_DIR, 'test_data/') +TEST_DATA = os.path.join(BASE_DIR, "test_data/") class DebianImportTest(TestCase): - @classmethod def setUpClass(cls) -> None: - fixture_path = os.path.join(TEST_DATA, 'debian.json') + fixture_path = os.path.join(TEST_DATA, "debian.json") with open(fixture_path) as f: cls.mock_response = json.load(f) cls.importer = models.Importer.objects.create( - name='debian_unittests', - license='', - last_run=None, - data_source='DebianDataSource', - data_source_cfg={ - 'debian_tracker_url': 'https://security.example.com/json', - }, + name="debian_unittests", + license="", + last_run="2019-08-05 13:14:17.733232+05:30", + data_source="DebianDataSource", + data_source_cfg={"debian_tracker_url": "https://security.example.com/json"}, ) @classmethod @@ -59,31 +58,43 @@ def test_import(self): runner = ImportRunner(self.importer, 5) with patch( - 'vulnerabilities.importers.DebianDataSource._fetch', - return_value=self.mock_response + "vulnerabilities.importers.DebianDataSource._fetch", return_value=self.mock_response ): - runner.run() + with patch( + "vulnerabilities.importers.DebianDataSource.response_is_new", return_value=True + ): + runner.run() assert models.Vulnerability.objects.count() == 3 assert models.VulnerabilityReference.objects.count() == 3 assert models.PackageRelatedVulnerability.objects.filter(is_vulnerable=True).count() == 2 - assert models.PackageRelatedVulnerability.objects.filter( - is_vulnerable=False).count() == 8 + assert models.PackageRelatedVulnerability.objects.filter(is_vulnerable=False).count() == 8 assert models.Package.objects.count() == 6 - self.assert_for_package('librsync', '0.9.7-10', 'jessie', cve_ids={'CVE-2014-8242'}) - self.assert_for_package('librsync', '0.9.7-10', 'buster', cve_ids={'CVE-2014-8242'}) - self.assert_for_package('mimetex', '1.50-1.1', 'stretch') - self.assert_for_package('mimetex', '1.74-1', 'stretch') - self.assert_for_package('mimetex', '1.50-1.1', 'buster') - self.assert_for_package('mimetex', '1.76-1', 'buster') + self.assert_for_package("librsync", "0.9.7-10", "jessie", cve_ids={"CVE-2014-8242"}) + self.assert_for_package("librsync", "0.9.7-10", "buster", cve_ids={"CVE-2014-8242"}) + self.assert_for_package("mimetex", "1.50-1.1", "stretch") + self.assert_for_package("mimetex", "1.74-1", "stretch") + self.assert_for_package("mimetex", "1.50-1.1", "buster") + self.assert_for_package("mimetex", "1.76-1", "buster") + + def test_response_is_new(self): + + test_data_source = self.importer.make_data_source(batch_size=1) + mock_resp = MagicMock() + mock_resp.headers = {"last-modified": "Wed, 05 Aug 2021 09:12:19 GMT"} + + with patch("vulnerabilities.importers.debian.requests.head", return_value=mock_resp): + assert test_data_source.response_is_new() is True + + mock_resp.headers = {"last-modified": "Wed, 05 Aug 2019 09:12:19 GMT"} + + with patch("vulnerabilities.importers.debian.requests.head", return_value=mock_resp): + assert test_data_source.response_is_new() is False def assert_for_package(self, name, version, release, cve_ids=None): qs = models.Package.objects.filter( - name=name, - version=version, - type='deb', - namespace='debian', + name=name, version=version, type="deb", namespace="debian", ) qs = qs.filter(qualifiers__distro=release) assert qs