<commit_before>#!/usr/bin/env python3
# Copyright (c) 2015-2018 The Bitcoin Unlimited developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.


# Test Schnorr signatures.
# This test first compares the signatures created in Python with those created by the bitcoind library.
# Next it enables Schnorr signatures on 2 out of 3 nodes, generates a lot of transactions and blocks and verifies that 2 out of 3 nodes accept them.
from test_framework.script import *
from test_framework.util import *
from test_framework.nodemessages import *
from test_framework.test_framework import BitcoinTestFramework
import test_framework.cashlib as cashlib
import test_framework.schnorr as schnorr
import logging
import test_framework.loginit

import time
import sys
if sys.version_info[0] < 3:
    raise "Use Python 3"

class PlaceHolder():
    def __init__(self, d=None):
        self.data = d


class SchnorrSigTest (BitcoinTestFramework):

    def setup_chain(self, bitcoinConfDict=None, wallets=None):
        initialize_chain(self.options.tmpdir, bitcoinConfDict, wallets)

    def setup_network(self, split=False):
        self.nodes = start_nodes(3, self.options.tmpdir)

        # Now interconnect the nodes
        connect_nodes_full(self.nodes)
        # Let the framework know if the network is fully connected.
        # If not, the framework assumes this partition: (0,1) and (2,3)
        # For more complex partitions, you can't use the self.sync* member functions
        self.is_network_split = False
        self.sync_blocks()

    def run_test(self):
        self.sync_blocks()
        self.nodes[0].generate(1)

        # First try a canned sig (taken from schnorr.py)
        privkey = bytes.fromhex("12b004fff7f4b69ef8650e767f18f11ede158148b425660723b9f9a66e61f747")

        pubkey = schnorr.getpubkey(privkey, compressed=True)
        assert pubkey == bytes.fromhex("030b4c866585dd868a9d62348a9cd008d6a312937048fff31670e7e920cfc7a744")

        msg = b"Very deterministic message"
        msghash = hash256(msg)
        assert msghash == bytes.fromhex("5255683da567900bfd3e786ed8836a4e7763c221bf1ac20ece2a5171b9199e8a")

        sig = schnorr.sign(privkey, msghash)
        assert sig == bytes.fromhex("2c56731ac2f7a7e7f11518fc7722a166b02438924ca9d8b4d111347b81d0717571846de67ad3d913a8fdf9d8f3f73161a4c48ae81cb183b214765feb86e255ce")
        sig2 = cashlib.signHashSchnorr(privkey, msghash)
        assert sig2 == sig

        logging.info("random Schnorr signature comparison")
        # Next try random signatures
        for i in range(1, 1000):
            privkey = cashlib.randombytes(32)
            pubkey = schnorr.getpubkey(privkey, compressed=True)
            pubkey2 = cashlib.pubkey(privkey)
            assert pubkey == pubkey2

            msg = cashlib.randombytes(random.randint(0, 10000))
            hsh = cashlib.hash256(msg)

            sigpy = schnorr.sign(privkey, hsh)
            sigcashlib = cashlib.signHashSchnorr(privkey, hsh)
            assert sigpy == sigcashlib

        logging.info("Schnorr signature transaction generation and commitment")

        # create a fork with 2 schnorr nodes and 2 regular nodes
        self.nodes[0].set("consensus.schnorrEnabled=true")
        self.nodes[1].set("consensus.schnorrEnabled=true")

        # grab inputs from 2 different full nodes and sign a single tx that spends them both

        resultWallet = []

        wallets = [self.nodes[0].listunspent(), self.nodes[1].listunspent()]
        for txcount in range(0, 10):
            inputs = [x[0] for x in wallets]
            for x in wallets:  # Remove this utxo so we don't use it in the next time around
                del x[0]
            privb58 = [self.nodes[0].dumpprivkey(inputs[0]["address"]), self.nodes[1].dumpprivkey(inputs[1]["address"])]

            privkeys = [decodeBase58(x)[1:-5] for x in privb58]
            pubkeys = [cashlib.pubkey(x) for x in privkeys]

            for doubleSpend in range(0, 2):  # Double spend this many times
                tx = CTransaction()
                for i in inputs:
                    tx.vin.append(CTxIn(COutPoint(i["txid"], i["vout"]), b"", 0xffffffff-doubleSpend))  # subtracting doubleSpend changes the tx slightly

                destPrivKey = cashlib.randombytes(32)
                destPubKey = cashlib.pubkey(destPrivKey)
                destHash = cashlib.addrbin(destPubKey)

                output = CScript([OP_DUP, OP_HASH160, destHash, OP_EQUALVERIFY, OP_CHECKSIG])

                amt = int(sum([x["amount"] for x in inputs]) * cashlib.BCH)
                tx.vout.append(CTxOut(amt, output))

                sighashtype = 0x41
                n = 0
                for i, priv in zip(inputs, privkeys):
                    sig = cashlib.signTxInputSchnorr(tx, n, i["amount"], i["scriptPubKey"], priv, sighashtype)
                    tx.vin[n].scriptSig = cashlib.spendscript(sig)  # P2PK
                    n += 1

                txhex = hexlify(tx.serialize()).decode("utf-8")
                txid = self.nodes[0].enqueuerawtransaction(txhex)

                if doubleSpend == 0:
                    resultWallet.append([destPrivKey, destPubKey, amt, PlaceHolder(txid), 0, output])

        # because enqueuerawtransaction and propagation is asynchronous we need to wait for it
        waitFor(10, lambda: self.nodes[1].getmempoolinfo()['size'] == txcount+1)
        mp = [i.getmempoolinfo() for i in self.nodes]
        assert txcount+1 == mp[0]['size'] == mp[1]['size']
        assert mp[2]['size'] == 0  # non schnorr enabled nodes should have rejected all of them

        nonSchnorrBlkHash = self.nodes[0].getbestblockhash()
        self.nodes[0].generate(1)
        assert self.nodes[0].getmempoolinfo()['size'] == 0
        waitFor(10, lambda: self.nodes[0].getbestblockhash() == self.nodes[1].getbestblockhash())
        assert self.nodes[2].getbestblockhash() == nonSchnorrBlkHash  # because node 2 will fork off

        FEEINSAT = 5000
        # now spend all the new utxos again
        for spendLoop in range(0, 3):
            incomingWallet = resultWallet
            resultWallet = []

            logging.info("spend iteration %d, num transactions %d" % (spendLoop, len(incomingWallet)))
            for w in incomingWallet:
                txidHolder = PlaceHolder()
                tx = CTransaction()
                tx.vin.append(CTxIn(COutPoint(w[3].data, w[4]), b"", 0xffffffff))

                NOUTS = 10 - spendLoop*2
                if NOUTS < 0:
                    NOUTS = 1
                amtPerOut = int((w[2]-FEEINSAT)/NOUTS)
                for outIdx in range(0, NOUTS):
                    destPrivKey = cashlib.randombytes(32)
                    destPubKey = cashlib.pubkey(destPrivKey)
                    destHash = cashlib.addrbin(destPubKey)
                    output = CScript([OP_DUP, OP_HASH160, destHash, OP_EQUALVERIFY, OP_CHECKSIG])
                    tx.vout.append(CTxOut(amtPerOut, output))
                    resultWallet.append([destPrivKey, destPubKey, amtPerOut, txidHolder, outIdx, output])

                sighashtype = 0x41
                n = 0
                sig = cashlib.signTxInputSchnorr(tx, n, w[2], w[5], w[0], sighashtype)
                tx.vin[n].scriptSig = cashlib.spendscript(sig, w[1])  # P2PKH

                txhex = hexlify(tx.serialize()).decode("utf-8")
                txid = self.nodes[1].enqueuerawtransaction(txhex)
                txidHolder.data = txid

            # because enqueuerawtransaction and propagation is asynchronous we need to wait for it
            waitFor(10, lambda: self.nodes[0].getmempoolinfo()['size'] == len(incomingWallet))
            while self.nodes[0].getmempoolinfo()['size'] != 0:
                self.nodes[0].generate(1)
            waitFor(10, lambda: self.nodes[0].getbestblockhash() == self.nodes[1].getbestblockhash())
            assert self.nodes[2].getbestblockhash() == nonSchnorrBlkHash  # because node 2 will fork off


