searx-qt/searxqt/core/instances.py

########################################################################
#  Searx-Qt - Lightweight desktop application for Searx.
#  Copyright (C) 2020-2022  CYBERDEViL
#
#  This file is part of Searx-Qt.
#
#  Searx-Qt is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#
#  Searx-Qt is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
#
########################################################################

import time

from searxqt.core import log
from searxqt.core.handler import HandlerProto, NetworkTypes
from searxqt.core.instanceVersions import InstanceVersion
from searxqt.core.http import HttpRequest, HttpJsonReponse, ErrorType
from searxqt.core.schema import Schemas


class SearchEngine:
    def __init__(self, name, data):
        """Model for a search engine data.

        @param name: Name of the search engine
        @type name: str

        @param data: Data of the search engine
        @type data: dict
        """
        self._name = name
        self._data = data

    def __repr__(self): return self._name

    @property
    def name(self):
        """
        @return: returns the name of this search engine.
        @rtype: str
        """
        return self._name


class TLS:
    def __init__(self, data):
        """Model for a instance it's TLS data.

        @param data: dict with the instance it's TLS data.
        @type data: dict
        """
        self._data = data

    @property
    def data(self): return self._data

    @property
    def version(self):
        """Returns the TLS version used.

        @return:
        @rtype: str
        """
        return self.data.get('version', "")

    @property
    def certificate(self):
        """Returns a TLSCertificate object.

        @return:
        @rtype: TLSCertificate
        """
        return TLSCertificate(self.data.get('certificate', {}))


class TLSCertificate:
    def __init__(self, data):
        """Model for a instance it's TLS certificate-data.

        @param data: dict with the instance it's TLS certificate-data.
        @type data: dict
        """
        self._data = data

    @property
    def data(self): return self._data

    @property
    def version(self):
        """
        @return:
        @rtype: int
        """
        return self.data.get('version', 0)

    @property
    def issuer(self):
        """
        @return:
        @rtype: TLSCertificateIssuer
        """
        return TLSCertificateIssuer(self.data.get('issuer', {}))


class TLSCertificateIssuer:
    def __init__(self, data):
        """Model for a instance it's TLS certificate-issuer-data.

        @param data: dict with the instance it's
                    TLS certificate-issuer-data.
        @type data: dict
        """
        self._data = data

    @property
    def data(self): return self._data

    @property
    def commonName(self):
        """
        @rtype: str
        """
        return self.data.get('commonName', "")

    @property
    def countryName(self):
        """
        @rtype: str
        """
        return self.data.get('countryName', "")

    @property
    def organizationName(self):
        """
        @rtype: str
        """
        return self.data.get('organizationName', "")


class Network:
    def __init__(self, data):
        """Model for a instance it's network data.

        @param data: dict with the instance it's network data.
        @type data: dict
        """
        self._data = data

    @property
    def data(self): return self._data

    @property
    def ipv6(self):
        """
        @return: If ipv6 is enabled on this instance.
        @rtype: bool
        """
        return self.data.get('ipv6', False)

    @property
    def ips(self):
        """
        @return: A list with NetworkIP objects
        @rtype: list
        """
        return [NetworkIP(ip, data)
                for ip, data in self.data.get('ips', {}).items()]

    @property
    def asnPrivacy(self):
        """
        @return: -1 no asn privacy, 0 asn privacy, -2 unknown.
        @rtype: int
        """
        return self.data.get('asn_privacy', -2)


class NetworkIP:
    def __init__(self, ip, data):
        """Model for a network ip data.

        @param ip: ip address
        @type ip: str

        @param data: dict with a network ip data.
        @type data: dict
        """
        self._ip = ip
        self._data = data

    def __repr__(self):
        return (
            f"IP: {self.ip} Reverse: {self.reverse} " \
            f"FieldType: {self.fieldType} asnCidr: {self.asnCidr}"
        )

    def __str__(self): return repr(self)

    @property
    def ip(self):
        """
        @return: ip address
        @rtype: str
        """
        return self._ip

    @property
    def data(self): return self._data

    @property
    def reverse(self):
        """
        @return:
        @rtype: str
        """
        return self.data.get('reverse', None)

    @property
    def fieldType(self):
        """
        @return: Record type
        @rtype: str
        """
        return self.data.get('field_type', None)

    @property
    def asnCidr(self):
        """
        @return:
        @rtype: str
        """
        return self.data.get('asn_cidr', None)

    @property
    def httpsPort(self):
        """
        @return:
        @rtype: bool
        """
        return self.data.get('https_port', None)


class Instance:
    def __init__(self, url, data):
        """Model for a SearX instance.

        @param url: Url of the instance
        @type url: str

        @param data: Data of the instance
        @type data: dict
        """
        self._url = url
        self._data = data
        self._version = InstanceVersion(self._data.get('version', ''))

        if not self._version.isValid():
            log.debug(f"Invalid version {self._version}: {self._version.error()}", cls=Instance)

    def __str__(self): return self.url

    def __repr__(self): return str(self)

    @property
    def data(self): return self._data

    @property
    def url(self):
        """
        @return: returns the url of this instance.
        @rtype: str
        """
        return self._url

    @property
    def main(self):
        """Returns False when not set.

        @return: ?
        @rtype: bool
        """
        return self._data.get('main', False)

    @property
    def networkType(self):
        """ Network type; see core/handler.py:NetworkTypes

        @return: Network protocol used (Web, Tor, I2P from NetworkTypes)
        @rtype: int
        """
        return self._data.get('network_type', 0)

    @property
    def version(self):
        """Returns a empty string when none found.

        @return: Returns the instance it's version.
        @rtype: InstanceVersion
        """
        return self._version

    @property
    def engines(self):
        """
        Returns a empty string when none found.

        @return: Returns a list with SearchEngine objects
        @rtype: list
        """
        return [SearchEngine(name, data) for name, data in self._data.get(
            'engines', {}).items()]

    @property
    def tls(self):
        """
        @rtype: TLS
        """
        return TLS(self._data.get('tls', {}))

    @property
    def network(self):
        """
        @rtype: Network
        """
        return Network(self._data.get('network', {}))


class Stats2Result(HttpJsonReponse):
    Schema = Schemas['searx_space_instances']


class Stats2(HandlerProto):
    """ This class holds the instances.json data and will be passed
    to other classes (Instances, Engines)
    """
    URL = "https://searx.space/"

    def __init__(self, httpThread, httpSettings):
        HandlerProto.__init__(self, httpThread, httpSettings)

    def updateFinishedCb(self, response):
        # FIXME, for now its working because in searx-qt we reimplement it.
        pass

    def _updateFinished(self, response):
        if response.error != ErrorType.Success:
            self.updateFinishedCb(response)
            return

        self.setData(response.json())
        self._lastUpdated = time.time()

        # Processing (use our own definition of network types
        for url, data in self.instances.items():
            data.update({"network_type": NetworkTypes.netTypeFromUrl(url)})

        self.updateFinishedCb(response)

    def updateInstances(self):
        """Fetches instances.json from a searx-stats2 instance.
        """
        url = Stats2.URL.rstrip('/') + '/data/instances.json'

        request = HttpRequest(url,
                              self.httpSettings.newRequestSettings())
        response = Stats2Result(request, self._updateFinished)
        self._httpThread.get(response)