eric: comparison src/eric7/WebBrowser/SafeBrowsing/SafeBrowsingManager.py

-:e9e7eca7efee
+:bf71ee032bb4
 import os
 import base64
 from PyQt6.QtCore import (
-pyqtSignal, pyqtSlot, QObject, QCoreApplication, QUrl, QDateTime, QTimer
+pyqtSignal,
+pyqtSlot,
+QObject,
+QCoreApplication,
+QUrl,
+QDateTime,
+QTimer,
 )
 import Preferences
 import Utilities
 class SafeBrowsingManager(QObject):
 """
 Class implementing the interface for Google Safe Browsing.
 @signal progressMessage(message,maximum) emitted to give a message for the
 action about to be performed and the maximum value
 @signal progress(current) emitted to signal the current progress
 """
 progressMessage = pyqtSignal(str, int)
 progress = pyqtSignal(int)
-enabled = (
+enabled = Preferences.getWebBrowser("SafeBrowsingEnabled") and bool(
-Preferences.getWebBrowser("SafeBrowsingEnabled") and
+Preferences.getWebBrowser("SafeBrowsingApiKey")
-bool(Preferences.getWebBrowser("SafeBrowsingApiKey"))
 )
 def __init__(self):
 """
 Constructor
 """
 super().__init__()
 self.__apiKey = Preferences.getWebBrowser("SafeBrowsingApiKey")
 if self.__apiKey:
-self.__apiClient = SafeBrowsingAPIClient(self.__apiKey,
+self.__apiClient = SafeBrowsingAPIClient(self.__apiKey, parent=self)
-parent=self)
 else:
 self.__apiClient = None
 gsbCachePath = os.path.join(
-Utilities.getConfigDir(), "web_browser", "safe_browsing")
+Utilities.getConfigDir(), "web_browser", "safe_browsing"
+)
 self.__cache = SafeBrowsingCache(gsbCachePath, self)
 self.__gsbDialog = None
 self.__setPlatforms()
 self.__setLookupMethod()
 self.__updatingThreatLists = False
 self.__threatListsUpdateTimer = QTimer(self)
 self.__threatListsUpdateTimer.setSingleShot(True)
 self.__threatListsUpdateTimer.timeout.connect(
-self.__threatListsUpdateTimerTimeout)
+self.__threatListsUpdateTimerTimeout
+)
 self.__setAutoUpdateThreatLists()
 def configurationChanged(self):
 """
 Public method to handle changes of the settings.
 """
 apiKey = Preferences.getWebBrowser("SafeBrowsingApiKey")
 self.__apiKey = apiKey
 if self.__apiKey:
 if self.__apiClient:
 self.__apiClient.setApiKey(self.__apiKey)
 else:
-self.__apiClient = SafeBrowsingAPIClient(self.__apiKey,
+self.__apiClient = SafeBrowsingAPIClient(self.__apiKey, parent=self)
-parent=self)
+SafeBrowsingManager.enabled = Preferences.getWebBrowser(
-SafeBrowsingManager.enabled = (
+"SafeBrowsingEnabled"
-Preferences.getWebBrowser("SafeBrowsingEnabled") and
+) and bool(self.__apiKey)
-bool(self.__apiKey))
 self.__setPlatforms()
 self.__setLookupMethod()
 self.__setAutoUpdateThreatLists()
 def __setPlatforms(self):
 """
 Private method to set the platforms to be checked against.
 """
 self.__platforms = None
 platform = "macos"
 else:
 # treat all other platforms like linux
 platform = "linux"
 self.__platforms = SafeBrowsingAPIClient.getPlatformTypes(platform)
 def __setLookupMethod(self):
 """
 Private method to set the lookup method (Update API or Lookup API).
 """
