aboutsummaryrefslogtreecommitdiff
path: root/test/functional
diff options
context:
space:
mode:
Diffstat (limited to 'test/functional')
-rw-r--r--test/functional/data/invalid_txs.py6
-rwxr-xr-xtest/functional/feature_assumevalid.py8
-rwxr-xr-xtest/functional/feature_block.py26
-rwxr-xr-xtest/functional/feature_pruning.py21
-rwxr-xr-xtest/functional/p2p_blocksonly.py58
-rwxr-xr-xtest/functional/p2p_segwit.py72
-rwxr-xr-xtest/functional/rpc_createmultisig.py74
-rwxr-xr-xtest/functional/rpc_psbt.py8
-rw-r--r--test/functional/test_framework/key.py574
-rwxr-xr-xtest/functional/test_framework/mininode.py8
-rw-r--r--test/functional/test_framework/util.py2
-rwxr-xr-xtest/functional/test_runner.py1
12 files changed, 578 insertions, 280 deletions
diff --git a/test/functional/data/invalid_txs.py b/test/functional/data/invalid_txs.py
index 02deae92f3..d262dae5aa 100644
--- a/test/functional/data/invalid_txs.py
+++ b/test/functional/data/invalid_txs.py
@@ -71,9 +71,13 @@ class InputMissing(BadTxTemplate):
reject_reason = "bad-txns-vin-empty"
expect_disconnect = False
+ # We use a blank transaction here to make sure
+ # it is interpreted as a non-witness transaction.
+ # Otherwise the transaction will fail the
+ # "surpufluous witness" check during deserialization
+ # rather than the input count check.
def get_tx(self):
tx = CTransaction()
- tx.vout.append(CTxOut(0, sc.CScript([sc.OP_TRUE] * 100)))
tx.calc_sha256()
return tx
diff --git a/test/functional/feature_assumevalid.py b/test/functional/feature_assumevalid.py
index 0eb3dd440b..7d6a56919d 100755
--- a/test/functional/feature_assumevalid.py
+++ b/test/functional/feature_assumevalid.py
@@ -32,7 +32,7 @@ Start three nodes:
import time
from test_framework.blocktools import (create_block, create_coinbase)
-from test_framework.key import CECKey
+from test_framework.key import ECKey
from test_framework.messages import (
CBlockHeader,
COutPoint,
@@ -104,9 +104,9 @@ class AssumeValidTest(BitcoinTestFramework):
self.blocks = []
# Get a pubkey for the coinbase TXO
- coinbase_key = CECKey()
- coinbase_key.set_secretbytes(b"horsebattery")
- coinbase_pubkey = coinbase_key.get_pubkey()
+ coinbase_key = ECKey()
+ coinbase_key.generate()
+ coinbase_pubkey = coinbase_key.get_pubkey().get_bytes()
# Create the first block with a coinbase output to our key
height = 1
diff --git a/test/functional/feature_block.py b/test/functional/feature_block.py
index 5253ff7aaa..1aa5ae1423 100755
--- a/test/functional/feature_block.py
+++ b/test/functional/feature_block.py
@@ -14,7 +14,7 @@ from test_framework.blocktools import (
get_legacy_sigopcount_block,
MAX_BLOCK_SIGOPS,
)
-from test_framework.key import CECKey
+from test_framework.key import ECKey
from test_framework.messages import (
CBlock,
COIN,
@@ -86,9 +86,9 @@ class FullBlockTest(BitcoinTestFramework):
self.bootstrap_p2p() # Add one p2p connection to the node
self.block_heights = {}
- self.coinbase_key = CECKey()
- self.coinbase_key.set_secretbytes(b"horsebattery")
- self.coinbase_pubkey = self.coinbase_key.get_pubkey()
+ self.coinbase_key = ECKey()
+ self.coinbase_key.generate()
+ self.coinbase_pubkey = self.coinbase_key.get_pubkey().get_bytes()
self.tip = None
self.blocks = {}
self.genesis_hash = int(self.nodes[0].getbestblockhash(), 16)
@@ -146,20 +146,6 @@ class FullBlockTest(BitcoinTestFramework):
badtx = template.get_tx()
if TxTemplate != invalid_txs.InputMissing:
self.sign_tx(badtx, attempt_spend_tx)
- else:
- # Segwit is active in regtest at this point, so to deserialize a
- # transaction without any inputs correctly, we set the outputs
- # to an empty list. This is a hack, as the serialization of an
- # empty list of outputs is deserialized as flags==0 and thus
- # deserialization of the outputs is skipped.
- # A policy check requires "loose" txs to be of a minimum size,
- # so vtx is not set to be empty in the TxTemplate class and we
- # only apply the workaround where txs are not "loose", i.e. in
- # blocks.
- #
- # The workaround has the purpose that both sides calculate
- # the same tx hash in the merkle tree
- badtx.vout = []
badtx.rehash()
badblock = self.update_block(blockname, [badtx])
self.sync_blocks(
@@ -528,7 +514,7 @@ class FullBlockTest(BitcoinTestFramework):
tx.vin.append(CTxIn(COutPoint(b39.vtx[i].sha256, 0), b''))
# Note: must pass the redeem_script (not p2sh_script) to the signature hash function
(sighash, err) = SignatureHash(redeem_script, tx, 1, SIGHASH_ALL)
- sig = self.coinbase_key.sign(sighash) + bytes(bytearray([SIGHASH_ALL]))
+ sig = self.coinbase_key.sign_ecdsa(sighash) + bytes(bytearray([SIGHASH_ALL]))
scriptSig = CScript([sig, redeem_script])
tx.vin[1].scriptSig = scriptSig
@@ -1284,7 +1270,7 @@ class FullBlockTest(BitcoinTestFramework):
tx.vin[0].scriptSig = CScript()
return
(sighash, err) = SignatureHash(spend_tx.vout[0].scriptPubKey, tx, 0, SIGHASH_ALL)
- tx.vin[0].scriptSig = CScript([self.coinbase_key.sign(sighash) + bytes(bytearray([SIGHASH_ALL]))])
+ tx.vin[0].scriptSig = CScript([self.coinbase_key.sign_ecdsa(sighash) + bytes(bytearray([SIGHASH_ALL]))])
def create_and_sign_transaction(self, spend_tx, value, script=CScript([OP_TRUE])):
tx = self.create_tx(spend_tx, 0, value, script)
diff --git a/test/functional/feature_pruning.py b/test/functional/feature_pruning.py
index 3e1ba88f0a..b3dd699fda 100755
--- a/test/functional/feature_pruning.py
+++ b/test/functional/feature_pruning.py
@@ -10,12 +10,10 @@ This test takes 30 mins or more (up to 2 hours)
"""
from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import assert_equal, assert_greater_than, assert_raises_rpc_error, connect_nodes, mine_large_block, sync_blocks, wait_until
+from test_framework.util import assert_equal, assert_raises_rpc_error, connect_nodes, mine_large_block, sync_blocks, wait_until
import os
-MIN_BLOCKS_TO_KEEP = 288
-
# Rescans start at the earliest block up to 2 hours before a key timestamp, so
# the manual prune RPC avoids pruning blocks in the same window to be
# compatible with pruning based on key creation time.
@@ -250,20 +248,9 @@ class PruneTest(BitcoinTestFramework):
else:
return index
- def prune(index, expected_ret=None):
+ def prune(index):
ret = node.pruneblockchain(height=height(index))
- # Check the return value. When use_timestamp is True, just check
- # that the return value is less than or equal to the expected
- # value, because when more than one block is generated per second,
- # a timestamp will not be granular enough to uniquely identify an
- # individual block.
- if expected_ret is None:
- expected_ret = index
- if use_timestamp:
- assert_greater_than(ret, 0)
- assert_greater_than(expected_ret + 1, ret)
- else:
- assert_equal(ret, expected_ret)
+ assert_equal(ret, node.getblockchaininfo()['pruneheight'])
def has_block(index):
return os.path.isfile(os.path.join(self.nodes[node_number].datadir, "regtest", "blocks", "blk{:05}.dat".format(index)))
@@ -308,7 +295,7 @@ class PruneTest(BitcoinTestFramework):
raise AssertionError("blk00001.dat is still there, should be pruned by now")
# height=1000 should not prune anything more, because tip-288 is in blk00002.dat.
- prune(1000, 1001 - MIN_BLOCKS_TO_KEEP)
+ prune(1000)
if not has_block(2):
raise AssertionError("blk00002.dat is still there, should be pruned by now")
diff --git a/test/functional/p2p_blocksonly.py b/test/functional/p2p_blocksonly.py
new file mode 100755
index 0000000000..12cb06a407
--- /dev/null
+++ b/test/functional/p2p_blocksonly.py
@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+# Copyright (c) 2019 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Test p2p blocksonly"""
+
+from test_framework.messages import msg_tx, CTransaction, FromHex
+from test_framework.mininode import P2PInterface
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import assert_equal
+
+
+class P2PBlocksOnly(BitcoinTestFramework):
+ def set_test_params(self):
+ self.setup_clean_chain = False
+ self.num_nodes = 1
+ self.extra_args = [["-blocksonly"]]
+
+ def run_test(self):
+ self.nodes[0].add_p2p_connection(P2PInterface())
+
+ self.log.info('Check that txs from p2p are rejected')
+ prevtx = self.nodes[0].getblock(self.nodes[0].getblockhash(1), 2)['tx'][0]
+ rawtx = self.nodes[0].createrawtransaction(
+ inputs=[{
+ 'txid': prevtx['txid'],
+ 'vout': 0
+ }],
+ outputs=[{
+ self.nodes[0].get_deterministic_priv_key().address: 50 - 0.00125
+ }],
+ )
+ sigtx = self.nodes[0].signrawtransactionwithkey(
+ hexstring=rawtx,
+ privkeys=[self.nodes[0].get_deterministic_priv_key().key],
+ prevtxs=[{
+ 'txid': prevtx['txid'],
+ 'vout': 0,
+ 'scriptPubKey': prevtx['vout'][0]['scriptPubKey']['hex'],
+ }],
+ )['hex']
+ assert_equal(self.nodes[0].getnetworkinfo()['localrelay'], False)
+ with self.nodes[0].assert_debug_log(['transaction sent in violation of protocol peer=0']):
+ self.nodes[0].p2p.send_message(msg_tx(FromHex(CTransaction(), sigtx)))
+ self.nodes[0].p2p.sync_with_ping()
+ assert_equal(self.nodes[0].getmempoolinfo()['size'], 0)
+
+ self.log.info('Check that txs from rpc are not rejected and relayed to other peers')
+ assert_equal(self.nodes[0].getpeerinfo()[0]['relaytxes'], True)
+ txid = self.nodes[0].testmempoolaccept([sigtx])[0]['txid']
+ with self.nodes[0].assert_debug_log(['received getdata for: tx {} peer=0'.format(txid)]):
+ self.nodes[0].sendrawtransaction(sigtx)
+ self.nodes[0].p2p.wait_for_tx(txid)
+ assert_equal(self.nodes[0].getmempoolinfo()['size'], 1)
+
+
+if __name__ == '__main__':
+ P2PBlocksOnly().main()
diff --git a/test/functional/p2p_segwit.py b/test/functional/p2p_segwit.py
index 8f8e89cf15..62b12507ed 100755
--- a/test/functional/p2p_segwit.py
+++ b/test/functional/p2p_segwit.py
@@ -10,7 +10,7 @@ import struct
import time
from test_framework.blocktools import create_block, create_coinbase, add_witness_commitment, get_witness_script, WITNESS_COMMITMENT_HEADER
-from test_framework.key import CECKey, CPubKey
+from test_framework.key import ECKey
from test_framework.messages import (
BIP125_SEQUENCE_NUMBER,
CBlock,
@@ -37,6 +37,7 @@ from test_framework.messages import (
ser_vector,
sha256,
uint256_from_str,
+ FromHex,
)
from test_framework.mininode import (
P2PInterface,
@@ -81,6 +82,7 @@ from test_framework.util import (
hex_str_to_bytes,
sync_blocks,
sync_mempools,
+ assert_raises_rpc_error,
)
# The versionbit bit used to signal activation of SegWit
@@ -104,7 +106,7 @@ def get_p2pkh_script(pubkeyhash):
def sign_p2pk_witness_input(script, tx_to, in_idx, hashtype, value, key):
"""Add signature for a P2PK witness program."""
tx_hash = SegwitVersion1SignatureHash(script, tx_to, in_idx, hashtype, value)
- signature = key.sign(tx_hash) + chr(hashtype).encode('latin-1')
+ signature = key.sign_ecdsa(tx_hash) + chr(hashtype).encode('latin-1')
tx_to.wit.vtxinwit[in_idx].scriptWitness.stack = [signature, script]
tx_to.rehash()
@@ -273,6 +275,7 @@ class SegWitTest(BitcoinTestFramework):
self.test_non_standard_witness()
self.test_upgrade_after_activation()
self.test_witness_sigops()
+ self.test_superfluous_witness()
# Individual tests
@@ -1483,10 +1486,9 @@ class SegWitTest(BitcoinTestFramework):
# Segwit transactions using uncompressed pubkeys are not accepted
# under default policy, but should still pass consensus.
- key = CECKey()
- key.set_secretbytes(b"9")
- key.set_compressed(False)
- pubkey = CPubKey(key.get_pubkey())
+ key = ECKey()
+ key.generate(False)
+ pubkey = key.get_pubkey().get_bytes()
assert_equal(len(pubkey), 65) # This should be an uncompressed pubkey
utxo = self.utxo.pop(0)
@@ -1516,7 +1518,7 @@ class SegWitTest(BitcoinTestFramework):
tx2.vout.append(CTxOut(tx.vout[0].nValue - 1000, script_wsh))
script = get_p2pkh_script(pubkeyhash)
sig_hash = SegwitVersion1SignatureHash(script, tx2, 0, SIGHASH_ALL, tx.vout[0].nValue)
- signature = key.sign(sig_hash) + b'\x01' # 0x1 is SIGHASH_ALL
+ signature = key.sign_ecdsa(sig_hash) + b'\x01' # 0x1 is SIGHASH_ALL
tx2.wit.vtxinwit.append(CTxInWitness())
tx2.wit.vtxinwit[0].scriptWitness.stack = [signature, pubkey]
tx2.rehash()
@@ -1570,7 +1572,7 @@ class SegWitTest(BitcoinTestFramework):
tx5.vin.append(CTxIn(COutPoint(tx4.sha256, 0), b""))
tx5.vout.append(CTxOut(tx4.vout[0].nValue - 1000, CScript([OP_TRUE])))
(sig_hash, err) = SignatureHash(script_pubkey, tx5, 0, SIGHASH_ALL)
- signature = key.sign(sig_hash) + b'\x01' # 0x1 is SIGHASH_ALL
+ signature = key.sign_ecdsa(sig_hash) + b'\x01' # 0x1 is SIGHASH_ALL
tx5.vin[0].scriptSig = CScript([signature, pubkey])
tx5.rehash()
# Should pass policy and consensus.
@@ -1583,9 +1585,9 @@ class SegWitTest(BitcoinTestFramework):
@subtest
def test_signature_version_1(self):
- key = CECKey()
- key.set_secretbytes(b"9")
- pubkey = CPubKey(key.get_pubkey())
+ key = ECKey()
+ key.generate()
+ pubkey = key.get_pubkey().get_bytes()
witness_program = CScript([pubkey, CScriptOp(OP_CHECKSIG)])
witness_hash = sha256(witness_program)
@@ -1720,7 +1722,7 @@ class SegWitTest(BitcoinTestFramework):
script = get_p2pkh_script(pubkeyhash)
sig_hash = SegwitVersion1SignatureHash(script, tx2, 0, SIGHASH_ALL, tx.vout[0].nValue)
- signature = key.sign(sig_hash) + b'\x01' # 0x1 is SIGHASH_ALL
+ signature = key.sign_ecdsa(sig_hash) + b'\x01' # 0x1 is SIGHASH_ALL
# Check that we can't have a scriptSig
tx2.vin[0].scriptSig = CScript([signature, pubkey])
@@ -2039,5 +2041,51 @@ class SegWitTest(BitcoinTestFramework):
# TODO: test p2sh sigop counting
+ def test_superfluous_witness(self):
+ # Serialization of tx that puts witness flag to 3 always
+ def serialize_with_bogus_witness(tx):
+ flags = 3
+ r = b""
+ r += struct.pack("<i", tx.nVersion)
+ if flags:
+ dummy = []
+ r += ser_vector(dummy)
+ r += struct.pack("<B", flags)
+ r += ser_vector(tx.vin)
+ r += ser_vector(tx.vout)
+ if flags & 1:
+ if (len(tx.wit.vtxinwit) != len(tx.vin)):
+ # vtxinwit must have the same length as vin
+ tx.wit.vtxinwit = tx.wit.vtxinwit[:len(tx.vin)]
+ for i in range(len(tx.wit.vtxinwit), len(tx.vin)):
+ tx.wit.vtxinwit.append(CTxInWitness())
+ r += tx.wit.serialize()
+ r += struct.pack("<I", tx.nLockTime)
+ return r
+
+ class msg_bogus_tx(msg_tx):
+ def serialize(self):
+ return serialize_with_bogus_witness(self.tx)
+
+ self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(address_type='bech32'), 5)
+ self.nodes[0].generate(1)
+ unspent = next(u for u in self.nodes[0].listunspent() if u['spendable'] and u['address'].startswith('bcrt'))
+
+ raw = self.nodes[0].createrawtransaction([{"txid": unspent['txid'], "vout": unspent['vout']}], {self.nodes[0].getnewaddress(): 1})
+ tx = FromHex(CTransaction(), raw)
+ assert_raises_rpc_error(-22, "TX decode failed", self.nodes[0].decoderawtransaction, serialize_with_bogus_witness(tx).hex())
+ with self.nodes[0].assert_debug_log(['Superfluous witness record']):
+ self.nodes[0].p2p.send_message(msg_bogus_tx(tx))
+ self.nodes[0].p2p.sync_with_ping()
+ raw = self.nodes[0].signrawtransactionwithwallet(raw)
+ assert raw['complete']
+ raw = raw['hex']
+ tx = FromHex(CTransaction(), raw)
+ assert_raises_rpc_error(-22, "TX decode failed", self.nodes[0].decoderawtransaction, serialize_with_bogus_witness(tx).hex())
+ with self.nodes[0].assert_debug_log(['Unknown transaction optional data']):
+ self.nodes[0].p2p.send_message(msg_bogus_tx(tx))
+ self.nodes[0].p2p.sync_with_ping()
+
+
if __name__ == '__main__':
SegWitTest().main()
diff --git a/test/functional/rpc_createmultisig.py b/test/functional/rpc_createmultisig.py
index 3cc35a7b9a..1f717f59d0 100755
--- a/test/functional/rpc_createmultisig.py
+++ b/test/functional/rpc_createmultisig.py
@@ -1,11 +1,19 @@
#!/usr/bin/env python3
-# Copyright (c) 2015-2018 The Bitcoin Core developers
+# Copyright (c) 2015-2019 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
-"""Test transaction signing using the signrawtransaction* RPCs."""
+"""Test multisig RPCs"""
from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import (
+ assert_raises_rpc_error,
+ assert_equal,
+)
+from test_framework.key import ECPubKey
+
+import binascii
import decimal
+import itertools
class RpcCreateMultiSigTest(BitcoinTestFramework):
def set_test_params(self):
@@ -17,29 +25,64 @@ class RpcCreateMultiSigTest(BitcoinTestFramework):
def get_keys(self):
node0, node1, node2 = self.nodes
- self.add = [node1.getnewaddress() for _ in range(self.nkeys)]
- self.pub = [node1.getaddressinfo(a)["pubkey"] for a in self.add]
- self.priv = [node1.dumpprivkey(a) for a in self.add]
+ add = [node1.getnewaddress() for _ in range(self.nkeys)]
+ self.pub = [node1.getaddressinfo(a)["pubkey"] for a in add]
+ self.priv = [node1.dumpprivkey(a) for a in add]
self.final = node2.getnewaddress()
def run_test(self):
- node0,node1,node2 = self.nodes
+ node0, node1, node2 = self.nodes
- # 50 BTC each, rest will be 25 BTC each
+ self.check_addmultisigaddress_errors()
+
+ self.log.info('Generating blocks ...')
node0.generate(149)
self.sync_all()
self.moved = 0
- for self.nkeys in [3,5]:
- for self.nsigs in [2,3]:
+ for self.nkeys in [3, 5]:
+ for self.nsigs in [2, 3]:
for self.output_type in ["bech32", "p2sh-segwit", "legacy"]:
self.get_keys()
self.do_multisig()
self.checkbalances()
+ # Test mixed compressed and uncompressed pubkeys
+ self.log.info('Mixed compressed and uncompressed multisigs are not allowed')
+ pk0 = node0.getaddressinfo(node0.getnewaddress())['pubkey']
+ pk1 = node1.getaddressinfo(node1.getnewaddress())['pubkey']
+ pk2 = node2.getaddressinfo(node2.getnewaddress())['pubkey']
+
+ # decompress pk2
+ pk_obj = ECPubKey()
+ pk_obj.set(binascii.unhexlify(pk2))
+ pk_obj.compressed = False
+ pk2 = binascii.hexlify(pk_obj.get_bytes()).decode()
+
+ # Check all permutations of keys because order matters apparently
+ for keys in itertools.permutations([pk0, pk1, pk2]):
+ # Results should be the same as this legacy one
+ legacy_addr = node0.createmultisig(2, keys, 'legacy')['address']
+ assert_equal(legacy_addr, node0.addmultisigaddress(2, keys, '', 'legacy')['address'])
+
+ # Generate addresses with the segwit types. These should all make legacy addresses
+ assert_equal(legacy_addr, node0.createmultisig(2, keys, 'bech32')['address'])
+ assert_equal(legacy_addr, node0.createmultisig(2, keys, 'p2sh-segwit')['address'])
+ assert_equal(legacy_addr, node0.addmultisigaddress(2, keys, '', 'bech32')['address'])
+ assert_equal(legacy_addr, node0.addmultisigaddress(2, keys, '', 'p2sh-segwit')['address'])
+
+ def check_addmultisigaddress_errors(self):
+ self.log.info('Check that addmultisigaddress fails when the private keys are missing')
+ addresses = [self.nodes[1].getnewaddress(address_type='legacy') for _ in range(2)]
+ assert_raises_rpc_error(-5, 'no full public key for address', lambda: self.nodes[0].addmultisigaddress(nrequired=1, keys=addresses))
+ for a in addresses:
+ # Importing all addresses should not change the result
+ self.nodes[0].importaddress(a)
+ assert_raises_rpc_error(-5, 'no full public key for address', lambda: self.nodes[0].addmultisigaddress(nrequired=1, keys=addresses))
+
def checkbalances(self):
- node0,node1,node2 = self.nodes
+ node0, node1, node2 = self.nodes
node0.generate(100)
self.sync_all()
@@ -49,13 +92,13 @@ class RpcCreateMultiSigTest(BitcoinTestFramework):
height = node0.getblockchaininfo()["blocks"]
assert 150 < height < 350
- total = 149*50 + (height-149-100)*25
+ total = 149 * 50 + (height - 149 - 100) * 25
assert bal1 == 0
assert bal2 == self.moved
- assert bal0+bal1+bal2 == total
+ assert bal0 + bal1 + bal2 == total
def do_multisig(self):
- node0,node1,node2 = self.nodes
+ node0, node1, node2 = self.nodes
msig = node2.createmultisig(self.nsigs, self.pub, self.output_type)
madd = msig["address"]
@@ -74,7 +117,7 @@ class RpcCreateMultiSigTest(BitcoinTestFramework):
txid = node0.sendtoaddress(madd, 40)
tx = node0.getrawtransaction(txid, True)
- vout = [v["n"] for v in tx["vout"] if madd in v["scriptPubKey"].get("addresses",[])]
+ vout = [v["n"] for v in tx["vout"] if madd in v["scriptPubKey"].get("addresses", [])]
assert len(vout) == 1
vout = vout[0]
scriptPubKey = tx["vout"][vout]["scriptPubKey"]["hex"]
@@ -86,7 +129,7 @@ class RpcCreateMultiSigTest(BitcoinTestFramework):
outval = value - decimal.Decimal("0.00001000")
rawtx = node2.createrawtransaction([{"txid": txid, "vout": vout}], [{self.final: outval}])
- rawtx2 = node2.signrawtransactionwithkey(rawtx, self.priv[0:self.nsigs-1], prevtxs)
+ rawtx2 = node2.signrawtransactionwithkey(rawtx, self.priv[0:self.nsigs - 1], prevtxs)
rawtx3 = node2.signrawtransactionwithkey(rawtx2["hex"], [self.priv[-1]], prevtxs)
self.moved += outval
@@ -97,5 +140,6 @@ class RpcCreateMultiSigTest(BitcoinTestFramework):
txinfo = node0.getrawtransaction(tx, True, blk)
self.log.info("n/m=%d/%d %s size=%d vsize=%d weight=%d" % (self.nsigs, self.nkeys, self.output_type, txinfo["size"], txinfo["vsize"], txinfo["weight"]))
+
if __name__ == '__main__':
RpcCreateMultiSigTest().main()
diff --git a/test/functional/rpc_psbt.py b/test/functional/rpc_psbt.py
index 53f606f2ec..6c0aec9228 100755
--- a/test/functional/rpc_psbt.py
+++ b/test/functional/rpc_psbt.py
@@ -144,10 +144,12 @@ class PSBTTest(BitcoinTestFramework):
new_psbt = self.nodes[0].converttopsbt(rawtx['hex'])
self.nodes[0].decodepsbt(new_psbt)
- # Make sure that a psbt with signatures cannot be converted
+ # Make sure that a non-psbt with signatures cannot be converted
+ # Error could be either "TX decode failed" (segwit inputs causes parsing to fail) or "Inputs must not have scriptSigs and scriptWitnesses"
+ # We must set iswitness=True because the serialized transaction has inputs and is therefore a witness transaction
signedtx = self.nodes[0].signrawtransactionwithwallet(rawtx['hex'])
- assert_raises_rpc_error(-22, "TX decode failed", self.nodes[0].converttopsbt, signedtx['hex'])
- assert_raises_rpc_error(-22, "TX decode failed", self.nodes[0].converttopsbt, signedtx['hex'], False)
+ assert_raises_rpc_error(-22, "", self.nodes[0].converttopsbt, hexstring=signedtx['hex'], iswitness=True)
+ assert_raises_rpc_error(-22, "", self.nodes[0].converttopsbt, hexstring=signedtx['hex'], permitsigdata=False, iswitness=True)
# Unless we allow it to convert and strip signatures
self.nodes[0].converttopsbt(signedtx['hex'], True)
diff --git a/test/functional/test_framework/key.py b/test/functional/test_framework/key.py
index 1b3e510dc4..912c0ca978 100644
--- a/test/functional/test_framework/key.py
+++ b/test/functional/test_framework/key.py
@@ -1,226 +1,386 @@
-# Copyright (c) 2011 Sam Rushing
-"""ECC secp256k1 OpenSSL wrapper.
+# Copyright (c) 2019 Pieter Wuille
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Test-only secp256k1 elliptic curve implementation
-WARNING: This module does not mlock() secrets; your private keys may end up on
-disk in swap! Use with caution!
+WARNING: This code is slow, uses bad randomness, does not properly protect
+keys, and is trivially vulnerable to side channel attacks. Do not use for
+anything but tests."""
+import random
-This file is modified from python-bitcoinlib.
-"""
-
-import ctypes
-import ctypes.util
-import hashlib
-
-ssl = ctypes.cdll.LoadLibrary(ctypes.util.find_library ('ssl') or 'libeay32')
-
-ssl.BN_new.restype = ctypes.c_void_p
-ssl.BN_new.argtypes = []
-
-ssl.BN_bin2bn.restype = ctypes.c_void_p
-ssl.BN_bin2bn.argtypes = [ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p]
-
-ssl.BN_CTX_free.restype = None
-ssl.BN_CTX_free.argtypes = [ctypes.c_void_p]
-
-ssl.BN_CTX_new.restype = ctypes.c_void_p
-ssl.BN_CTX_new.argtypes = []
-
-ssl.ECDH_compute_key.restype = ctypes.c_int
-ssl.ECDH_compute_key.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p]
-
-ssl.ECDSA_sign.restype = ctypes.c_int
-ssl.ECDSA_sign.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p]
-
-ssl.ECDSA_verify.restype = ctypes.c_int
-ssl.ECDSA_verify.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p]
-
-ssl.EC_KEY_free.restype = None
-ssl.EC_KEY_free.argtypes = [ctypes.c_void_p]
-
-ssl.EC_KEY_new_by_curve_name.restype = ctypes.c_void_p
-ssl.EC_KEY_new_by_curve_name.argtypes = [ctypes.c_int]
-
-ssl.EC_KEY_get0_group.restype = ctypes.c_void_p
-ssl.EC_KEY_get0_group.argtypes = [ctypes.c_void_p]
-
-ssl.EC_KEY_get0_public_key.restype = ctypes.c_void_p
-ssl.EC_KEY_get0_public_key.argtypes = [ctypes.c_void_p]
-
-ssl.EC_KEY_set_private_key.restype = ctypes.c_int
-ssl.EC_KEY_set_private_key.argtypes = [ctypes.c_void_p, ctypes.c_void_p]
-
-ssl.EC_KEY_set_conv_form.restype = None
-ssl.EC_KEY_set_conv_form.argtypes = [ctypes.c_void_p, ctypes.c_int]
-
-ssl.EC_KEY_set_public_key.restype = ctypes.c_int
-ssl.EC_KEY_set_public_key.argtypes = [ctypes.c_void_p, ctypes.c_void_p]
-
-ssl.i2o_ECPublicKey.restype = ctypes.c_void_p
-ssl.i2o_ECPublicKey.argtypes = [ctypes.c_void_p, ctypes.c_void_p]
-
-ssl.EC_POINT_new.restype = ctypes.c_void_p
-ssl.EC_POINT_new.argtypes = [ctypes.c_void_p]
-
-ssl.EC_POINT_free.restype = None
-ssl.EC_POINT_free.argtypes = [ctypes.c_void_p]
-
-ssl.EC_POINT_mul.restype = ctypes.c_int
-ssl.EC_POINT_mul.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p]
-
-# this specifies the curve used with ECDSA.
-NID_secp256k1 = 714 # from openssl/obj_mac.h
+def modinv(a, n):
+ """Compute the modular inverse of a modulo n
+ See https://en.wikipedia.org/wiki/Extended_Euclidean_algorithm#Modular_integers.
+ """
+ t1, t2 = 0, 1
+ r1, r2 = n, a
+ while r2 != 0:
+ q = r1 // r2
+ t1, t2 = t2, t1 - q * t2
+ r1, r2 = r2, r1 - q * r2
+ if r1 > 1:
+ return None
+ if t1 < 0:
+ t1 += n
+ return t1
+
+def jacobi_symbol(n, k):
+ """Compute the Jacobi symbol of n modulo k
+
+ See http://en.wikipedia.org/wiki/Jacobi_symbol
+
+ For our application k is always prime, so this is the same as the Legendre symbol."""
+ assert k > 0 and k & 1, "jacobi symbol is only defined for positive odd k"
+ n %= k
+ t = 0
+ while n != 0:
+ while n & 1 == 0:
+ n >>= 1
+ r = k & 7
+ t ^= (r == 3 or r == 5)
+ n, k = k, n
+ t ^= (n & k & 3 == 3)
+ n = n % k
+ if k == 1:
+ return -1 if t else 1
+ return 0
+
+def modsqrt(a, p):
+ """Compute the square root of a modulo p when p % 4 = 3.
+
+ The Tonelli-Shanks algorithm can be used. See https://en.wikipedia.org/wiki/Tonelli-Shanks_algorithm
+
+ Limiting this function to only work for p % 4 = 3 means we don't need to
+ iterate through the loop. The highest n such that p - 1 = 2^n Q with Q odd
+ is n = 1. Therefore Q = (p-1)/2 and sqrt = a^((Q+1)/2) = a^((p+1)/4)
+
+ secp256k1's is defined over field of size 2**256 - 2**32 - 977, which is 3 mod 4.
+ """
+ if p % 4 != 3:
+ raise NotImplementedError("modsqrt only implemented for p % 4 = 3")
+ sqrt = pow(a, (p + 1)//4, p)
+ if pow(sqrt, 2, p) == a % p:
+ return sqrt
+ return None
+
+class EllipticCurve:
+ def __init__(self, p, a, b):
+ """Initialize elliptic curve y^2 = x^3 + a*x + b over GF(p)."""
+ self.p = p
+ self.a = a % p
+ self.b = b % p
+
+ def affine(self, p1):
+ """Convert a Jacobian point tuple p1 to affine form, or None if at infinity.
+
+ An affine point is represented as the Jacobian (x, y, 1)"""
+ x1, y1, z1 = p1
+ if z1 == 0:
+ return None
+ inv = modinv(z1, self.p)
+ inv_2 = (inv**2) % self.p
+ inv_3 = (inv_2 * inv) % self.p
+ return ((inv_2 * x1) % self.p, (inv_3 * y1) % self.p, 1)
+
+ def negate(self, p1):
+ """Negate a Jacobian point tuple p1."""
+ x1, y1, z1 = p1
+ return (x1, (self.p - y1) % self.p, z1)
+
+ def on_curve(self, p1):
+ """Determine whether a Jacobian tuple p is on the curve (and not infinity)"""
+ x1, y1, z1 = p1
+ z2 = pow(z1, 2, self.p)
+ z4 = pow(z2, 2, self.p)
+ return z1 != 0 and (pow(x1, 3, self.p) + self.a * x1 * z4 + self.b * z2 * z4 - pow(y1, 2, self.p)) % self.p == 0
+
+ def is_x_coord(self, x):
+ """Test whether x is a valid X coordinate on the curve."""
+ x_3 = pow(x, 3, self.p)
+ return jacobi_symbol(x_3 + self.a * x + self.b, self.p) != -1
+
+ def lift_x(self, x):
+ """Given an X coordinate on the curve, return a corresponding affine point."""
+ x_3 = pow(x, 3, self.p)
+ v = x_3 + self.a * x + self.b
+ y = modsqrt(v, self.p)
+ if y is None:
+ return None
+ return (x, y, 1)
+
+ def double(self, p1):
+ """Double a Jacobian tuple p1
+
+ See https://en.wikibooks.org/wiki/Cryptography/Prime_Curve/Jacobian_Coordinates - Point Doubling"""
+ x1, y1, z1 = p1
+ if z1 == 0:
+ return (0, 1, 0)
+ y1_2 = (y1**2) % self.p
+ y1_4 = (y1_2**2) % self.p
+ x1_2 = (x1**2) % self.p
+ s = (4*x1*y1_2) % self.p
+ m = 3*x1_2
+ if self.a:
+ m += self.a * pow(z1, 4, self.p)
+ m = m % self.p
+ x2 = (m**2 - 2*s) % self.p
+ y2 = (m*(s - x2) - 8*y1_4) % self.p
+ z2 = (2*y1*z1) % self.p
+ return (x2, y2, z2)
+
+ def add_mixed(self, p1, p2):
+ """Add a Jacobian tuple p1 and an affine tuple p2
+
+ See https://en.wikibooks.org/wiki/Cryptography/Prime_Curve/Jacobian_Coordinates - Point Addition (with affine point)"""
+ x1, y1, z1 = p1
+ x2, y2, z2 = p2
+ assert(z2 == 1)
+ # Adding to the point at infinity is a no-op
+ if z1 == 0:
+ return p2
+ z1_2 = (z1**2) % self.p
+ z1_3 = (z1_2 * z1) % self.p
+ u2 = (x2 * z1_2) % self.p
+ s2 = (y2 * z1_3) % self.p
+ if x1 == u2:
+ if (y1 != s2):
+ # p1 and p2 are inverses. Return the point at infinity.
+ return (0, 1, 0)
+ # p1 == p2. The formulas below fail when the two points are equal.
+ return self.double(p1)
+ h = u2 - x1
+ r = s2 - y1
+ h_2 = (h**2) % self.p
+ h_3 = (h_2 * h) % self.p
+ u1_h_2 = (x1 * h_2) % self.p
+ x3 = (r**2 - h_3 - 2*u1_h_2) % self.p
+ y3 = (r*(u1_h_2 - x3) - y1*h_3) % self.p
+ z3 = (h*z1) % self.p
+ return (x3, y3, z3)
+
+ def add(self, p1, p2):
+ """Add two Jacobian tuples p1 and p2
+
+ See https://en.wikibooks.org/wiki/Cryptography/Prime_Curve/Jacobian_Coordinates - Point Addition"""
+ x1, y1, z1 = p1
+ x2, y2, z2 = p2
+ # Adding the point at infinity is a no-op
+ if z1 == 0:
+ return p2
+ if z2 == 0:
+ return p1
+ # Adding an Affine to a Jacobian is more efficient since we save field multiplications and squarings when z = 1
+ if z1 == 1:
+ return self.add_mixed(p2, p1)
+ if z2 == 1:
+ return self.add_mixed(p1, p2)
+ z1_2 = (z1**2) % self.p
+ z1_3 = (z1_2 * z1) % self.p
+ z2_2 = (z2**2) % self.p
+ z2_3 = (z2_2 * z2) % self.p
+ u1 = (x1 * z2_2) % self.p
+ u2 = (x2 * z1_2) % self.p
+ s1 = (y1 * z2_3) % self.p
+ s2 = (y2 * z1_3) % self.p
+ if u1 == u2:
+ if (s1 != s2):
+ # p1 and p2 are inverses. Return the point at infinity.
+ return (0, 1, 0)
+ # p1 == p2. The formulas below fail when the two points are equal.
+ return self.double(p1)
+ h = u2 - u1
+ r = s2 - s1
+ h_2 = (h**2) % self.p
+ h_3 = (h_2 * h) % self.p
+ u1_h_2 = (u1 * h_2) % self.p
+ x3 = (r**2 - h_3 - 2*u1_h_2) % self.p
+ y3 = (r*(u1_h_2 - x3) - s1*h_3) % self.p
+ z3 = (h*z1*z2) % self.p
+ return (x3, y3, z3)
+
+ def mul(self, ps):
+ """Compute a (multi) point multiplication
+
+ ps is a list of (Jacobian tuple, scalar) pairs.
+ """
+ r = (0, 1, 0)
+ for i in range(255, -1, -1):
+ r = self.double(r)
+ for (p, n) in ps:
+ if ((n >> i) & 1):
+ r = self.add(r, p)
+ return r
+
+SECP256K1 = EllipticCurve(2**256 - 2**32 - 977, 0, 7)
+SECP256K1_G = (0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, 0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8, 1)
SECP256K1_ORDER = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
SECP256K1_ORDER_HALF = SECP256K1_ORDER // 2
-# Thx to Sam Devlin for the ctypes magic 64-bit fix.
-def _check_result(val, func, args):
- if val == 0:
- raise ValueError
- else:
- return ctypes.c_void_p (val)
-
-ssl.EC_KEY_new_by_curve_name.restype = ctypes.c_void_p
-ssl.EC_KEY_new_by_curve_name.errcheck = _check_result
-
-class CECKey():
- """Wrapper around OpenSSL's EC_KEY"""
-
- POINT_CONVERSION_COMPRESSED = 2
- POINT_CONVERSION_UNCOMPRESSED = 4
+class ECPubKey():
+ """A secp256k1 public key"""
def __init__(self):
- self.k = ssl.EC_KEY_new_by_curve_name(NID_secp256k1)
-
- def __del__(self):
- if ssl:
- ssl.EC_KEY_free(self.k)
- self.k = None
-
- def set_secretbytes(self, secret):
- priv_key = ssl.BN_bin2bn(secret, 32, ssl.BN_new())
- group = ssl.EC_KEY_get0_group(self.k)
- pub_key = ssl.EC_POINT_new(group)
- ctx = ssl.BN_CTX_new()
- if not ssl.EC_POINT_mul(group, pub_key, priv_key, None, None, ctx):
- raise ValueError("Could not derive public key from the supplied secret.")
- ssl.EC_POINT_mul(group, pub_key, priv_key, None, None, ctx)
- ssl.EC_KEY_set_private_key(self.k, priv_key)
- ssl.EC_KEY_set_public_key(self.k, pub_key)
- ssl.EC_POINT_free(pub_key)
- ssl.BN_CTX_free(ctx)
- return self.k
-
- def set_privkey(self, key):
- self.mb = ctypes.create_string_buffer(key)
- return ssl.d2i_ECPrivateKey(ctypes.byref(self.k), ctypes.byref(ctypes.pointer(self.mb)), len(key))
-
- def set_pubkey(self, key):
- self.mb = ctypes.create_string_buffer(key)
- return ssl.o2i_ECPublicKey(ctypes.byref(self.k), ctypes.byref(ctypes.pointer(self.mb)), len(key))
-
- def get_privkey(self):
- size = ssl.i2d_ECPrivateKey(self.k, 0)
- mb_pri = ctypes.create_string_buffer(size)
- ssl.i2d_ECPrivateKey(self.k, ctypes.byref(ctypes.pointer(mb_pri)))
- return mb_pri.raw
-
- def get_pubkey(self):
- size = ssl.i2o_ECPublicKey(self.k, 0)
- mb = ctypes.create_string_buffer(size)
- ssl.i2o_ECPublicKey(self.k, ctypes.byref(ctypes.pointer(mb)))
- return mb.raw
-
- def get_raw_ecdh_key(self, other_pubkey):
- ecdh_keybuffer = ctypes.create_string_buffer(32)
- r = ssl.ECDH_compute_key(ctypes.pointer(ecdh_keybuffer), 32,
- ssl.EC_KEY_get0_public_key(other_pubkey.k),
- self.k, 0)
- if r != 32:
- raise Exception('CKey.get_ecdh_key(): ECDH_compute_key() failed')
- return ecdh_keybuffer.raw
-
- def get_ecdh_key(self, other_pubkey, kdf=lambda k: hashlib.sha256(k).digest()):
- # FIXME: be warned it's not clear what the kdf should be as a default
- r = self.get_raw_ecdh_key(other_pubkey)
- return kdf(r)
-
- def sign(self, hash, low_s = True):
- # FIXME: need unit tests for below cases
- if not isinstance(hash, bytes):
- raise TypeError('Hash must be bytes instance; got %r' % hash.__class__)
- if len(hash) != 32:
- raise ValueError('Hash must be exactly 32 bytes long')
-
- sig_size0 = ctypes.c_uint32()
- sig_size0.value = ssl.ECDSA_size(self.k)
- mb_sig = ctypes.create_string_buffer(sig_size0.value)
- result = ssl.ECDSA_sign(0, hash, len(hash), mb_sig, ctypes.byref(sig_size0), self.k)
- assert 1 == result
- assert mb_sig.raw[0] == 0x30
- assert mb_sig.raw[1] == sig_size0.value - 2
- total_size = mb_sig.raw[1]
- assert mb_sig.raw[2] == 2
- r_size = mb_sig.raw[3]
- assert mb_sig.raw[4 + r_size] == 2
- s_size = mb_sig.raw[5 + r_size]
- s_value = int.from_bytes(mb_sig.raw[6+r_size:6+r_size+s_size], byteorder='big')
- if (not low_s) or s_value <= SECP256K1_ORDER_HALF:
- return mb_sig.raw[:sig_size0.value]
- else:
- low_s_value = SECP256K1_ORDER - s_value
- low_s_bytes = (low_s_value).to_bytes(33, byteorder='big')
- while len(low_s_bytes) > 1 and low_s_bytes[0] == 0 and low_s_bytes[1] < 0x80:
- low_s_bytes = low_s_bytes[1:]
- new_s_size = len(low_s_bytes)
- new_total_size_byte = (total_size + new_s_size - s_size).to_bytes(1,byteorder='big')
- new_s_size_byte = (new_s_size).to_bytes(1,byteorder='big')
- return b'\x30' + new_total_size_byte + mb_sig.raw[2:5+r_size] + new_s_size_byte + low_s_bytes
-
- def verify(self, hash, sig):
- """Verify a DER signature"""
- return ssl.ECDSA_verify(0, hash, len(hash), sig, len(sig), self.k) == 1
-
- def set_compressed(self, compressed):
- if compressed:
- form = self.POINT_CONVERSION_COMPRESSED
+ """Construct an uninitialized public key"""
+ self.valid = False
+
+ def set(self, data):
+ """Construct a public key from a serialization in compressed or uncompressed format"""
+ if (len(data) == 65 and data[0] == 0x04):
+ p = (int.from_bytes(data[1:33], 'big'), int.from_bytes(data[33:65], 'big'), 1)
+ self.valid = SECP256K1.on_curve(p)
+ if self.valid:
+ self.p = p
+ self.compressed = False
+ elif (len(data) == 33 and (data[0] == 0x02 or data[0] == 0x03)):
+ x = int.from_bytes(data[1:33], 'big')
+ if SECP256K1.is_x_coord(x):
+ p = SECP256K1.lift_x(x)
+ # if the oddness of the y co-ord isn't correct, find the other
+ # valid y
+ if (p[1] & 1) != (data[0] & 1):
+ p = SECP256K1.negate(p)
+ self.p = p
+ self.valid = True
+ self.compressed = True
+ else:
+ self.valid = False
else:
- form = self.POINT_CONVERSION_UNCOMPRESSED
- ssl.EC_KEY_set_conv_form(self.k, form)
-
+ self.valid = False
-class CPubKey(bytes):
- """An encapsulated public key
-
- Attributes:
+ @property
+ def is_compressed(self):
+ return self.compressed
- is_valid - Corresponds to CPubKey.IsValid()
- is_fullyvalid - Corresponds to CPubKey.IsFullyValid()
- is_compressed - Corresponds to CPubKey.IsCompressed()
- """
+ @property
+ def is_valid(self):
+ return self.valid
+
+ def get_bytes(self):
+ assert(self.valid)
+ p = SECP256K1.affine(self.p)
+ if p is None:
+ return None
+ if self.compressed:
+ return bytes([0x02 + (p[1] & 1)]) + p[0].to_bytes(32, 'big')
+ else:
+ return bytes([0x04]) + p[0].to_bytes(32, 'big') + p[1].to_bytes(32, 'big')
+
+ def verify_ecdsa(self, sig, msg, low_s=True):
+ """Verify a strictly DER-encoded ECDSA signature against this pubkey.
+
+ See https://en.wikipedia.org/wiki/Elliptic_Curve_Digital_Signature_Algorithm for the
+ ECDSA verifier algorithm"""
+ assert(self.valid)
+
+ # Extract r and s from the DER formatted signature. Return false for
+ # any DER encoding errors.
+ if (sig[1] + 2 != len(sig)):
+ return False
+ if (len(sig) < 4):
+ return False
+ if (sig[0] != 0x30):
+ return False
+ if (sig[2] != 0x02):
+ return False
+ rlen = sig[3]
+ if (len(sig) < 6 + rlen):
+ return False
+ if rlen < 1 or rlen > 33:
+ return False
+ if sig[4] >= 0x80:
+ return False
+ if (rlen > 1 and (sig[4] == 0) and not (sig[5] & 0x80)):
+ return False
+ r = int.from_bytes(sig[4:4+rlen], 'big')
+ if (sig[4+rlen] != 0x02):
+ return False
+ slen = sig[5+rlen]
+ if slen < 1 or slen > 33:
+ return False
+ if (len(sig) != 6 + rlen + slen):
+ return False
+ if sig[6+rlen] >= 0x80:
+ return False
+ if (slen > 1 and (sig[6+rlen] == 0) and not (sig[7+rlen] & 0x80)):
+ return False
+ s = int.from_bytes(sig[6+rlen:6+rlen+slen], 'big')
+
+ # Verify that r and s are within the group order
+ if r < 1 or s < 1 or r >= SECP256K1_ORDER or s >= SECP256K1_ORDER:
+ return False
+ if low_s and s >= SECP256K1_ORDER_HALF:
+ return False
+ z = int.from_bytes(msg, 'big')
+
+ # Run verifier algorithm on r, s
+ w = modinv(s, SECP256K1_ORDER)
+ u1 = z*w % SECP256K1_ORDER
+ u2 = r*w % SECP256K1_ORDER
+ R = SECP256K1.affine(SECP256K1.mul([(SECP256K1_G, u1), (self.p, u2)]))
+ if R is None or R[0] != r:
+ return False
+ return True
+
+class ECKey():
+ """A secp256k1 private key"""
- def __new__(cls, buf, _cec_key=None):
- self = super(CPubKey, cls).__new__(cls, buf)
- if _cec_key is None:
- _cec_key = CECKey()
- self._cec_key = _cec_key
- self.is_fullyvalid = _cec_key.set_pubkey(self) != 0
- return self
+ def __init__(self):
+ self.valid = False
+
+ def set(self, secret, compressed):
+ """Construct a private key object with given 32-byte secret and compressed flag."""
+ assert(len(secret) == 32)
+ secret = int.from_bytes(secret, 'big')
+ self.valid = (secret > 0 and secret < SECP256K1_ORDER)
+ if self.valid:
+ self.secret = secret
+ self.compressed = compressed
+
+ def generate(self, compressed=True):
+ """Generate a random private key (compressed or uncompressed)."""
+ self.set(random.randrange(1, SECP256K1_ORDER).to_bytes(32, 'big'), compressed)
+
+ def get_bytes(self):
+ """Retrieve the 32-byte representation of this key."""
+ assert(self.valid)
+ return self.secret.to_bytes(32, 'big')
@property
def is_valid(self):
- return len(self) > 0
+ return self.valid
@property
def is_compressed(self):
- return len(self) == 33
-
- def verify(self, hash, sig):
- return self._cec_key.verify(hash, sig)
-
- def __str__(self):
- return repr(self)
-
- def __repr__(self):
- return '%s(%s)' % (self.__class__.__name__, super(CPubKey, self).__repr__())
+ return self.compressed
+ def get_pubkey(self):
+ """Compute an ECPubKey object for this secret key."""
+ assert(self.valid)
+ ret = ECPubKey()
+ p = SECP256K1.mul([(SECP256K1_G, self.secret)])
+ ret.p = p
+ ret.valid = True
+ ret.compressed = self.compressed
+ return ret
+
+ def sign_ecdsa(self, msg, low_s=True):
+ """Construct a DER-encoded ECDSA signature with this key.
+
+ See https://en.wikipedia.org/wiki/Elliptic_Curve_Digital_Signature_Algorithm for the
+ ECDSA signer algorithm."""
+ assert(self.valid)
+ z = int.from_bytes(msg, 'big')
+ # Note: no RFC6979, but a simple random nonce (some tests rely on distinct transactions for the same operation)
+ k = random.randrange(1, SECP256K1_ORDER)
+ R = SECP256K1.affine(SECP256K1.mul([(SECP256K1_G, k)]))
+ r = R[0] % SECP256K1_ORDER
+ s = (modinv(k, SECP256K1_ORDER) * (z + self.secret * r)) % SECP256K1_ORDER
+ if low_s and s > SECP256K1_ORDER_HALF:
+ s = SECP256K1_ORDER - s
+ # Represent in DER format. The byte representations of r and s have
+ # length rounded up (255 bits becomes 32 bytes and 256 bits becomes 33
+ # bytes).
+ rb = r.to_bytes((r.bit_length() + 8) // 8, 'big')
+ sb = s.to_bytes((s.bit_length() + 8) // 8, 'big')
+ return b'\x30' + bytes([4 + len(rb) + len(sb), 2, len(rb)]) + rb + bytes([2, len(sb)]) + sb
diff --git a/test/functional/test_framework/mininode.py b/test/functional/test_framework/mininode.py
index ac7cc068bd..6afdd50176 100755
--- a/test/functional/test_framework/mininode.py
+++ b/test/functional/test_framework/mininode.py
@@ -364,6 +364,14 @@ class P2PInterface(P2PConnection):
# Message receiving helper methods
+ def wait_for_tx(self, txid, timeout=60):
+ def test_function():
+ if not self.last_message.get('tx'):
+ return False
+ return self.last_message['tx'].tx.rehash() == txid
+
+ wait_until(test_function, timeout=timeout, lock=mininode_lock)
+
def wait_for_block(self, blockhash, timeout=60):
test_function = lambda: self.last_message.get("block") and self.last_message["block"].block.rehash() == blockhash
wait_until(test_function, timeout=timeout, lock=mininode_lock)
diff --git a/test/functional/test_framework/util.py b/test/functional/test_framework/util.py
index fef9982412..87e2dbaf16 100644
--- a/test/functional/test_framework/util.py
+++ b/test/functional/test_framework/util.py
@@ -219,7 +219,7 @@ def wait_until(predicate, *, attempts=float('inf'), timeout=float('inf'), lock=N
time.sleep(0.05)
# Print the cause of the timeout
- predicate_source = inspect.getsourcelines(predicate)
+ predicate_source = "''''\n" + inspect.getsource(predicate) + "'''"
logger.error("wait_until() failed. Predicate: {}".format(predicate_source))
if attempt >= attempts:
raise AssertionError("Predicate {} not true after {} attempts".format(predicate_source, attempts))
diff --git a/test/functional/test_runner.py b/test/functional/test_runner.py
index bb9ac58e46..06d3552195 100755
--- a/test/functional/test_runner.py
+++ b/test/functional/test_runner.py
@@ -134,6 +134,7 @@ BASE_SCRIPTS = [
'rpc_net.py',
'wallet_keypool.py',
'p2p_mempool.py',
+ 'p2p_blocksonly.py',
'mining_prioritisetransaction.py',
'p2p_invalid_locator.py',
'p2p_invalid_block.py',
generated by cgit v1.2.3 (git 2.26.2) at 2024-07-08 09:18:02 +0000