if __name__ == '__main__':
    binpath = findBitcoind()
    cashlib.init(binpath + os.sep + ".libs" + os.sep + "libbitcoincash.so")
    SchnorrSigTest().main()

# Create a convenient function for an interactive python debugging session
def Test():
    t = SchnorrSigTest()
    t.drop_to_pdb = True
    bitcoinConf = {
        "debug": ["net", "blk", "thin", "mempool", "req", "bench", "evict"],
        "blockprioritysize": 2000000  # we don't want any transactions rejected due to insufficient fees...
    }

    logging.info("PID is %d" % os.getpid())
    flags = []  # ["--nocleanup", "--noshutdown"]
    if os.path.isdir("/ramdisk/test/t"):
        flags.append("--tmpdir=/ramdisk/test/t")
    binpath = findBitcoind()
    flags.append("--srcdir=%s" % binpath)
    cashlib.init(binpath + os.sep + ".libs" + os.sep + "libbitcoincash.so")

    t.main(flags, bitcoinConf, None)<commit_msg>don't run the test if cashlib is not available for this platform<commit_after>#!/usr/bin/env python3
# Copyright (c) 2015-2018 The Bitcoin Unlimited developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.


# Test Schnorr signatures.
# This test first compares the signatures created in Python with those created by the bitcoind library.
# Next it enables Schnorr signatures on 2 out of 3 nodes, generates a lot of transactions and blocks and verifies that 2 out of 3 nodes accept them.
from test_framework.script import *
from test_framework.util import *
from test_framework.nodemessages import *
from test_framework.test_framework import BitcoinTestFramework
import test_framework.cashlib as cashlib
import test_framework.schnorr as schnorr
import logging
import test_framework.loginit