-self.__useLookupApi = Preferences.getWebBrowser(
+self.__useLookupApi = Preferences.getWebBrowser("SafeBrowsingUseLookupApi")
-"SafeBrowsingUseLookupApi")
 @classmethod
 def isEnabled(cls):
 """
 Class method to check, if safe browsing is enabled.
 @return flag indicating the enabled state
 @rtype bool
 """
 return cls.enabled
 def close(self):
 """
 Public method to close the safe browsing interface.
 """
 self.__cache.close()
 def fairUseDelayExpired(self):
 """
 Public method to check, if the fair use wait period has expired.
 @return flag indicating expiration
 @rtype bool
 """
 return self.isEnabled() and self.__apiClient.fairUseDelayExpired()
 def __showNotificationMessage(self, message, timeout=5):
 """
 Private method to show some message in a notification widget.
 @param message message to be shown
 @type str
 @param timeout amount of time in seconds the message should be shown
 (0 = indefinitely)
 @type int
 """
 from WebBrowser.WebBrowserWindow import WebBrowserWindow
 kind = (
 NotificationTypes.CRITICAL
-if timeout == 0 else
+if timeout == 0
-NotificationTypes.INFORMATION
+else NotificationTypes.INFORMATION
 )
 WebBrowserWindow.showNotification(
 UI.PixmapCache.getPixmap("safeBrowsing48"),
 self.tr("Google Safe Browsing"),
 message,
 kind=kind,
 timeout=timeout,
 )
 def __setAutoUpdateThreatLists(self):
 """
 Private method to set auto update for the threat lists.
 """
-autoUpdateEnabled = (
+autoUpdateEnabled = Preferences.getWebBrowser(
-Preferences.getWebBrowser("SafeBrowsingAutoUpdate") and
+"SafeBrowsingAutoUpdate"
-not Preferences.getWebBrowser("SafeBrowsingUseLookupApi")
+) and not Preferences.getWebBrowser("SafeBrowsingUseLookupApi")
-)
 if autoUpdateEnabled and self.isEnabled():
-nextUpdateDateTime = Preferences.getWebBrowser(
+nextUpdateDateTime = Preferences.getWebBrowser("SafeBrowsingUpdateDateTime")
-"SafeBrowsingUpdateDateTime")
 if nextUpdateDateTime.isValid():
 interval = (
-QDateTime.currentDateTime().secsTo(nextUpdateDateTime) + 2
+QDateTime.currentDateTime().secsTo(nextUpdateDateTime)
++ 2
 # 2 seconds extra wait time; interval in milliseconds
 )
 if interval < 5:
 interval = 5
 # minimum 5 seconds interval
 else:
 interval = 5
 # just wait 5 seconds
 self.__threatListsUpdateTimer.start(interval * 1000)
 else:
 if self.__threatListsUpdateTimer.isActive():
 self.__threatListsUpdateTimer.stop()
 @pyqtSlot()
 def __threatListsUpdateTimerTimeout(self):
 """
 Private slot to perform the auto update of the threat lists.
 """
 ok = False
 if self.isEnabled():
-self.__showNotificationMessage(
+self.__showNotificationMessage(self.tr("Updating threat lists..."), 0)
-self.tr("Updating threat lists..."), 0)
 ok = self.updateHashPrefixCache()[0]
 if ok:
-self.__showNotificationMessage(
+self.__showNotificationMessage(self.tr("Updating threat lists done."))
-self.tr("Updating threat lists done."))
 else:
 self.__showNotificationMessage(
-self.tr("Updating threat lists failed."),
+self.tr("Updating threat lists failed."), timeout=0
-timeout=0)
+)
 if ok:
