Inside Bitcoin

1251    # == sighash caching tests ==
1252
1253    # Sighash caching in legacy.
1254    for p2sh in [False, True]:
1255        for witv0 in [False, True]:
1256            eckey1, pubkey1 = generate_keypair(compressed=compressed)
1257            for _ in range(10):
1258                # Construct a script with 20 checksig operations (10 sighash types, each 2 times),
1259                # randomly ordered and interleaved with 4 OP_CODESEPARATORS.
1260                ops = [1, 2, 3, 0x21, 0x42, 0x63, 0x81, 0x83, 0xe1, 0xc2, -1, -1] * 2
1261                # Make sure no OP_CODESEPARATOR appears last.
1262                while True:
1263                    random.shuffle(ops)
1264                    if ops[-1] != -1:
1265                        break
1266                script = [pubkey1]
1267                inputs = []
1268                codeseps = -1
1269                for pos, op in enumerate(ops):
1270                    if op == -1:
1271                        codeseps += 1
1272                        script.append(OP_CODESEPARATOR)
1273                    elif pos + 1 != len(ops):
1274                        script += [OP_TUCK, OP_CHECKSIGVERIFY]
1275                        inputs.append(getter("sign", codesepnum=codeseps, hashtype=op))
1276                    else:
1277                        script += [OP_CHECKSIG]
1278                        inputs.append(getter("sign", codesepnum=codeseps, hashtype=op))
1279                inputs.reverse()
1280                script = CScript(script)
1281                add_spender(spenders, "sighashcache/legacy", p2sh=p2sh, witv0=witv0, standard=False, script=script, inputs=inputs, key=eckey1, sigops_weight=12*8*(4-3*witv0), no_fail=True)
1282
1283    # Sighash caching in tapscript.
1284    for _ in range(10):
1285        # Construct a script with 700 checksig operations (7 sighash types, each 100 times),
1286        # randomly ordered and interleaved with 100 OP_CODESEPARATORS.
1287        ops = [0, 1, 2, 3, 0x81, 0x82, 0x83, -1] * 100
1288        # Make sure no OP_CODESEPARATOR appears last.
1289        while True:
1290            random.shuffle(ops)
1291            if ops[-1] != -1:
1292                 break
1293        script = [pubs[1]]
1294        inputs = []
1295        opcount = 1
1296        codeseppos = 0xffffffff
1297        for pos, op in enumerate(ops):
1298            if op == -1:
1299                codeseppos = opcount
1300                opcount += 1
1301                script.append(OP_CODESEPARATOR)
1302            elif pos + 1 != len(ops):
1303                opcount += 2
1304                script += [OP_TUCK, OP_CHECKSIGVERIFY]
1305                inputs.append(getter("sign", codeseppos=codeseppos, hashtype=op))
1306            else:
1307                opcount += 1
1308                script += [OP_CHECKSIG]
1309                inputs.append(getter("sign", codeseppos=codeseppos, hashtype=op))
1310        inputs.reverse()
1311        script = CScript(script)
1312        tap = taproot_construct(pubs[0], [("leaf", script)])
1313        add_spender(spenders, "sighashcache/taproot", tap=tap, leaf="leaf", inputs=inputs, standard=True, key=secs[1], no_fail=True)
1314
1315    return spenders
1316
1317
1318def spenders_taproot_nonstandard():
1319    """Spenders for testing that post-activation Taproot rules may be nonstandard."""
1320
1321    spenders = []
1322
1323    sec = generate_privkey()
1324    pub, _ = compute_xonly_pubkey(sec)
1325    scripts = [
1326        ("future_leaf", CScript([pub, OP_CHECKSIG]), 0xc2),
1327        ("op_success", CScript([pub, OP_CHECKSIG, OP_0, OP_IF, CScriptOp(0x50), OP_ENDIF])),
1328    ]
1329    tap = taproot_construct(pub, scripts)
1330
1331    # Test that features like annex, leaf versions, or OP_SUCCESS are valid but non-standard
1332    add_spender(spenders, "inactive/scriptpath_valid_unkleaf", key=sec, tap=tap, leaf="future_leaf", standard=False, inputs=[getter("sign")])
1333    add_spender(spenders, "inactive/scriptpath_invalid_unkleaf", key=sec, tap=tap, leaf="future_leaf", standard=False, inputs=[getter("sign")], sighash=bitflipper(default_sighash))
1334    add_spender(spenders, "inactive/scriptpath_valid_opsuccess", key=sec, tap=tap, leaf="op_success", standard=False, inputs=[getter("sign")])
1335    add_spender(spenders, "inactive/scriptpath_valid_opsuccess", key=sec, tap=tap, leaf="op_success", standard=False, inputs=[getter("sign")], sighash=bitflipper(default_sighash))
1336
1337    return spenders
1338
1339def sample_spenders():
1340
1341    # Create key(s) for output creation, as well as key and script-spends
1342    secs = [generate_privkey() for _ in range(2)]
1343    pubs = [compute_xonly_pubkey(sec)[0] for sec in secs]
1344
1345    # Create a list of scripts which will be built into a taptree
1346    scripts = [
1347        # leaf label, followed by CScript
1348        ("2byte_push", CScript([OP_DROP, b'\xaa\xaa'])),
1349        ("nonstd_2byte_push", CScript.fromhex("4c02aaaa")),
1350        ("dummyleaf", CScript([])),
1351    ]
1352
1353    # Build TaprootInfo using scripts and appropriate pubkey for output creation
1354    tap = taproot_construct(pubs[0], scripts)
1355
1356    # Finally, add spender(s).
1357    # Each spender embodies a test with an optional failure condition.
1358    # These failure conditions allow for fine-grained success/failure
1359    # conditions that are tested randomly.
1360    spenders = []
1361
1362    # Named comment, using first leaf from scripts, with empty string as witness data, no optional fail condition
1363    add_spender(spenders, comment="tutorial/push", tap=tap, leaf="2byte_push", inputs=[b'\x00'], no_fail=True)
1364
1365    # Spender with alternative failure tapscript via over-riding "failure" dictionary, along with the failure's expected err_msg / ERR_*
1366    add_spender(spenders, comment="tutorial/pushredux", tap=tap, leaf="2byte_push", inputs=[b'\x00'], failure={"leaf": "dummyleaf"}, **ERR_EVAL_FALSE)
1367
1368    # Spender that is non-standard but otherwise valid, with extraneous signature data from inner key for optional failure condition
1369    add_spender(spenders, comment="tutorial/nonminpush", tap=tap, leaf="nonstd_2byte_push", key=secs[0], standard=False, failure={"inputs": [getter("sign")]}, **ERR_CLEANSTACK)
1370
1371    # New scripts=[] can be defined, and rinse-repeated as necessary until the spenders list is returned for execution
1372    return spenders
1373
1374# Consensus validation flags to use in dumps for tests with "legacy/" or "inactive/" prefix.
1375LEGACY_FLAGS = "P2SH,DERSIG,CHECKLOCKTIMEVERIFY,CHECKSEQUENCEVERIFY,WITNESS,NULLDUMMY"
1376# Consensus validation flags to use in dumps for all other tests.
1377TAPROOT_FLAGS = "P2SH,DERSIG,CHECKLOCKTIMEVERIFY,CHECKSEQUENCEVERIFY,WITNESS,NULLDUMMY,TAPROOT"
1378
1379def dump_json_test(tx, input_utxos, idx, success, failure):
1380    spender = input_utxos[idx].spender
1381    # Determine flags to dump
1382    flags = LEGACY_FLAGS if spender.comment.startswith("legacy/") or spender.comment.startswith("inactive/") else TAPROOT_FLAGS
1383
1384    fields = [
1385        ("tx", tx.serialize().hex()),
1386        ("prevouts", [x.output.serialize().hex() for x in input_utxos]),
1387        ("index", idx),
1388        ("flags", flags),
1389        ("comment", spender.comment)
1390    ]
1391
1392    # The "final" field indicates that a spend should be always valid, even with more validation flags enabled
1393    # than the listed ones. Use standardness as a proxy for this (which gives a conservative underestimate).
1394    if spender.is_standard:
1395        fields.append(("final", True))
1396
1397    def dump_witness(wit):
1398        return OrderedDict([("scriptSig", wit[0].hex()), ("witness", [x.hex() for x in wit[1]])])
1399    if success is not None:
1400        fields.append(("success", dump_witness(success)))
1401    if failure is not None:
1402        fields.append(("failure", dump_witness(failure)))
1403
1404    # Write the dump to $TEST_DUMP_DIR/x/xyz... where x,y,z,... are the SHA1 sum of the dump (which makes the
1405    # file naming scheme compatible with fuzzing infrastructure).
1406    dump = json.dumps(OrderedDict(fields)) + ",\n"
1407    sha1 = hashlib.sha1(dump.encode("utf-8")).hexdigest()
1408    dirname = os.environ.get("TEST_DUMP_DIR", ".") + ("/%s" % sha1[0])
1409    os.makedirs(dirname, exist_ok=True)
1410    with open(dirname + ("/%s" % sha1), 'w') as f:
1411        f.write(dump)
1412
1413# Data type to keep track of UTXOs, where they were created, and how to spend them.
1414UTXOData = namedtuple('UTXOData', 'outpoint,output,spender')
1415
1416
1417class TaprootTest(BitcoinTestFramework):
1418    def add_options(self, parser):
1419        parser.add_argument("--dumptests", dest="dump_tests", default=False, action="store_true",
1420                            help="Dump generated test cases to directory set by TEST_DUMP_DIR environment variable")
1421
1422    def set_test_params(self):
1423        self.num_nodes = 1
1424        self.setup_clean_chain = True
1425
1426    def block_submit(self, node, txs, msg, err_msg, cb_pubkey=None, fees=0, sigops_weight=0, witness=False, accept=False):
1427
1428        # Deplete block of any non-tapscript sigops using a single additional 0-value coinbase output.
1429        # It is not impossible to fit enough tapscript sigops to hit the old 80k limit without
1430        # busting txin-level limits. We simply have to account for the p2pk outputs in all
1431        # transactions.
1432        extra_output_script = CScript(bytes([OP_CHECKSIG]*((MAX_BLOCK_SIGOPS_WEIGHT - sigops_weight) // WITNESS_SCALE_FACTOR)))
1433
1434        coinbase_tx = create_coinbase(self.lastblockheight + 1, pubkey=cb_pubkey, extra_output_script=extra_output_script, fees=fees)
1435        block = create_block(self.tip, coinbase_tx, ntime=self.lastblocktime + 1, txlist=txs)
1436        witness and add_witness_commitment(block)
1437        block.solve()
1438        block_response = node.submitblock(block.serialize().hex())
1439        if err_msg is not None:
1440            assert block_response is not None and err_msg in block_response, "Missing error message '%s' from block response '%s': %s" % (err_msg, "(None)" if block_response is None else block_response, msg)
1441        if accept:
1442            assert node.getbestblockhash() == block.hash_hex, "Failed to accept: %s (response: %s)" % (msg, block_response)
1443            self.tip = block.hash_int
1444            self.lastblockhash = block.hash_hex
1445            self.lastblocktime += 1
1446            self.lastblockheight += 1
1447        else:
1448            assert node.getbestblockhash() == self.lastblockhash, "Failed to reject: " + msg
1449
1450    def init_blockinfo(self, node):
1451        # Initialize variables used by block_submit().
1452        self.lastblockhash = node.getbestblockhash()
1453        self.tip = int(self.lastblockhash, 16)
1454        block = node.getblock(self.lastblockhash)
1455        self.lastblockheight = block['height']
1456        self.lastblocktime = block['time']
1457
1458    def test_spenders(self, node, spenders, input_counts):
1459        """Run randomized tests with a number of "spenders".
1460
1461        Steps:
1462            1) Generate an appropriate UTXO for each spender to test spend conditions
1463            2) Generate 100 random addresses of all wallet types: pkh/sh_wpkh/wpkh
1464            3) Select random number of inputs from (1)
1465            4) Select random number of addresses from (2) as outputs
1466
1467        Each spender embodies a test; in a large randomized test, it is verified
1468        that toggling the valid argument to each lambda toggles the validity of
1469        the transaction. This is accomplished by constructing transactions consisting
1470        of all valid inputs, except one invalid one.
1471        """
1472
1473        # Construct a bunch of sPKs that send coins back to the host wallet
1474        self.log.info("- Constructing addresses for returning coins")
1475        host_spks = []
1476        host_pubkeys = []
1477        for i in range(16):
1478            pubkey, spk, _ = self.nodesigner.getnewaddress(address_type=random.choice(["legacy", "p2sh-segwit", "bech32"]))
1479            host_spks.append(spk)
1480            host_pubkeys.append(pubkey)
1481
1482        self.init_blockinfo(node)
1483
1484        # Create transactions spending up to 50 of the wallet's inputs, with one output for each spender, and
1485        # one change output at the end. The transaction is constructed on the Python side to enable
1486        # having multiple outputs to the same address and outputs with no assigned address. The node
1487        # is then asked to sign it through signrawtransactionwithkey, and then added to a block on the
1488        # Python side (to bypass standardness rules).
1489        self.log.info("- Creating test UTXOs...")
1490        random.shuffle(spenders)
1491        normal_utxos = []
1492        mismatching_utxos = [] # UTXOs with input that requires mismatching output position
1493        done = 0
1494        while done < len(spenders):
1495            # Compute how many UTXOs to create with this transaction
1496            count_this_tx = min(len(spenders) - done, (len(spenders) + 4) // 5, 10000)
1497
1498            fund_tx = CTransaction()
1499            # Add the 50 highest-value inputs
1500            unspents = self.nodesigner.listunspent()