eric: comparison eric7/PipInterface/PipVulnerabilityChecker.py

-:ca442cd49b9e
+:663521af48b2
+# -*- coding: utf-8 -*-
+# Copyright (c) 2022 Detlev Offenbach <detlev@die-offenbachs.de>
+#
+"""
+Module implementing a Python package vulnerability checker.
+The vulnerability data is provided by the open Python vulnerability database
+<a href="https://github.com/pyupio/safety-db">Safety DB</a>.
+"""
+import contextlib
+import enum
+import json
+import os
+import time
+from collections import namedtuple
+from dataclasses import dataclass
+from packaging.specifiers import SpecifierSet
+from PyQt6.QtCore import QCoreApplication, QObject, QThread, QUrl
+from PyQt6.QtNetwork import QNetworkReply, QNetworkRequest
+from EricWidgets import EricMessageBox
+import Globals
+import Preferences
+Package = namedtuple("Package", ["name", "version"])
+@dataclass
+class Vulnerability:
+"""
+Class containing the vulnerability data.
+"""
+name: str               # package name
+spec: dict              # package specification record
+version: str            # package version
+cve: str                # CVE ID
+advisory: str           # CVE advisory text
+vulnerabilityId: str    # vulnerability ID
+class VulnerabilityCheckError(enum.Enum):
+"""
+Class defining various vulnerability check error states.
+"""
+OK = 0
+SummaryDbUnavailable = 1
+FullDbUnavailable = 2
+class PipVulnerabilityChecker(QObject):
+"""
+Class implementing a Python package vulnerability checker.
+"""
+def __init__(self, pip, parent=None):
+"""
+Constructor
+@param pip reference to the global pip interface
+@type Pip
+@param parent reference to the parent widget (defaults to None)
+@type QWidget (optional)
+"""
+super().__init__(parent)
+self.__pip = pip
+securityDir = os.path.join(Globals.getConfigDir(), "security")
+os.makedirs(securityDir, mode=0o700, exist_ok=True)
+self.__cacheFile = os.path.join(securityDir,
+"vulnerability_cache.json")
+if not os.path.exists(self.__cacheFile):
+self.__createCacheFile()
+def __createCacheFile(self):
+"""
+Private method to create the cache file.
+The cache file has the following structure.
+{
+"insecure.json": {
+"cachedAt": 12345678
+"db": {}
+},
+"insecure_full.json": {
+"cachedAt": 12345678
+"db": {}
+},
+}
+"""
+structure = {
+"insecure.json": {
+"cachedAt": 0,
+"db": {},
+},
+"insecure_full.json": {
+"cachedAt": 0,
+"db": {},
+},
+}
+with open(self.__cacheFile, "w") as f:
+json.dump(structure, f, indent=2)
+def __getDataFromCache(self, dbName):
+"""
+Private method to get the vulnerability database from the cache.
+@param dbName name of the vulnerability database
+@type str
+@return dictionary containing the requested vulnerability data
+@rtype dict
+"""
+if os.path.exists(self.__cacheFile):
+with open(self.__cacheFile, "r") as f:
+with contextlib.suppress(json.JSONDecodeError, OSError):
+cachedData = json.load(f)
+if (
+dbName in cachedData and
+"cachedAt" in cachedData[dbName]
+):
+cacheValidPeriod = Preferences.getPip(
+"VulnerabilityDbCacheValidity")
+if (
+cachedData[dbName]["cachedAt"] + cacheValidPeriod >
+time.time()
+):
+return cachedData[dbName]["db"]
+return {}
+def __writeDataToCache(self, dbName, data):
+"""
+Private method to write the vulnerability data for a database to the
+cache.
+@param dbName name of the vulnerability database
+@type str
+@param data dictionary containing the vulnerability data
+@type dict
+"""
+if not os.path.exists(self.__cacheFile):
+self.__createCacheFile()
+with open(self.__cacheFile, "r") as f:
+try:
+cache = json.load(f)
+except json.JSONDecodeError:
+cache = {}
+cache[dbName] = {
+"cachedAt": time.time(),
+"db": data,
+}
+with open(self.__cacheFile, "w") as f:
+json.dump(cache, f, indent=2)
+def __fetchVulnerabilityDatabase(self, full=False):
+"""
+Private method to get the data of the vulnerability database.
+If the cached data is still valid, this data will be used.
+Otherwise a copy of the requested database will be downloaded
+and cached.
+@param full flag indicating to get the database containing the full
+data set (defaults to False)
+@type bool (optional)
+@return dictionary containing the vulnerability data (full data set or
+just package name and version specifier)
+"""
+dbName = "insecure_full.json" if full else "insecure.json"
+cachedData = self.__getDataFromCache(dbName)
+if cachedData:
+return cachedData
+url = Preferences.getPip("VulnerabilityDbMirror") + dbName
+request = QNetworkRequest(QUrl(url))
+reply = self.__pip.getNetworkAccessManager().get(request)
+while not reply.isFinished():
+QCoreApplication.processEvents()
+QThread.msleep(100)
+reply.deleteLater()
+if reply.error() == QNetworkReply.NetworkError.NoError:
+data = str(reply.readAll(),
+Preferences.getSystem("IOEncoding"),
+'replace')
+with contextlib.suppress(json.JSONDecodeError):
+data = json.loads(data)
+self.__writeDataToCache(dbName, data)
+return data
+EricMessageBox.critical(
+None,
+self.tr("Fetching Vulnerability Database"),
+self.tr("""<p>The vulnerability database <b>{0}</b> could not"""
+""" be loaded from <b>{1}</b>.</p><p>The vulnerability"""
+""" check is not available.</p>""")
+)
+return {}
+def __getVulnerabilities(self, package, specifier, db):
+"""
+Private method to get the vulnerabilities for a package.
+@param package name of the package
+@type str
+@param specifier package specifier
+@type Specifier
+@param db vulnerability data
+@type dict
+@yield dictionary containing the vulnerability data for the package
+@ytype dict
+"""
+for entry in db[package]:
+for entrySpec in entry["specs"]:
+if entrySpec == specifier:
+yield entry
+def check(self, packages):
+"""
+Public method to check the given packages for vulnerabilities.
+@param packages list of packages
+@type Package
+@return tuple containing an error status and the list of vulnerable
+packages detected
+@rtype tuple of (VulnerabilityCheckError, list of Vulnerability)
+"""
+db = self.__fetchVulnerabilityDatabase()
+if not db:
+return VulnerabilityCheckError.SummaryDbUnavailable, []
+fullDb = None
+vulnerablePackages = frozenset(db.keys())
+vulnerabilities = []            # TODO: fill this list
+for package in packages:
+# normalize the package name, the safety-db is converting
+# underscores to dashes and uses lowercase
+name = package.name.replace("_", "-").lower()
+if name in vulnerablePackages:
+# we have a candidate here, build the spec set
+for specifier in db[name]:
+specifierSet = SpecifierSet(specifiers=specifier)
+if specifierSet.contains(package.version):
+if not fullDb:
+fullDb = self.__fetchVulnerabilityDatabase(
+full=True)
+for data in self.__getVulnerabilities(
+package=name, specifier=specifier, db=fullDb
+):
+vulnarabilityId = (
+data.get("id").replace("pyup.io-", "")
+)
+cveId = data.get("cve")
+if cveId:
+cveId = cveId.split(",", 1)[0].strip()
+return VulnerabilityCheckError.OK, vulnerabilities

Mercurial Repositories > eric / file comparison

comparison: eric7/PipInterface/PipVulnerabilityChecker.py

eric7/PipInterface/PipVulnerabilityChecker.py