import time
import sys
if sys.version_info[0] < 3:
    raise "Use Python 3"

class PlaceHolder():
    def __init__(self, d=None):
        self.data = d


class SchnorrSigTest (BitcoinTestFramework):

    def setup_chain(self, bitcoinConfDict=None, wallets=None):
        initialize_chain(self.options.tmpdir, bitcoinConfDict, wallets)

    def setup_network(self, split=False):
        self.nodes = start_nodes(3, self.options.tmpdir)

        # Now interconnect the nodes
        connect_nodes_full(self.nodes)
        # Let the framework know if the network is fully connected.
        # If not, the framework assumes this partition: (0,1) and (2,3)
        # For more complex partitions, you can't use the self.sync* member functions
        self.is_network_split = False
        self.sync_blocks()

    def run_test(self):
        self.sync_blocks()
        self.nodes[0].generate(1)

        # First try a canned sig (taken from schnorr.py)
        privkey = bytes.fromhex("12b004fff7f4b69ef8650e767f18f11ede158148b425660723b9f9a66e61f747")

        pubkey = schnorr.getpubkey(privkey, compressed=True)
        assert pubkey == bytes.fromhex("030b4c866585dd868a9d62348a9cd008d6a312937048fff31670e7e920cfc7a744")

        msg = b"Very deterministic message"
        msghash = hash256(msg)
        assert msghash == bytes.fromhex("5255683da567900bfd3e786ed8836a4e7763c221bf1ac20ece2a5171b9199e8a")

        sig = schnorr.sign(privkey, msghash)
        assert sig == bytes.fromhex("2c56731ac2f7a7e7f11518fc7722a166b02438924ca9d8b4d111347b81d0717571846de67ad3d913a8fdf9d8f3f73161a4c48ae81cb183b214765feb86e255ce")
        sig2 = cashlib.signHashSchnorr(privkey, msghash)
        assert sig2 == sig

        logging.info("random Schnorr signature comparison")
        # Next try random signatures
        for i in range(1, 1000):
            privkey = cashlib.randombytes(32)
            pubkey = schnorr.getpubkey(privkey, compressed=True)
            pubkey2 = cashlib.pubkey(privkey)
            assert pubkey == pubkey2

            msg = cashlib.randombytes(random.randint(0, 10000))
            hsh = cashlib.hash256(msg)

            sigpy = schnorr.sign(privkey, hsh)
            sigcashlib = cashlib.signHashSchnorr(privkey, hsh)
            assert sigpy == sigcashlib

        logging.info("Schnorr signature transaction generation and commitment")

        # create a fork with 2 schnorr nodes and 2 regular nodes
        self.nodes[0].set("consensus.schnorrEnabled=true")
        self.nodes[1].set("consensus.schnorrEnabled=true")

        # grab inputs from 2 different full nodes and sign a single tx that spends them both

        resultWallet = []

        wallets = [self.nodes[0].listunspent(), self.nodes[1].listunspent()]
        for txcount in range(0, 10):
            inputs = [x[0] for x in wallets]
            for x in wallets:  # Remove this utxo so we don't use it in the next time around
                del x[0]
            privb58 = [self.nodes[0].dumpprivkey(inputs[0]["address"]), self.nodes[1].dumpprivkey(inputs[1]["address"])]

            privkeys = [decodeBase58(x)[1:-5] for x in privb58]
            pubkeys = [cashlib.pubkey(x) for x in privkeys]

            for doubleSpend in range(0, 2):  # Double spend this many times
                tx = CTransaction()
                for i in inputs:
                    tx.vin.append(CTxIn(COutPoint(i["txid"], i["vout"]), b"", 0xffffffff-doubleSpend))  # subtracting doubleSpend changes the tx slightly

                destPrivKey = cashlib.randombytes(32)
                destPubKey = cashlib.pubkey(destPrivKey)
                destHash = cashlib.addrbin(destPubKey)

                output = CScript([OP_DUP, OP_HASH160, destHash, OP_EQUALVERIFY, OP_CHECKSIG])

                amt = int(sum([x["amount"] for x in inputs]) * cashlib.BCH)
                tx.vout.append(CTxOut(amt, output))

                sighashtype = 0x41
                n = 0
                for i, priv in zip(inputs, privkeys):
                    sig = cashlib.signTxInputSchnorr(tx, n, i["amount"], i["scriptPubKey"], priv, sighashtype)
                    tx.vin[n].scriptSig = cashlib.spendscript(sig)  # P2PK
                    n += 1

                txhex = hexlify(tx.serialize()).decode("utf-8")
                txid = self.nodes[0].enqueuerawtransaction(txhex)

                if doubleSpend == 0:
                    resultWallet.append([destPrivKey, destPubKey, amt, PlaceHolder(txid), 0, output])

        # because enqueuerawtransaction and propagation is asynchronous we need to wait for it
        waitFor(10, lambda: self.nodes[1].getmempoolinfo()['size'] == txcount+1)
        mp = [i.getmempoolinfo() for i in self.nodes]
        assert txcount+1 == mp[0]['size'] == mp[1]['size']
        assert mp[2]['size'] == 0  # non schnorr enabled nodes should have rejected all of them

        nonSchnorrBlkHash = self.nodes[0].getbestblockhash()
        self.nodes[0].generate(1)
        assert self.nodes[0].getmempoolinfo()['size'] == 0
        waitFor(10, lambda: self.nodes[0].getbestblockhash() == self.nodes[1].getbestblockhash())
        assert self.nodes[2].getbestblockhash() == nonSchnorrBlkHash  # because node 2 will fork off

        FEEINSAT = 5000
        # now spend all the new utxos again
        for spendLoop in range(0, 3):
            incomingWallet = resultWallet
            resultWallet = []

            logging.info("spend iteration %d, num transactions %d" % (spendLoop, len(incomingWallet)))
            for w in incomingWallet:
                txidHolder = PlaceHolder()
                tx = CTransaction()
                tx.vin.append(CTxIn(COutPoint(w[3].data, w[4]), b"", 0xffffffff))

                NOUTS = 10 - spendLoop*2
                if NOUTS < 0:
                    NOUTS = 1
                amtPerOut = int((w[2]-FEEINSAT)/NOUTS)
                for outIdx in range(0, NOUTS):
                    destPrivKey = cashlib.randombytes(32)
                    destPubKey = cashlib.pubkey(destPrivKey)
                    destHash = cashlib.addrbin(destPubKey)
                    output = CScript([OP_DUP, OP_HASH160, destHash, OP_EQUALVERIFY, OP_CHECKSIG])
                    tx.vout.append(CTxOut(amtPerOut, output))
                    resultWallet.append([destPrivKey, destPubKey, amtPerOut, txidHolder, outIdx, output])

                sighashtype = 0x41
                n = 0
                sig = cashlib.signTxInputSchnorr(tx, n, w[2], w[5], w[0], sighashtype)
                tx.vin[n].scriptSig = cashlib.spendscript(sig, w[1])  # P2PKH

                txhex = hexlify(tx.serialize()).decode("utf-8")
                txid = self.nodes[1].enqueuerawtransaction(txhex)
                txidHolder.data = txid

            # because enqueuerawtransaction and propagation is asynchronous we need to wait for it
            waitFor(10, lambda: self.nodes[0].getmempoolinfo()['size'] == len(incomingWallet))
            while self.nodes[0].getmempoolinfo()['size'] != 0:
                self.nodes[0].generate(1)
            waitFor(10, lambda: self.nodes[0].getbestblockhash() == self.nodes[1].getbestblockhash())
            assert self.nodes[2].getbestblockhash() == nonSchnorrBlkHash  # because node 2 will fork off


