refactor: migrate from urllib to requests (fixes intel#1311) (intel#1569

)

* refactor: migrate from urllib to requests

* chore(deps): add request in requirements

* fix: readlines() in sqlite checker

* refactor: `requests.exceptions.HTTPError` -> `requests.ConnectionError`

* refactor: requests.ConnectionError -> requests.RequestException

cve_bin_tool/available_fix/debian_cve_tracker.py

@@ -1,11 +1,12 @@
 # Copyright (C) 2021 Intel Corporation
 # SPDX-License-Identifier: GPL-3.0-or-later
 
-from json import dump, load, loads
+from json import dump, load
 from os.path import exists, expanduser, getmtime, join
 from time import time
 from typing import Dict
-from urllib import request
+
+import requests
 
 from cve_bin_tool.cve_scanner import CVEData
 from cve_bin_tool.log import LOGGER
@@ -96,8 +97,7 @@ def update_json():
     """Update the Debian CVE JSON file"""
 
     LOGGER.info("Updating Debian CVE JSON file for checking available fixes.")
-    response = request.urlopen(JSON_URL).read().decode("utf-8")  # nosec - static url
-    response = loads(response)
+    response = requests.get(JSON_URL).json()
     with open(DEB_CVE_JSON_PATH, "w") as debian_json:
         dump(response, debian_json, indent=4)
         LOGGER.info("Debian CVE JSON file for checking available fixes is updated.")

cve_bin_tool/available_fix/redhat_cve_tracker.py

@@ -1,10 +1,10 @@
 # Copyright (C) 2021 Intel Corporation
 # SPDX-License-Identifier: GPL-3.0-or-later
 
-from json import loads
 from re import search, split
 from typing import Dict
-from urllib import error, request
+
+import requests
 
 from cve_bin_tool.cve_scanner import CVEData
 from cve_bin_tool.log import LOGGER
@@ -74,10 +74,9 @@ def cve_info(
 
     def get_data(self, cve_number: str, product: str):
         try:
-            full_query = f"{RH_CVE_API}/{cve_number}.json"  # static https url above
-            response = request.urlopen(full_query).read().decode("utf-8")  # nosec
-            return loads(response)
-        except error.HTTPError as e:
+            full_query = f"{RH_CVE_API}/{cve_number}.json"
+            return requests.get(full_query).json()
+        except requests.HTTPError as e:
             LOGGER.debug(e)
 
     def parse_package_data(self, package_data: str) -> str:

cve_bin_tool/checkers/sqlite.py

@@ -12,8 +12,8 @@
 
 """
 import re
-import urllib.error as error
-import urllib.request as request
+
+import requests
 
 from cve_bin_tool.checkers import Checker
 from cve_bin_tool.log import LOGGER
@@ -31,13 +31,11 @@ def get_version_map():
         re.compile(r'"*(\d{4}-\d{2}-\d{2} \d+:\d+:\d+ [\w]+)"*'),
     ]
     try:
-        response = request.urlopen(changeurl)  # nosec - static url above
-        lines = response.readlines()
+        response = requests.get(changeurl).text
+        lines = response.splitlines()
 
         last_version = "UNKNOWN"
-        for line_encoded in lines:
-            line = line_encoded.decode("UTF-8")
-
+        for line in lines:
             ver_match = version_pattern.search(line)
             if ver_match:
                 last_version = ver_match.group(1)
@@ -47,7 +45,7 @@ def get_version_map():
                     version_map.append([last_version, id_match.group(1)])
                     break
 
-    except error.URLError as err:
+    except requests.RequestException as err:
         LOGGER.error("Could not fetch " + changeurl + ", " + str(err))
 
     return version_map

cve_bin_tool/version.py

@@ -1,10 +1,9 @@
 # Copyright (C) 2021 Intel Corporation
 # SPDX-License-Identifier: GPL-3.0-or-later
 
-import json
 import textwrap
-from urllib import request
 
+import requests
 from packaging import version
 
 from cve_bin_tool.log import LOGGER
@@ -18,19 +17,18 @@ def check_latest_version():
     name: str = "cve-bin-tool"
     url: str = f"https://pypi.org/pypi/{name}/json"
     try:
-        with request.urlopen(url) as resp:  # nosec - static url above
-            package_json = json.load(resp)
-            pypi_version = package_json["info"]["version"]
-            if pypi_version != VERSION:
+        package_json = requests.get(url).json()
+        pypi_version = package_json["info"]["version"]
+        if pypi_version != VERSION:
+            LOGGER.info(
+                f"[bold red]You are running version {VERSION} of {name} but the latest PyPI Version is {pypi_version}.[/]",
+                extra={"markup": True},
+            )
+            if version.parse(VERSION) < version.parse(pypi_version):
                 LOGGER.info(
-                    f"[bold red]You are running version {VERSION} of {name} but the latest PyPI Version is {pypi_version}.[/]",
+                    "[bold yellow]Alert: We recommend using the latest stable release.[/]",
                     extra={"markup": True},
                 )
-                if version.parse(VERSION) < version.parse(pypi_version):
-                    LOGGER.info(
-                        "[bold yellow]Alert: We recommend using the latest stable release.[/]",
-                        extra={"markup": True},
-                    )
     except Exception as error:
         LOGGER.warning(
             textwrap.dedent(

requirements.csv

@@ -18,4 +18,5 @@ srossross_not_in_db,rpmfile
 indygreg_not_in_db,zstandard
 nir0s_not_in_db,distro
 tiran_not_in_db,defusedxml
-python_not_in_db,importlib_metadata
+python_not_in_db,importlib_metadata
+python,requests

requirements.txt

@@ -17,4 +17,5 @@ zstandard; python_version >= "3.4"
 reportlab
 distro
 defusedxml
-importlib_metadata; python_version < "3.8"
+importlib_metadata; python_version < "3.8"
+requests

test/test_json.py

@@ -12,9 +12,9 @@
 import os
 import unittest
 from test.utils import LONG_TESTS
-from urllib.request import urlopen
 
 import pytest
+import requests
 from jsonschema import validate
 from jsonschema.exceptions import ValidationError
 
@@ -27,7 +27,7 @@
 
 class TestJSON:
     # Download the schema
-    SCHEMA = json.loads(urlopen(NVD_SCHEMA).read().decode("utf-8"))
+    SCHEMA = requests.get(NVD_SCHEMA).json()
     LOGGER.info("Schema loaded successfully")
 
     @unittest.skipUnless(LONG_TESTS() > 0, "Skipping long tests")

test/utils.py

@@ -7,9 +7,9 @@
 import os
 import shutil
 import tempfile
-from urllib.request import urlopen
 
 import pytest
+import requests
 
 from cve_bin_tool.async_utils import get_event_loop
 
@@ -36,9 +36,9 @@ def teardown_class(cls):
 
 def download_file(url, target):
     """helper method to download a file"""
-    download = urlopen(url)
+    download = requests.get(url)
     with open(target, "wb") as target_file:
-        target_file.write(download.read())
+        target_file.write(download.content)
     download.close()