eric: comparison WebBrowser/SafeBrowsing/SafeBrowsingAPIClient.py

-:a64b986abb54
+:fb1f9e681848
 self.__lookupApiCache = {}
 # Temporary cache used by the lookup API (v4)
 # key: URL as string
 # value: dictionary with these entries:
 #   "validUntil": (QDateTime)
-#   "threatInfo": (ThreatList)
+#   "threatInfo": (list of ThreatList)
 def setApiKey(self, apiKey):
 """
 Public method to set the API key.
 def getThreatLists(self):
 """
 Public method to retrieve all available threat lists.
-@return list of threat lists
+@return tuple containing list of threat lists and an error message
-@rtype list of dict containing 'threatType', 'platformType' and
+@rtype tuple of (list of dict containing 'threatType', 'platformType'
-'threatEntryType'
+and 'threatEntryType', bool)
 """
 url = QUrl(self.GsbUrlTemplate.format("threatLists", self.__apiKey))
 req = QNetworkRequest(url)
 reply = WebBrowserWindow.networkManager().get(req)
 while reply.isRunning():
 QCoreApplication.processEvents(QEventLoop.AllEvents, 200)
 # max. 200 ms processing
 res = None
+error = ""
 if reply.error() != QNetworkReply.NoError:
-self.networkError.emit(reply.errorString())
+error = reply.errorString()
+self.networkError.emit(error)
 else:
 result = self.__extractData(reply)
 res = result["threatLists"]
 reply.deleteLater()
-return res
+return res, error
 #######################################################################
 ## Methods below implement the 'Update API (v4)'
 #######################################################################
 Public method to fetch hash prefix updates for the given threat list.
 @param clientStates dictionary of client states with keys like
 (threatType, platformType, threatEntryType)
 @type dict
-@return list of threat updates
+@return tuple containing the list of threat updates and an error
-@rtype list of dict
+message
+@rtype tuple of (list of dict, bool)
 """
 requestBody = {
 "client": {
 "clientId": self.ClientId,
 "clientVersion": self.ClientVersion,
 while reply.isRunning():
 QCoreApplication.processEvents(QEventLoop.AllEvents, 200)
 # max. 200 ms processing
 res = None
+error = ""
 if reply.error() != QNetworkReply.NoError:
-self.networkError.emit(reply.errorString())
+error = reply.errorString()
+self.networkError.emit(error)
 else:
 result = self.__extractData(reply)
 res = result["listUpdateResponses"]
 reply.deleteLater()
-return res
+return res, error
 def getFullHashes(self, prefixes, clientState):
 """
 Public method to find full hashes matching hash prefixes.
 while reply.isRunning():
 QCoreApplication.processEvents(QEventLoop.AllEvents, 200)
 # max. 200 ms processing
-res = None
+res = []
 if reply.error() != QNetworkReply.NoError:
 self.networkError.emit(reply.errorString())
 else:
 res = self.__extractData(reply)
 @param url URL to be checked
 @type QUrl
 @param platforms list of platform types to check against
 @type list of str
-@return list of threat list info objects
+@return tuple containing the list of threat list info objects and
-@rtype list of ThreatList
+an error message
-"""
+@rtype tuple of (list of ThreatList, str)
+"""
+error = ""
 # sanitize the URL by removing user info and query data
 url = url.adjusted(
 QUrl.RemoveUserInfo | QUrl.RemoveQuery | QUrl.RemoveFragment
 )
 urlStr = url.toString()
+# check the local cache first
 if urlStr in self.__lookupApiCache:
 if self.__lookupApiCache[urlStr]["validUntil"] > \
 QDateTime.currentDateTime():
 # cached entry is still valid
-return self.__lookupApiCache[urlStr]["threatInfo"]
+return self.__lookupApiCache[urlStr]["threatInfo"], error
 else:
 del self.__lookupApiCache[urlStr]
+# no valid entry found, ask the safe browsing server
 requestBody = {
 "client": {
 "clientId": self.ClientId,
 "clientVersion": self.ClientVersion,
 },
 "threatInfo": {
-"threatTypes": [
+"threatTypes": SafeBrowsingAPIClient.definedThreatTypes(),
-"MALWARE", "SOCIAL_ENGINEERING", "UNWANTED_SOFTWARE",
-"POTENTIALLY_HARMFUL_APPLICATION",
-],
 "platformTypes": platforms,
-"threatEntryTypes": ["URL", "EXECUTABLE"],
+"threatEntryTypes":
+SafeBrowsingAPIClient.definedThreatEntryTypes(),
 "threatEntries": [
 {"url": urlStr},
 ],
 },
 }
 QCoreApplication.processEvents(QEventLoop.AllEvents, 200)
 # max. 200 ms processing
 threats = []
 if reply.error() != QNetworkReply.NoError:
-self.networkError.emit(reply.errorString())
+error = reply.errorString()
+self.networkError.emit(error)
 else:
 res = json.loads(str(reply.readAll(), "utf-8"))
 if res and "matches" in res:
 cacheDuration = 0
 for match in res["matches"]:
 "validUntil": validUntil,
 "threatInfo": threats
 }
 reply.deleteLater()
-return threats
+return threats, error
 #######################################################################
 ## Methods below implement global (class wide) functionality
 #######################################################################
 "SafeBrowsingAPI", "Unknown Threat")
 return displayString
 @classmethod
+def definedThreatTypes(cls):
+"""
+Class method to get all threat types defined in API v4.
+@return list of defined threat types
+@rtype list of str
+"""
+return [
+"THREAT_TYPE_UNSPECIFIED", "MALWARE", "SOCIAL_ENGINEERING",
+"UNWANTED_SOFTWARE", "POTENTIALLY_HARMFUL_APPLICATION",
+]
+@classmethod
+def getThreatEntryString(cls, threatEntry):
+"""
+Class 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
+"""
+if threatEntry == "URL":
+return "URL"
+elif threatEntry == "EXECUTABLE":
+return QCoreApplication.translate(
+"SafeBrowsingAPI", "executable program")
+else:
+return QCoreApplication.translate(
+"SafeBrowsingAPI", "unknown type")
+@classmethod
+def definedThreatEntryTypes(cls):
+"""
+Class method to get all threat entry types defined in API v4.
+@return list of all defined threat entry types
+@rtype list of str
+"""
+return [
+"THREAT_ENTRY_TYPE_UNSPECIFIED", "URL", "EXECUTABLE",
+]
+@classmethod
 def getPlatformString(cls, platformType):
 """
 Class method to get the platform string for a given platform type.
 @param platformType platform type as defined in the v4 API
 else:
 return QCoreApplication.translate(
 "SafeBrowsingAPI", "unknown platform")
 @classmethod
-def getThreatEntryString(cls, threatEntry):
-"""
-Class 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
-"""
-if threatEntry == "URL":
-return "URL"
-elif threatEntry == "EXECUTABLE":
-return QCoreApplication.translate(
-"SafeBrowsingAPI", "executable program")
-else:
-return QCoreApplication.translate(
-"SafeBrowsingAPI", "unknown type")
-@classmethod
 def getPlatformTypes(cls, platform):
 """
 Class method to get the platform types for a given platform.
 @param platform platform string
 platformTypes.append("OSX")
 else:
 raise ValueError("Unsupported platform")
 return platformTypes
+@classmethod
+def definedPlatformTypes(cls):
+"""
+Class method to get all platform types defined in API v4.
+@return list of all defined platform types
+@rtype list of str
+"""
+return [
+"PLATFORM_TYPE_UNSPECIFIED", "WINDOWS", "LINUX", "ANDROID", "OSX",
+"IOS", "ANY_PLATFORM", "ALL_PLATFORMS", "CHROME",
+]

Mercurial Repositories > eric / file comparison

comparison: WebBrowser/SafeBrowsing/SafeBrowsingAPIClient.py

WebBrowser/SafeBrowsing/SafeBrowsingAPIClient.py