ZeroNet/plugins/Benchmark/BenchmarkPlugin.py

import os
import time
import io
import math
import hashlib
import re
import sys

from Config import config
from Crypt import CryptHash
from Plugin import PluginManager
from Debug import Debug
from util import helper

plugin_dir = os.path.dirname(__file__)

benchmark_key = None


@PluginManager.registerTo("UiRequest")
class UiRequestPlugin(object):
    @helper.encodeResponse
    def actionBenchmark(self):
        global benchmark_key
        script_nonce = self.getScriptNonce()
        if not benchmark_key:
            benchmark_key = CryptHash.random(encoding="base64")
        self.sendHeader(script_nonce=script_nonce)

        if "Multiuser" in PluginManager.plugin_manager.plugin_names and not config.multiuser_local:
            yield "This function is disabled on this proxy"
            return

        data = self.render(
            plugin_dir + "/media/benchmark.html",
            script_nonce=script_nonce,
            benchmark_key=benchmark_key,
            filter=re.sub("[^A-Za-z0-9]", "", self.get.get("filter", ""))
        )
        yield data

    @helper.encodeResponse
    def actionBenchmarkResult(self):
        global benchmark_key
        if self.get.get("benchmark_key", "") != benchmark_key:
            return self.error403("Invalid benchmark key")

        self.sendHeader(content_type="text/plain", noscript=True)

        if "Multiuser" in PluginManager.plugin_manager.plugin_names and not config.multiuser_local:
            yield "This function is disabled on this proxy"
            return

        yield " " * 1024  # Head (required for streaming)

        import main
        s = time.time()

        for part in main.actions.testBenchmark(filter=self.get.get("filter", "")):
            yield part

        yield "\n - Total time: %.3fs" % (time.time() - s)