if __name__ == '__main__':
    binpath = findBitcoind()
    try:
        cashlib.init(binpath + os.sep + ".libs" + os.sep + "libbitcoincash.so")
        SchnorrSigTest().main()
    except OSError as e:
        print("Issue loading cashlib shared library.  This is expected during cross compilation since the native python will not load the .so so no error will be reported: %s" % str(e))


# Create a convenient function for an interactive python debugging session
def Test():
    t = SchnorrSigTest()
    t.drop_to_pdb = True
    bitcoinConf = {
        "debug": ["net", "blk", "thin", "mempool", "req", "bench", "evict"],
        "blockprioritysize": 2000000  # we don't want any transactions rejected due to insufficient fees...
    }

    logging.info("PID is %d" % os.getpid())
    flags = []  # ["--nocleanup", "--noshutdown"]
    if os.path.isdir("/ramdisk/test/t"):
        flags.append("--tmpdir=/ramdisk/test/t")
    binpath = findBitcoind()
    flags.append("--srcdir=%s" % binpath)
    cashlib.init(binpath + os.sep + ".libs" + os.sep + "libbitcoincash.so")

    t.main(flags, bitcoinConf, None)
{'fraction_non_alphanumeric': 0.10533260032985156, 'fraction_numerical': 0.04145134689389775, 'mean_word_length': 2.569662480376766, 'pattern_counts': {'":': 4, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1974396', 'n_tokens_mistral': 6170, 'n_tokens_neox': 5584, 'n_words': 1373}