WebBrowser/SafeBrowsing/SafeBrowsingAPIClient.py

# -*- coding: utf-8 -*-

# Copyright (c) 2017 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the low level interface for Google Safe Browsing.
"""

from __future__ import unicode_literals
try:
    str = unicode       # __IGNORE_EXCEPTION__
except NameError:
    pass

import json
import base64

from PyQt5.QtCore import pyqtSignal, QObject, QDateTime, QUrl, QByteArray, \
    QCoreApplication, QEventLoop
from PyQt5.QtNetwork import QNetworkRequest, QNetworkReply

from WebBrowser.WebBrowserWindow import WebBrowserWindow


class SafeBrowsingAPIClient(QObject):
    """
    Class implementing the low level interface for Google Safe Browsing.
    
    @signal networkError(str) emitted to indicate a network error
    """
    ClientId = "eric6_API_client"
    ClientVersion = "1.0.0"
    
    GsbUrlTemplate = "https://safebrowsing.googleapis.com/v4/{0}?key={1}"
    
    networkError = pyqtSignal(str)
    
    def __init__(self, apiKey, fairUse=True, parent=None):
        """
        Constructor
        
        @param apiKey API key to be used
        @type str
        @param fairUse flag indicating to follow the fair use policy
        @type bool
        @param parent reference to the parent object
        @type QObject
        """
        self.__apiKey = apiKey
        self.__fairUse = fairUse
        
        self.__nextRequestNoSoonerThan = QDateTime()
        self.__failCount = 0
    
    def setApiKey(self, apiKey):
        """
        Public method to set the API key.
        
        @param apiKey API key to be set
        @type str
        """
        self.__apiKey = apiKey
    
    def getThreatLists(self):
        """
        Public method to retrieve all available threat lists.
        
        @return list of threat lists
        @rtype list of dict containing 'threatType', 'platformType' and
            'threatEntryType'
        """
        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
        if reply.error() != QNetworkReply.NoError:
            self.networkError.emit(reply.errorString())
        else:
            result = self.__extractData(reply)
            res = result["threatLists"]
        
        reply.deleteLater()
        return res
    
    def getThreatsUpdate(self, clientStates):
        """
        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
        @rtype list of dict
        """
        requestBody = {
            "client": {
                "clientId": self.ClientId,
                "clientVersion": self.ClientVersion,
            },
            "listUpdateRequests": [],
        }
        
        for (threatType, platformType, threatEntryType), currentState in \
                clientStates.items():
            requestBody["listUpdateRequests"].append(
                {
                    "threatType": threatType,
                    "platformType": platformType,
                    "threatEntryType": threatEntryType,
                    "state": currentState,
                    "constraints": {
                        "supportedCompressions": ["RAW"],
                    }
                }
            )
        
        data = QByteArray(json.dumps(requestBody).encode("utf-8"))
        url = QUrl(self.GsbUrlTemplate.format("threatListUpdates:fetch",
                                              self.__apiKey))
        req = QNetworkRequest(url)
        req.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
        reply = WebBrowserWindow.networkManager().post(req, data)
        
        while reply.isRunning():
            QCoreApplication.processEvents(QEventLoop.AllEvents, 200)
            # max. 200 ms processing
        
        res = None
        if reply.error() != QNetworkReply.NoError:
            self.networkError.emit(reply.errorString())
        else:
            result = self.__extractData(reply)
            res = result["listUpdateResponses"]
        
        reply.deleteLater()
        return res
    
    def getFullHashes(self, prefixes, clientState):
        """
        Public method to find full hashes matching hash prefixes.
        
        @param prefixes list of hash prefixes to find
        @type list of str (Python 2) or list of bytes (Python 3)
        @param clientState dictionary of client states with keys like
            (threatType, platformType, threatEntryType)
        @type dict
        @return dictionary containing the list of found hashes and the
            negative cache duration
        @rtype dict
        """
        requestBody = {
            "client": {
                "clientId": self.ClientId,
                "clientVersion": self.ClientVersion,
            },
            "clientStates": [],
            "threatInfo": {
                "threatTypes": [],
                "platformTypes": [],
                "threatEntryTypes": [],
                "threatEntries": [],
            },
        }
        
        for prefix in prefixes:
            requestBody["threatInfo"]["threatEntries"].append(
                {"hash": base64.b64encode(prefix).decode("ascii")})
        
        for (threatType, platformType, threatEntryType), currentState in \
                clientState.items():
            requestBody["clientStates"].append(currentState)
            if threatType not in requestBody["threatInfo"]["threatTypes"]:
                requestBody["threatInfo"]["threatTypes"].append(threatType)
            if platformType not in \
                    requestBody["threatInfo"]["platformTypes"]:
                requestBody["threatInfo"]["platformTypes"].append(
                    platformType)
            if threatEntryType not in \
                    requestBody["threatInfo"]["threatEntryTypes"]:
                requestBody["threatInfo"]["threatEntryTypes"].append(
                    threatEntryType)
        
        data = QByteArray(json.dumps(requestBody).encode("utf-8"))
        url = QUrl(self.GsbUrlTemplate.format("fullHashes:find",
                                              self.__apiKey))
        req = QNetworkRequest(url)
        req.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
        reply = WebBrowserWindow.networkManager().post(req, data)
        
        while reply.isRunning():
            QCoreApplication.processEvents(QEventLoop.AllEvents, 200)
            # max. 200 ms processing
        
        res = None
        if reply.error() != QNetworkReply.NoError:
            self.networkError.emit(reply.errorString())
        else:
            res = self.__extractData(reply)
        
        reply.deleteLater()
        return res
    
    def __extractData(self, reply):
        """
        Private method to extract the data of a network reply.
        
        @param reply reference to the network reply object
        @type QNetworkReply
        @return extracted data
        @type list or dict
        """
        result = json.loads(str(reply.readAll(), "utf-8"))
        self.__setWaitDuration(result.get("minimumWaitDuration"))
        return result
    
    def __setWaitDuration(self, minimumWaitDuration):
        """
        Private method to set the minimum wait duration.
        
        @param minimumWaitDuration duration to be set
        @type str
        """
        if not self.__fairUse or minimumWaitDuration is None:
            self.__nextRequestNoSoonerThan = QDateTime()
        else:
            waitDuration = int(float(minimumWaitDuration.rstrip("s")))
            self.__nextRequestNoSoonerThan = \
                QDateTime.currentDateTime().addSecs(waitDuration)
    
    def fairUseDelayExpired(self):
        """
        Public method to check, if the fair use wait period has expired.
        
        @return flag indicating expiration
        @rtype bool
        """
        return (
            self.__fairUse and
            QDateTime.currentDateTime() >= self.__nextRequestNoSoonerThan
        ) or not self.__fairUse
    
    def getFairUseDelayExpirationDateTime(self):
        """
        Public method to get the date and time the fair use delay will expire.
        
        @return fair use delay expiration date and time
        @rtype QDateTime
        """
        return self.__nextRequestNoSoonerThan
    
    @classmethod
    def getThreatMessage(cls, threatType):
        """
        Class 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
        """
        threatType = threatType.lower()
        if threatType == "malware":
            msg = QCoreApplication.translate(
                "SafeBrowsingAPI",
                "<h3>Malware Warning</h3>"
                "<p>The web site you are about to visit may try to install"
                " harmful programs on your computer in order to steal or"
                " destroy your data.</p>")
        elif threatType == "social_engineering":
            msg = QCoreApplication.translate(
                "SafeBrowsingAPI",
                "<h3>Phishing Warning</h3>"
                "<p>The web site you are about to visit may try to trick you"
                " into doing something dangerous online, such as revealing"
                " passwords or personal information, usually through a fake"
                " website.</p>")
        elif threatType == "unwanted_software":
            msg = QCoreApplication.translate(
                "SafeBrowsingAPI",
                "<h3>Unwanted Software Warning</h3>"
                "<p>The software you are about to download may negatively"
                " affect your browsing or computing experience.</p>")
        elif threatType == "potentially_harmful_application":
            msg = QCoreApplication.translate(
                "SafeBrowsingAPI",
                "<h3>Potentially Harmful Application</h3>"
                "<p>The web site you are about to visit may try to trick you"
                " into installing applications, that may negatively affect"
                " your browsing experience.</p>")
        else:
            # unknow threat
            msg = QCoreApplication.translate(
                "SafeBrowsingAPI",
                "<h3>Unknown Threat Warning</h3>"
                "<p>The web site you are about to visit was found in the Safe"
                " Browsing Database but was not classified yet.</p>")
        
        return msg
    
    @classmethod
    def getThreatType(cls, threatType):
        """
        Class method to get a display string for a given threat type.
        
        @param threatType threat type to get display string for
        @type str
        @return display string
        @rtype str
        """
        threatType = threatType.lower()
        if threatType == "malware":
            displayString = QCoreApplication.translate(
                "SafeBrowsingAPI", "Malware")
        elif threatType == "social_engineering":
            displayString = QCoreApplication.translate(
                "SafeBrowsingAPI", "Phishing")
        elif threatType == "unwanted_software":
            displayString = QCoreApplication.translate(
                "SafeBrowsingAPI", "Unwanted Software")
        elif threatType == "potentially_harmful_application":
            displayString = QCoreApplication.translate(
                "SafeBrowsingAPI", "Harmful Application")
        elif threatType == "malcious_binary":
            displayString = QCoreApplication.translate(
                "SafeBrowsingAPI", "Malicious Binary")
        else:
            displayString = QCoreApplication.translate(
                "SafeBrowsingAPI", "Unknown Threat")
        
        return displayString
    
    @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
        @type str
        @return platform string
        @rtype str
        """
        platformStrings = {
            "WINDOWS": "Windows",
            "LINUX": "Linux",
            "ANDROID": "Android",
            "OSX": "macOS",
            "IOS": "iOS",
            "CHROME": "Chrome OS",
        }
        if platformType in platformStrings:
            return platformStrings[platformType]
        
        if platformType == "ANY_PLATFORM":
            return QCoreApplication.translate(
                "SafeBrowsingAPI", "any defined platform")
        elif platformType == "ALL_PLATFORMS":
            return QCoreApplication.translate(
                "SafeBrowsingAPI", "all defined platforms")
        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
        @type str (one of 'linux', 'windows', 'macos')
        @return list of platform types as defined in the v4 API for the
            given platform
        @rtype list of str
        @exception ValueError raised to indicate an invalid platform string
        """
        platform = platform.lower()
        
        platformTypes = ["ANY_PLATFORM", "ALL_PLATFORMS"]
        if platform == "linux":
            platformTypes.append("LINUX")
        elif platform == "windows":
            platformTypes.append("WINDOWS")
        elif platform == "macos":
            platformTypes.append("OSX")
        else:
            raise ValueError("Unsupported platform")
        
        return platformTypes