src/eric7/WebBrowser/SafeBrowsing/SafeBrowsingManager.py

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

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

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

#
# Some part of this code were ported from gglsbl.client and adapted
# to Qt.
#
# https://github.com/afilipovich/gglsbl
#

import os
import base64

from PyQt6.QtCore import (
    pyqtSignal, pyqtSlot, QObject, QCoreApplication, QUrl, QDateTime, QTimer
)

import Preferences
import Utilities

import UI.PixmapCache
from UI.NotificationWidget import NotificationTypes

from .SafeBrowsingAPIClient import SafeBrowsingAPIClient
from .SafeBrowsingCache import SafeBrowsingCache
from .SafeBrowsingThreatList import ThreatList, HashPrefixList
from .SafeBrowsingUrl import SafeBrowsingUrl


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 = (
        Preferences.getWebBrowser("SafeBrowsingEnabled") and
        bool(Preferences.getWebBrowser("SafeBrowsingApiKey"))
    )
    
    def __init__(self):
        """
        Constructor
        """
        super().__init__()
        
        self.__apiKey = Preferences.getWebBrowser("SafeBrowsingApiKey")
        if self.__apiKey:
            self.__apiClient = SafeBrowsingAPIClient(self.__apiKey,
                                                     parent=self)
        else:
            self.__apiClient = None
        
        gsbCachePath = os.path.join(
            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.__setAutoUpdateThreatLists()
    
    def configurationChanged(self):
        """
        Public method to handle changes of the settings.
        """
        apiKey = Preferences.getWebBrowser("SafeBrowsingApiKey")
        if apiKey != self.__apiKey:
            self.__apiKey = apiKey
            if self.__apiKey:
                if self.__apiClient:
                    self.__apiClient.setApiKey(self.__apiKey)
                else:
                    self.__apiClient = SafeBrowsingAPIClient(self.__apiKey,
                                                             parent=self)
        
        SafeBrowsingManager.enabled = (
            Preferences.getWebBrowser("SafeBrowsingEnabled") and
            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
        if Preferences.getWebBrowser("SafeBrowsingFilterPlatform"):
            if Utilities.isWindowsPlatform():
                platform = "windows"
            elif Utilities.isMacPlatform():
                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(
            "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
            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 = (
            Preferences.getWebBrowser("SafeBrowsingAutoUpdate") and
            not Preferences.getWebBrowser("SafeBrowsingUseLookupApi")
        )
        if autoUpdateEnabled and self.isEnabled():
            nextUpdateDateTime = Preferences.getWebBrowser(
                "SafeBrowsingUpdateDateTime")
            if nextUpdateDateTime.isValid():
                interval = (
                    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.tr("Updating threat lists..."), 0)
            ok = self.updateHashPrefixCache()[0]
            if ok:
                self.__showNotificationMessage(
                    self.tr("Updating threat lists done."))
            else:
                self.__showNotificationMessage(
                    self.tr("Updating threat lists failed."),
                    timeout=0)
        
        if ok:
            nextUpdateDateTime = (
                self.__apiClient.getFairUseDelayExpirationDateTime()
            )
            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",
                                      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."
                        "Expiration will be at {0}.").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 (
                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"),
                                  maximum)
        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(
            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"])
                QCoreApplication.processEvents()
            for addition in response.get("additions", []):
                hashPrefixList = HashPrefixList(
                    addition["rawHashes"]["prefixSize"],
                    base64.b64decode(addition["rawHashes"]["rawHashes"]))
                self.__cache.populateHashPrefixList(responseThreatList,
                                                    hashPrefixList)
                QCoreApplication.processEvents()
            expectedChecksum = base64.b64decode(response["checksum"]["sha256"])
            if self.__verifyThreatListChecksum(responseThreatList,
                                               expectedChecksum):
                self.__cache.updateThreatListClientState(
                    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.")
        
        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.__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)
        @exception ValueError raised for an invalid URL
        """
        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(
                    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 (
            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(
                        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(
                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
            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(
                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(
            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"])
            cacheDuration = int(match["cacheDuration"].rstrip("s"))
            malwareThreatType = None
            for metadata in match["threatEntryMetadata"].get("entries", []):
                key = base64.b64decode(metadata["key"])
                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,
                                       malwareThreatType)
        
        negativeCacheDuration = int(
            fullHashResponses["negativeCacheDuration"].rstrip("s"))
        for prefixValue in hashPrefixes:
            for threatList, _clientState in threatLists:
                self.__cache.updateHashPrefixExpiration(
                    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",
            "eric",
            "qrc",
            "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 = (
            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(
                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
        """
        if self.__apiClient:
            return self.__apiClient.getThreatEntryString(threatEntry)
        else:
            return ""