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.handler import HandlerProto, NetworkTypes
from searxqt.core.instanceVersions import parseVersionString
from searxqt.core.requests import JsonResult, ErrorType
from searxqt.core import jsonVerify


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 (
            "IP: {0} Reverse: {1} FieldType: {2} asnCidr: {3}"
            .format(self.ip, self.reverse, self.fieldType, 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

    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 parseVersionString(self._data.get('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(JsonResult):
    v_str         = jsonVerify.Value(str)
    v_int         = jsonVerify.Value(int)
    v_float       = jsonVerify.Value(float)
    v_bool        = jsonVerify.Value(bool)
    v_noneStr     = jsonVerify.MultiValue((jsonVerify.NoneType, str))
    v_noneInt     = jsonVerify.MultiValue((jsonVerify.NoneType, int))
    v_noneFloat   = jsonVerify.MultiValue((jsonVerify.NoneType, float))
    v_noneBoolStr = jsonVerify.MultiValue((jsonVerify.NoneType, bool, str))

    TimingStructure = {
        "success_percentage": v_float,
        "all": {
            "median": v_float,
            "stdev": v_float,
            "value": v_float,
            "mean": v_float
        },
        "server": {
            "median": v_float,
            "stdev": v_float,
            "value": v_float,
            "mean": v_float
        },
        "load": {
            "median": v_float,
            "stdev": v_float,
            "value": v_float,
            "mean": v_float,
            "mean": v_float
        },
        "error": v_str
    }

    ExpectedStructure = {
        "metadata": {
            "timestamp": v_int,
            "ips": {
                "": {
                    "reverse": v_noneStr,
                    "field_type": v_str,
                    "asn_cidr": v_str
                }
            },
            "ipv6": v_bool
        },
        "instances": {
            "": {
                "analytics": v_bool,
                "comments": [v_str],
                "alternativeUrls": {},
                "docs_url": v_noneStr,
                "contact_url": v_noneBoolStr,
                "main": v_bool,
                "network_type": v_noneStr,
                "http": {
                    "status_code": v_noneInt,
                    "error": v_noneStr,
                    "grade": v_noneStr,
                    "gradeUrl": v_noneStr
                },
                "version": v_noneStr,
                "git_url": v_noneStr,
                "error": v_noneStr,
                "timing": {
                    "initial": TimingStructure,
                    "search": TimingStructure,
                    "search_wp": TimingStructure,
                    "search_go": TimingStructure
                },
                "tls": {
                    "version": v_str,
                    "certificate": {
                        "issuer": {
                            "commonName": v_str,
                            "countryName": v_str,
                            "organizationName": v_str
                        },
                        "subject": {
                            "commonName": v_noneStr,
                            "countryName": v_noneStr,
                            "organizationName": v_noneStr,
                            "altName": v_str
                        },
                        "serialNumber": v_str,
                        "notBefore": v_str,
                        "notAfter": v_str,
                        "OCSP": [v_str],
                        "caIssuers": [v_str],
                        "sha256": v_str,
                        "signatureAlgorithm": v_str,
                        "crlDistributionPoints": [v_str],
                        "version": v_int
                    },
                    "grade": v_str,
                    "gradeUrl": v_str
                },
                "html": {},
                "network": {
                    "dnssec": v_int,
                    "asn_privacy": v_int,
                    "ips": {
                        "": {
                            "https_port": v_bool,
                            "field_type": v_str,
                            "reverse": v_noneStr,
                            "asn_cidr": v_noneStr,
                            "whois_error": v_str,
                            "https_port_error": v_str
                        }
                    },
                    "ipv6": v_bool
                },
                "engines": {
                    "": {
                        "error_rate": v_noneInt,
                        "errors": [v_int],
                        "checker": {},
                        "categories": [v_str]
                    }
                }
            }
        },
        "engines": {
            "": {
                "categories": [v_str],
                "language_support": v_bool,
                "paging": v_bool,
                "safesearch": v_bool,
                "time_range_support": v_bool,
                "shortcut": v_str,
                "stats": {
                    "instance_count": v_int,
                    "stats_count": v_int,
                    "error_rate": v_noneFloat
                }
            }
        },
        "engine_errors": [v_str],
        "categories": [v_str],
        "hashes": [{
            "count": v_int,
            "hash": v_str,
            "unknown": v_bool,
            "forks": [v_int]
        }],
        "cidrs": {
            "": {
                "asn": v_noneStr,
                "asn_country_code": v_noneStr,
                "network_country": v_noneStr,
                "asn_privacy": v_int,
                "asn_description": v_noneStr
            }
        },
        "forks": [v_str]
    }

    def __init__(self, url, response, err="", errType=ErrorType.Success):
        JsonResult.__init__(self, url, response, err=err, errType=errType)


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, requestsHandler):
        HandlerProto.__init__(self, requestsHandler)

    def updateInstances(self):
        """Fetches instances.json from a searx-stats2 instance.

        @param requestKwargs: Kwargs to pass to requests.get
        @type requestKwargs: dict

        @return: A tuple (bool Success/Failed, str Message)
        @rtype: tuple
        """
        url = Stats2.URL.rstrip('/') + '/data/instances.json'
        result = self.requestsHandler.get(url, ResultType=Stats2Result)
        if result:
            self.setData(result.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)})

        return result