Put network calls in try/except block for redhat importer

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/importers/redhat.py

@@ -44,6 +44,14 @@
 requests_session = requests_with_5xx_retry(max_retries=5, backoff_factor=1)
 
 
+def get_response(url):
+    try:
+        return requests_session.get(url)
+    except Exception as e:
+        logger.error(f"Failed to fetch results from {url} {e}")
+        return None
+
+
 def fetch_list_of_cves() -> Iterable[List[Dict]]:
     page_no = 1
     cve_data = None
@@ -64,14 +72,12 @@ def fetch_list_of_cves() -> Iterable[List[Dict]]:
         yield cve_data
 
 
-def get_bugzilla_data(bugzilla):
-    return requests_session.get(f"https://bugzilla.redhat.com/rest/bug/{bugzilla}").json()
-
-
-def get_rhsa_data(rh_adv):
-    return requests_session.get(
-        f"https://access.redhat.com/hydra/rest/securitydata/cvrf/{rh_adv}.json"
-    ).json()
+def get_data_from_url(url):
+    try:
+        return requests_session.get(url).json()
+    except Exception as e:
+        logger.error(f"Failed to fetch results from {url} {e!r}")
+        return
 
 
 class RedhatImporter(Importer):
@@ -112,9 +118,11 @@ def to_advisory(advisory_data):
     bugzilla = advisory_data.get("bugzilla")
     if bugzilla:
         url = "https://bugzilla.redhat.com/show_bug.cgi?id={}".format(bugzilla)
-        bugzilla_data = get_bugzilla_data(bugzilla)
+        bugzilla_url = f"https://bugzilla.redhat.com/rest/bug/{bugzilla}"
+        bugzilla_data = get_data_from_url(bugzilla_url)
         if (
-            bugzilla_data.get("bugs")
+            bugzilla_data
+            and bugzilla_data.get("bugs")
             and len(bugzilla_data["bugs"])
             and bugzilla_data["bugs"][0].get("severity")
         ):
@@ -141,8 +149,10 @@ def to_advisory(advisory_data):
             continue
 
         if "RHSA" in rh_adv.upper():
-            rhsa_data = get_rhsa_data(rh_adv)
-
+            rhsa_url = f"https://access.redhat.com/hydra/rest/securitydata/cvrf/{rh_adv}.json"
+            rhsa_data = get_data_from_url(rhsa_url)
+            if not rhsa_data:
+                continue
             rhsa_aggregate_severities = []
             if rhsa_data.get("cvrfdoc"):
                 # not all RHSA errata have a corresponding CVRF document

vulnerabilities/tests/test_data/redhat/redhat-input.json

@@ -20,7 +20,8 @@
     "public_date": "2022-04-19T20:00:00Z",
     "advisories": [
       "RHSA-2022:1439",
-      "RHSA-2022:1437"
+      "RHSA-2022:1437",
+      "RHSA-2022:1436"
     ],
     "bugzilla": 2075788,
     "bugzilla_description": "CVE-2022-21426 OpenJDK: Unbounded memory allocation when compiling crafted XPath expressions (JAXP, 8270504)",

vulnerabilities/tests/test_redhat_importer.py

@@ -46,25 +46,23 @@ def test_rpm_to_purl():
 
 
 @patch("vulnerabilities.importers.redhat.fetch_list_of_cves")
-@patch("vulnerabilities.importers.redhat.get_rhsa_data")
-@patch("vulnerabilities.importers.redhat.get_bugzilla_data")
-def test_redhat_importer(bugzilla, rhsa, fetcher):
+@patch("vulnerabilities.importers.redhat.get_data_from_url")
+def test_redhat_importer(get_data_from_url, fetcher, caplog):
     redhat_importer = redhat.RedhatImporter()
     response_file = os.path.join(TEST_DATA, f"redhat-input.json")
 
     with open(response_file) as f:
         fetcher.return_value = [json.load(f)]
     bugzilla_2075788_response_file = os.path.join(TEST_DATA, f"bugzilla-2075788.json")
     bugzilla_2077736_response_file = os.path.join(TEST_DATA, f"bugzilla-2077736.json")
-    bugzilla.side_effect = [
-        json.load(open(bugzilla_2075788_response_file)),
-        json.load(open(bugzilla_2077736_response_file)),
-    ]
     rhsa_1437 = os.path.join(TEST_DATA, f"RHSA-2022:1437.json")
     rhsa_1439 = os.path.join(TEST_DATA, f"RHSA-2022:1439.json")
-    rhsa.side_effect = [
-        json.load(open(rhsa_1437)),
+    get_data_from_url.side_effect = [
+        json.load(open(bugzilla_2075788_response_file)),
+        json.load(open(bugzilla_2077736_response_file)),
         json.load(open(rhsa_1439)),
+        json.load(open(rhsa_1437)),
+        None,
     ]
     expected_file = os.path.join(TEST_DATA, f"redhat-expected.json")
     imported_data = list(redhat_importer.advisory_data())