-nextUpdateDateTime = (
+nextUpdateDateTime = self.__apiClient.getFairUseDelayExpirationDateTime()
-self.__apiClient.getFairUseDelayExpirationDateTime()
+Preferences.setWebBrowser("SafeBrowsingUpdateDateTime", nextUpdateDateTime)
+self.__threatListsUpdateTimer.start(
+(QDateTime.currentDateTime().secsTo(nextUpdateDateTime) + 2) * 1000
 )
-Preferences.setWebBrowser("SafeBrowsingUpdateDateTime",
-nextUpdateDateTime)
-self.__threatListsUpdateTimer.start(
-(QDateTime.currentDateTime().secsTo(nextUpdateDateTime) + 2) *
-1000)
 # 2 seconds extra wait time; interval in milliseconds
 else:
-Preferences.setWebBrowser("SafeBrowsingUpdateDateTime",
+Preferences.setWebBrowser("SafeBrowsingUpdateDateTime", QDateTime())
-QDateTime())
 def updateHashPrefixCache(self):
 """
 Public method to load or update the locally cached threat lists.
 @return flag indicating success and an error message
 @rtype tuple of (bool, str)
 """
 if not self.isEnabled():
 return False, self.tr("Safe Browsing is disabled.")
 if not self.__apiClient.fairUseDelayExpired():
 return (
 False,
-self.tr("The fair use wait period has not expired yet."
+self.tr(
-"Expiration will be at {0}.").format(
+"The fair use wait period has not expired yet."
-self.__apiClient.getFairUseDelayExpirationDateTime()
+"Expiration will be at {0}."
-.toString("yyyy-MM-dd, HH:mm:ss"))
+).format(
+self.__apiClient.getFairUseDelayExpirationDateTime().toString(
+"yyyy-MM-dd, HH:mm:ss"
+)
+),
 )
 self.__updatingThreatLists = True
 ok = True
 errorMessage = ""
 # step 1: remove expired hashes
 self.__cache.cleanupFullHashes()
 QCoreApplication.processEvents()
 # step 2: update threat lists
 threatListsForRemove = {}
 for threatList, _clientState in self.__cache.getThreatLists():
 threatListsForRemove[repr(threatList)] = threatList
 threatLists, error = self.__apiClient.getThreatLists()
 if error:
 return False, error
 maximum = len(threatLists)
 self.progressMessage.emit(self.tr("Updating threat lists"), maximum)
 for current, entry in enumerate(threatLists, start=1):
 self.progress.emit(current)
 QCoreApplication.processEvents()
 threatList = ThreatList.fromApiEntry(entry)
-if (
+if self.__platforms is None or threatList.platformType in self.__platforms:
-self.__platforms is None or
-threatList.platformType in self.__platforms
-):
 self.__cache.addThreatList(threatList)
 key = repr(threatList)
 if key in threatListsForRemove:
 del threatListsForRemove[key]
 maximum = len(threatListsForRemove.values())
-self.progressMessage.emit(self.tr("Deleting obsolete threat lists"),
+self.progressMessage.emit(self.tr("Deleting obsolete threat lists"), maximum)
-maximum)
+for current, threatList in enumerate(threatListsForRemove.values(), start=1):
-for current, threatList in enumerate(
-threatListsForRemove.values(), start=1
-):
 self.progress.emit(current)
 QCoreApplication.processEvents()
 self.__cache.deleteHashPrefixList(threatList)
 self.__cache.deleteThreatList(threatList)
 del threatListsForRemove
 # step 3: update threats
 threatLists = self.__cache.getThreatLists()
 clientStates = {}
 for threatList, clientState in threatLists:
 clientStates[threatList.asTuple()] = clientState
-threatsUpdateResponses, error = self.__apiClient.getThreatsUpdate(
+threatsUpdateResponses, error = self.__apiClient.getThreatsUpdate(clientStates)
-clientStates)
 if error:
 return False, error
 maximum = len(threatsUpdateResponses)
 self.progressMessage.emit(self.tr("Updating hash prefixes"), maximum)
 for current, response in enumerate(threatsUpdateResponses, start=1):
 self.progress.emit(current)
 QCoreApplication.processEvents()
 responseThreatList = ThreatList.fromApiEntry(response)
 if response["responseType"] == "FULL_UPDATE":
 self.__cache.deleteHashPrefixList(responseThreatList)
 for removal in response.get("removals", []):
 self.__cache.removeHashPrefixIndices(
-responseThreatList, removal["rawIndices"]["indices"])
+responseThreatList, removal["rawIndices"]["indices"]
+)
 QCoreApplication.processEvents()
 for addition in response.get("additions", []):
 hashPrefixList = HashPrefixList(
 addition["rawHashes"]["prefixSize"],
-base64.b64decode(addition["rawHashes"]["rawHashes"]))
+base64.b64decode(addition["rawHashes"]["rawHashes"]),
-self.__cache.populateHashPrefixList(responseThreatList,
+)
-hashPrefixList)
+self.__cache.populateHashPrefixList(responseThreatList, hashPrefixList)
 QCoreApplication.processEvents()
 expectedChecksum = base64.b64decode(response["checksum"]["sha256"])
