Merge pull request #241 from sbs2001/use_last_date

Check date before downloading data

Author: sbs2001

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
@@ -85,8 +86,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 = []
@@ -97,8 +102,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 = []
@@ -150,3 +154,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

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