@PluginManager.registerTo("Actions")
class ActionsPlugin:
    def getMultiplerTitle(self, multipler):
        if multipler < 0.3:
            multipler_title = "Sloooow"
        elif multipler < 0.6:
            multipler_title = "Ehh"
        elif multipler < 0.8:
            multipler_title = "Goodish"
        elif multipler < 1.2:
            multipler_title = "OK"
        elif multipler < 1.7:
            multipler_title = "Fine"
        elif multipler < 2.5:
            multipler_title = "Fast"
        elif multipler < 3.5:
            multipler_title = "WOW"
        else:
            multipler_title = "Insane!!"
        return multipler_title

    def formatResult(self, taken, standard):
        if not standard:
            return " Done in %.3fs" % taken

        if taken > 0:
            multipler = standard / taken
        else:
            multipler = 99
        multipler_title = self.getMultiplerTitle(multipler)

        return " Done in %.3fs = %s (%.2fx)" % (taken, multipler_title, multipler)

    def getBenchmarkTests(self, online=False):
        if hasattr(super(), "getBenchmarkTests"):
            tests = super().getBenchmarkTests(online)
        else:
            tests = []

        tests.extend([
            {"func": self.testHdPrivatekey, "num": 50, "time_standard": 0.57},
            {"func": self.testSign, "num": 20, "time_standard": 0.46},
            {"func": self.testVerify, "kwargs": {"lib_verify": "sslcrypto_fallback"}, "num": 20, "time_standard": 0.38},
            {"func": self.testVerify, "kwargs": {"lib_verify": "sslcrypto"}, "num": 200, "time_standard": 0.30},
            {"func": self.testVerify, "kwargs": {"lib_verify": "libsecp256k1"}, "num": 200, "time_standard": 0.10},

            {"func": self.testPackMsgpack, "num": 100, "time_standard": 0.35},
            {"func": self.testUnpackMsgpackStreaming, "kwargs": {"fallback": False}, "num": 100, "time_standard": 0.35},
            {"func": self.testUnpackMsgpackStreaming, "kwargs": {"fallback": True}, "num": 10, "time_standard": 0.5},

            {"func": self.testPackZip, "num": 5, "time_standard": 0.065},
            {"func": self.testPackArchive, "kwargs": {"archive_type": "gz"}, "num": 5, "time_standard": 0.08},
            {"func": self.testPackArchive, "kwargs": {"archive_type": "bz2"}, "num": 5, "time_standard": 0.68},
            {"func": self.testPackArchive, "kwargs": {"archive_type": "xz"}, "num": 5, "time_standard": 0.47},
            {"func": self.testUnpackZip, "num": 20, "time_standard": 0.25},
            {"func": self.testUnpackArchive, "kwargs": {"archive_type": "gz"}, "num": 20, "time_standard": 0.28},
            {"func": self.testUnpackArchive, "kwargs": {"archive_type": "bz2"}, "num": 20, "time_standard": 0.83},
            {"func": self.testUnpackArchive, "kwargs": {"archive_type": "xz"}, "num": 20, "time_standard": 0.38},

            {"func": self.testCryptHash, "kwargs": {"hash_type": "sha256"}, "num": 10, "time_standard": 0.50},
            {"func": self.testCryptHash, "kwargs": {"hash_type": "sha512"}, "num": 10, "time_standard": 0.33},
            {"func": self.testCryptHashlib, "kwargs": {"hash_type": "sha3_256"}, "num": 10, "time_standard": 0.33},
            {"func": self.testCryptHashlib, "kwargs": {"hash_type": "sha3_512"}, "num": 10, "time_standard": 0.65},

            {"func": self.testRandom, "num": 100, "time_standard": 0.08},
        ])

        if online:
            tests += [
                {"func": self.testHttps, "num": 1, "time_standard": 2.1}
            ]
        return tests

    def testBenchmark(self, num_multipler=1, online=False, num_run=None, filter=None):
        """
        Run benchmark on client functions
        """
        tests = self.getBenchmarkTests(online=online)

        if filter:
            tests = [test for test in tests[:] if filter.lower() in test["func"].__name__.lower()]

        yield "\n"
        res = {}
        res_time_taken = {}
        multiplers = []
        for test in tests:
            s = time.time()
            if num_run:
                num_run_test = num_run
            else:
                num_run_test = math.ceil(test["num"] * num_multipler)
            func = test["func"]
            func_name = func.__name__
            kwargs = test.get("kwargs", {})
            key = "%s %s" % (func_name, kwargs)
            if kwargs:
                yield "* Running %s (%s) x %s " % (func_name, kwargs, num_run_test)
            else:
                yield "* Running %s x %s " % (func_name, num_run_test)
            i = 0
            try:
                for progress in func(num_run_test, **kwargs):
                    i += 1
                    if num_run_test > 10:
                        should_print = i % (num_run_test / 10) == 0 or progress != "."
                    else:
                        should_print = True

                    if should_print:
                        if num_run_test == 1 and progress == ".":
                            progress = "..."
                        yield progress
                time_taken = time.time() - s
                if num_run:
                    time_standard = 0
                else:
                    time_standard = test["time_standard"] * num_multipler
                yield self.formatResult(time_taken, time_standard)
                yield "\n"
                res[key] = "ok"
                res_time_taken[key] = time_taken
                multiplers.append(time_standard / max(time_taken, 0.001))
            except Exception as err:
                res[key] = err
                yield "Failed!\n! Error: %s\n\n" % Debug.formatException(err)

        yield "\n== Result ==\n"

        # Check verification speed
        if "testVerify {'lib_verify': 'sslcrypto'}" in res_time_taken:
            speed_order = ["sslcrypto_fallback", "sslcrypto", "libsecp256k1"]
            time_taken = {}
            for lib_verify in speed_order:
                time_taken[lib_verify] = res_time_taken["testVerify {'lib_verify': '%s'}" % lib_verify]

            time_taken["sslcrypto_fallback"] *= 10  # fallback benchmark only run 20 times instead of 200
            speedup_sslcrypto = time_taken["sslcrypto_fallback"] / time_taken["sslcrypto"]
            speedup_libsecp256k1 = time_taken["sslcrypto_fallback"] / time_taken["libsecp256k1"]

            yield "\n* Verification speedup:\n"
            yield " - OpenSSL: %.1fx (reference: 7.0x)\n" % speedup_sslcrypto
            yield " - libsecp256k1: %.1fx (reference: 23.8x)\n" % speedup_libsecp256k1

            if speedup_sslcrypto < 2:
                res["Verification speed"] = "error: OpenSSL speedup low: %.1fx" % speedup_sslcrypto

            if speedup_libsecp256k1 < speedup_sslcrypto:
                res["Verification speed"] = "error: libsecp256k1 speedup low: %.1fx" % speedup_libsecp256k1

        if not res:
            yield "! No tests found"
            if config.action == "test":
                sys.exit(1)
        else:
            num_failed = len([res_key for res_key, res_val in res.items() if res_val != "ok"])
            num_success = len([res_key for res_key, res_val in res.items() if res_val == "ok"])
            yield "\n* Tests:\n"
            yield " - Total: %s tests\n" % len(res)
            yield " - Success: %s tests\n" % num_success
            yield " - Failed: %s tests\n" % num_failed
            if any(multiplers):
                multipler_avg = sum(multiplers) / len(multiplers)
                multipler_title = self.getMultiplerTitle(multipler_avg)
                yield " - Average speed factor: %.2fx (%s)\n" % (multipler_avg, multipler_title)

            # Display errors
            for res_key, res_val in res.items():
                if res_val != "ok":
                    yield " ! %s %s\n" % (res_key, res_val)

            if num_failed != 0 and config.action == "test":
                sys.exit(1)

    def testHttps(self, num_run=1):
        """
        Test https connection with valid and invalid certs
        """
        import urllib.request
        import urllib.error

        body = urllib.request.urlopen("https://google.com").read()
        assert len(body) > 100
        yield "."

        badssl_urls = [
            "https://expired.badssl.com/",
            "https://wrong.host.badssl.com/",
            "https://self-signed.badssl.com/",
            "https://untrusted-root.badssl.com/"
        ]
        for badssl_url in badssl_urls:
            try:
                body = urllib.request.urlopen(badssl_url).read()
                https_err = None
            except urllib.error.URLError as err:
                https_err = err
            assert https_err
            yield "."

    def testCryptHash(self, num_run=1, hash_type="sha256"):
        """
        Test hashing functions
        """
        yield "(5MB) "

        from Crypt import CryptHash

        hash_types = {
            "sha256": {"func": CryptHash.sha256sum, "hash_valid": "8cd629d9d6aff6590da8b80782a5046d2673d5917b99d5603c3dcb4005c45ffa"},
            "sha512": {"func": CryptHash.sha512sum, "hash_valid": "9ca7e855d430964d5b55b114e95c6bbb114a6d478f6485df93044d87b108904d"}
        }
        hash_func = hash_types[hash_type]["func"]
        hash_valid = hash_types[hash_type]["hash_valid"]

        data = io.BytesIO(b"Hello" * 1024 * 1024)  # 5MB
        for i in range(num_run):
            data.seek(0)
            hash = hash_func(data)
            yield "."
        assert hash == hash_valid, "%s != %s" % (hash, hash_valid)

    def testCryptHashlib(self, num_run=1, hash_type="sha3_256"):
        """
        Test SHA3 hashing functions
        """
        yield "x 5MB "

        hash_types = {
            "sha3_256": {"func": hashlib.sha3_256, "hash_valid": "c8aeb3ef9fe5d6404871c0d2a4410a4d4e23268e06735648c9596f436c495f7e"},
            "sha3_512": {"func": hashlib.sha3_512, "hash_valid": "b75dba9472d8af3cc945ce49073f3f8214d7ac12086c0453fb08944823dee1ae83b3ffbc87a53a57cc454521d6a26fe73ff0f3be38dddf3f7de5d7692ebc7f95"},
        }

        hash_func = hash_types[hash_type]["func"]
        hash_valid = hash_types[hash_type]["hash_valid"]

        data = io.BytesIO(b"Hello" * 1024 * 1024)  # 5MB
        for i in range(num_run):
            data.seek(0)
            h = hash_func()
            while 1:
                buff = data.read(1024 * 64)
                if not buff:
                    break
                h.update(buff)
            hash = h.hexdigest()
            yield "."
        assert hash == hash_valid, "%s != %s" % (hash, hash_valid)

    def testRandom(self, num_run=1):
        """
        Test generating random data
        """
        yield "x 1000 x 256 bytes "
        for i in range(num_run):
            data_last = None
            for y in range(1000):
                data = os.urandom(256)
                assert data != data_last
                assert len(data) == 256
                data_last = data
            yield "."

    def testHdPrivatekey(self, num_run=2):
        """
        Test generating deterministic private keys from a master seed
        """
        from Crypt import CryptBitcoin
        seed = "e180efa477c63b0f2757eac7b1cce781877177fe0966be62754ffd4c8592ce38"
        privatekeys = []
        for i in range(num_run):
            privatekeys.append(CryptBitcoin.hdPrivatekey(seed, i * 10))
            yield "."
        valid = "5JSbeF5PevdrsYjunqpg7kAGbnCVYa1T4APSL3QRu8EoAmXRc7Y"
        assert privatekeys[0] == valid, "%s != %s" % (privatekeys[0], valid)
        if len(privatekeys) > 1:
            assert privatekeys[0] != privatekeys[-1]

    def testSign(self, num_run=1):
        """
        Test signing data using a private key
        """
        from Crypt import CryptBitcoin
        data = "Hello" * 1024
        privatekey = "5JsunC55XGVqFQj5kPGK4MWgTL26jKbnPhjnmchSNPo75XXCwtk"
        for i in range(num_run):
            yield "."
            sign = CryptBitcoin.sign(data, privatekey)
            valid = "G1GXaDauZ8vX/N9Jn+MRiGm9h+I94zUhDnNYFaqMGuOiBHB+kp4cRPZOL7l1yqK5BHa6J+W97bMjvTXtxzljp6w="
            assert sign == valid, "%s != %s" % (sign, valid)

    def testVerify(self, num_run=1, lib_verify="sslcrypto"):
        """
        Test verification of generated signatures
        """
        from Crypt import CryptBitcoin
        CryptBitcoin.loadLib(lib_verify, silent=True)


        data = "Hello" * 1024
        privatekey = "5JsunC55XGVqFQj5kPGK4MWgTL26jKbnPhjnmchSNPo75XXCwtk"
        address = CryptBitcoin.privatekeyToAddress(privatekey)
        sign = "G1GXaDauZ8vX/N9Jn+MRiGm9h+I94zUhDnNYFaqMGuOiBHB+kp4cRPZOL7l1yqK5BHa6J+W97bMjvTXtxzljp6w="

        for i in range(num_run):
            ok = CryptBitcoin.verify(data, address, sign, lib_verify=lib_verify)
            yield "."
            assert ok, "does not verify from %s" % address

        if lib_verify == "sslcrypto":
            yield("(%s)" % CryptBitcoin.sslcrypto.ecc.get_backend())

    def testPortCheckers(self):
        """
        Test all active open port checker
        """
        from Peer import PeerPortchecker
        for ip_type, func_names in PeerPortchecker.PeerPortchecker.checker_functions.items():
            yield "\n- %s:" % ip_type
            for func_name in func_names:
                yield "\n - Tracker %s: " % func_name
                try:
                    for res in self.testPortChecker(func_name):
                        yield res
                except Exception as err:
                    yield Debug.formatException(err)

    def testPortChecker(self, func_name):
        """
        Test single open port checker
        """
        from Peer import PeerPortchecker
        peer_portchecker = PeerPortchecker.PeerPortchecker(None)
        announce_func = getattr(peer_portchecker, func_name)
        res = announce_func(3894)
        yield res

    def testAll(self):
        """
        Run all tests to check system compatibility with ZeroNet functions
        """
        for progress in self.testBenchmark(online=not config.offline, num_run=1):
            yield progress


@PluginManager.registerTo("ConfigPlugin")
class ConfigPlugin(object):
    def createArguments(self):
        back = super(ConfigPlugin, self).createArguments()
        if self.getCmdlineValue("test") == "benchmark":
            self.test_parser.add_argument(
                '--num_multipler', help='Benchmark run time multipler',
                default=1.0, type=float, metavar='num'
            )
            self.test_parser.add_argument(
                '--filter', help='Filter running benchmark',
                default=None, metavar='test name'
            )
        elif self.getCmdlineValue("test") == "portChecker":
            self.test_parser.add_argument(
                '--func_name', help='Name of open port checker function',
                default=None, metavar='func_name'
            )
        return back