-if self.__verifyThreatListChecksum(responseThreatList,
+if self.__verifyThreatListChecksum(responseThreatList, expectedChecksum):
-expectedChecksum):
 self.__cache.updateThreatListClientState(
-responseThreatList, response["newClientState"])
+responseThreatList, response["newClientState"]
+)
 else:
 ok = False
 errorMessage = self.tr(
 "Local cache checksum does not match the server. Consider"
-" cleaning the cache. Threat update has been aborted.")
+" cleaning the cache. Threat update has been aborted."
+)
 self.__updatingThreatLists = False
 return ok, errorMessage
 def isUpdatingThreatLists(self):
 """
 Public method to check, if we are in the process of updating the
 threat lists.
 @return flag indicating an update process is active
 @rtype bool
 """
 return self.__updatingThreatLists
 def __verifyThreatListChecksum(self, threatList, remoteChecksum):
 """
 Private method to verify the local checksum of a threat list with the
 checksum of the safe browsing server.
 @param threatList threat list to calculate checksum for
 @type ThreatList
 @param remoteChecksum SHA256 checksum as reported by the Google server
 @type bytes
 @return flag indicating equality
 @rtype bool
 """
 localChecksum = self.__cache.hashPrefixListChecksum(threatList)
 return remoteChecksum == localChecksum
 def fullCacheCleanup(self):
 """
 Public method to clean up the cache completely.
 """
 self.__cache.prepareCacheDb()
 def showSafeBrowsingDialog(self):
 """
 Public slot to show the safe browsing management dialog.
 """
 if self.__gsbDialog is None:
 from WebBrowser.WebBrowserWindow import WebBrowserWindow
 from .SafeBrowsingDialog import SafeBrowsingDialog
 self.__gsbDialog = SafeBrowsingDialog(
-self, parent=WebBrowserWindow.mainWindow())
+self, parent=WebBrowserWindow.mainWindow()
+)
 self.__gsbDialog.show()
 def lookupUrl(self, url):
 """
 Public method to lookup an URL.
 @param url URL to be checked
 @type str or QUrl
 @return tuple containing the list of threat lists the URL was found in
 and an error message
 @rtype tuple of (list of ThreatList, str)
 """
 if self.isEnabled():
 if self.__useLookupApi:
 if isinstance(url, str):
 url = QUrl(url.strip())
 if url.isEmpty():
 raise ValueError("Empty URL given.")
-listNames, error = self.__apiClient.lookupUrl(
+listNames, error = self.__apiClient.lookupUrl(url, self.__platforms)
-url, self.__platforms)
 return listNames, error
 else:
 if isinstance(url, QUrl):
 urlStr = url.toString().strip()
 else:
 urlStr = url.strip()
 if not urlStr:
 raise ValueError("Empty URL given.")
 urlHashes = SafeBrowsingUrl(urlStr).hashes()
 listNames = self.__lookupHashes(urlHashes)
 return listNames, ""
 return None, ""
 def __lookupHashes(self, fullHashes):
 """
 Private method to lookup the given hashes.
 @param fullHashes list of hashes to lookup
 @type list of bytes
 @return names of threat lists hashes were found in
 @rtype list of ThreatList
 """
 fullHashes = list(fullHashes)
 cues = [fh[:4].hex() for fh in fullHashes]
 result = []
 matchingPrefixes = {}
 matchingFullHashes = set()
 isPotentialThreat = False
 # Lookup hash prefixes which match full URL hash
-for _threatList, hashPrefix, negativeCacheExpired in (
+for (
-self.__cache.lookupHashPrefix(cues)
+_threatList,
-):
+hashPrefix,
+negativeCacheExpired,
+) in self.__cache.lookupHashPrefix(cues):
 for fullHash in fullHashes:
 if fullHash.startswith(hashPrefix):
 isPotentialThreat = True
 # consider hash prefix negative cache as expired if it
 # is expired in at least one threat list
-matchingPrefixes[hashPrefix] = matchingPrefixes.get(
+matchingPrefixes[hashPrefix] = (
-hashPrefix, False) or negativeCacheExpired
+matchingPrefixes.get(hashPrefix, False) or negativeCacheExpired
+)
 matchingFullHashes.add(fullHash)
 # if none matches, url hash is clear
 if not isPotentialThreat:
 return []
 # if there is non-expired full hash, URL is blacklisted
 matchingExpiredThreatLists = set()
-for threatList, hasExpired in self.__cache.lookupFullHashes(
+for threatList, hasExpired in self.__cache.lookupFullHashes(matchingFullHashes):
-matchingFullHashes):
 if hasExpired:
 matchingExpiredThreatLists.add(threatList)
 else:
 result.append(threatList)
 if result:
 return result
 # If there are no matching expired full hash entries and negative
 # cache is still current for all prefixes, consider it safe.
 if (
-len(matchingExpiredThreatLists) == 0 and
+len(matchingExpiredThreatLists) == 0
-sum(map(int, matchingPrefixes.values())) == 0
+and sum(map(int, matchingPrefixes.values())) == 0
 ):
 return []
 # Now it can be assumed that there are expired matching full hash
 # entries and/or cache prefix entries with expired negative cache.
 # Both require full hash synchronization.
 self.__syncFullHashes(matchingPrefixes.keys())
 # Now repeat full hash lookup
-for threatList, hasExpired in self.__cache.lookupFullHashes(
+for threatList, hasExpired in self.__cache.lookupFullHashes(matchingFullHashes):
-matchingFullHashes):
 if not hasExpired:
 result.append(threatList)
 return result
 def __syncFullHashes(self, hashPrefixes):
 """
 Private method to download full hashes matching given prefixes.
 This also updates the cache expiration timestamps.
 @param hashPrefixes list of hash prefixes to get full hashes for
 @type list of bytes
 """
 threatLists = self.__cache.getThreatLists()
 clientStates = {}
 for threatList, clientState in threatLists:
 clientStates[threatList.asTuple()] = clientState
-fullHashResponses = self.__apiClient.getFullHashes(
+fullHashResponses = self.__apiClient.getFullHashes(hashPrefixes, clientStates)
-hashPrefixes, clientStates)
 # update negative cache for each hash prefix
 # store full hash with positive cache bumped up
 for match in fullHashResponses["matches"]:
 threatList = ThreatList.fromApiEntry(match)
 hashValue = base64.b64decode(match["threat"]["hash"])
 value = base64.b64decode(metadata["value"])
 if key == b"malware_threat_type":
 malwareThreatType = value
 if not isinstance(malwareThreatType, str):
 malwareThreatType = malwareThreatType.decode()
-self.__cache.storeFullHash(threatList, hashValue, cacheDuration,
+self.__cache.storeFullHash(
-malwareThreatType)
+threatList, hashValue, cacheDuration, malwareThreatType
+)
 negativeCacheDuration = int(
-fullHashResponses["negativeCacheDuration"].rstrip("s"))
+fullHashResponses["negativeCacheDuration"].rstrip("s")
+)
 for prefixValue in hashPrefixes:
 for threatList, _clientState in threatLists:
 self.__cache.updateHashPrefixExpiration(
-threatList, prefixValue, negativeCacheDuration)
+threatList, prefixValue, negativeCacheDuration
+)
 @classmethod
 def getIgnoreSchemes(cls):
 """
 Class method to get the schemes not to be checked.
 @return list of schemes to be ignored
 @rtype list of str
 """
 return [
 "about",
 "qthelp",
 "chrome",
 "abp",
 "file",
 ]
 def getThreatMessage(self, threatType):
 """
 Public method to get a warning message for the given threat type.
 @param threatType threat type to get the message for
 @type str
 @return threat message
 @rtype str
 """
-msg = (
+msg = self.__apiClient.getThreatMessage(threatType) if self.__apiClient else ""
-self.__apiClient.getThreatMessage(threatType)
-if self.__apiClient else
-""
-)
 return msg
 def getThreatMessages(self, threatLists):
 """
 Public method to get threat messages for the given threats.
 @param threatLists list of threat lists to get a message for
 @type list of ThreatList
 @return list of threat messages, one per unique threat type
 @rtype list of str
 """
 threatTypes = set()
 for threatList in threatLists:
 threatTypes.add(threatList.threatType)
 messages = []
 if self.__apiClient:
 for threatType in sorted(threatTypes):
 msg = self.__apiClient.getThreatMessage(threatType)
 messages.append(msg)
 return messages
 def getThreatType(self, threatList):
 """
 Public method to get a display string for a given threat type.
 @param threatList threat list to get display string for
 @type str
 @return display string
 @rtype str
 """
 displayString = ""
 if self.__apiClient:
-displayString = self.__apiClient.getThreatType(
+displayString = self.__apiClient.getThreatType(threatList.threatType)
-threatList.threatType)
 return displayString
 def getPlatformString(self, platformType):
 """
 Public method to get the platform string for a given platform type.
 @param platformType platform type as defined in the v4 API
 @type str
 @return platform string
 @rtype str
 """
 if self.__apiClient:
 return self.__apiClient.getPlatformString(platformType)
 else:
 return ""
 def getThreatEntryString(self, threatEntry):
 """
 Public method to get the threat entry string.
 @param threatEntry threat entry type as defined in the v4 API
 @type str
 @return threat entry string
 @rtype str
 """

Mercurial Repositories > eric / file comparison

comparison: src/eric7/WebBrowser/SafeBrowsing/SafeBrowsingManager.py

src/eric7/WebBrowser/SafeBrowsing/SafeBrowsingManager.py