fork_blukecoin_detect_test.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2026 The Limenka developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 """Blukecoin (bitcoin-blake2b) node detection and gossip.
6
7 A blukecoin node keeps the parent chain's P2P identity (magic, port,
8 genesis) but hashes block headers with blake2b, so a header it transmits
9 fails SHA256d proof of work while claiming a valid mainnet-range
10 difficulty. This test connects a crafted peer that sends exactly such a
11 header, verifies the node tags it (getpeerinfo blukecoin=true), records it
12 in the runtime database (getblukecoinpeers) and gossips the address to a
13 second node via the BLUKECOIN message.
14 """
15 import hashlib
16
17 from test_framework.address import byte_to_base58
18 from test_framework.descriptors import descsum_create
19 from test_framework.key import ECKey
20 from test_framework.messages import CBlockHeader, msg_headers
21 from test_framework.p2p import P2PInterface
22 from test_framework.test_framework import LimenkaTestFramework
23 from test_framework.util import p2p_port
24
25 ACTIVATION_FAR = 1999999999
26 GENESIS_TIME = 1231006505
27
28 # A header with the mainnet powLimit difficulty whose SHA256d hash does
29 # not meet the target - the blake2b-chain signature. Any random header at
30 # this difficulty fails SHA256d PoW with overwhelming probability.
31 def blake2b_style_header():
32 h = CBlockHeader()
33 h.nVersion = 0x20000000
34 h.nTime = 1500000000
35 h.nBits = 0x1d00ffff
36 h.nNonce = 1
37 return h
38
39
40 class BlukecoinInterface(P2PInterface):
41 def __init__(self):
42 super().__init__()
43 self.blukecoin_seen = False
44
45 def on_blukecoin(self, message):
46 self.blukecoin_seen = True
47
48
49 class ForkBlukecoinTest(LimenkaTestFramework):
50 def set_test_params(self):
51 self.setup_clean_chain = True
52 self.num_nodes = 2
53 self.chain = 'limenka'
54 self.extra_args = [[
55 '-forkactivationtime=%d' % ACTIVATION_FAR,
56 '-forkdelaysteps=1024',
57 '-forkmineondemand',
58 '-forkstandalone',
59 '-fallbackfee=0.0002',
60 '-addresstype=bech32',
61 ]] * self.num_nodes
62 self.rpc_timeout = 240
63
64 def skip_test_if_missing_module(self):
65 self.skip_if_no_wallet()
66
67 def add_options(self, parser):
68 self.add_wallet_options(parser, descriptors=True, legacy=False)
69
70 def init_wallet(self, *, node):
71 wallet_name = self.default_wallet_name if self.wallet_names is None else self.wallet_names[node] if node < len(self.wallet_names) else False
72 if wallet_name is not False:
73 n = self.nodes[node]
74 if wallet_name is not None and node == 0:
75 n.createwallet(wallet_name=wallet_name, descriptors=True, load_on_startup=True)
76 seed = hashlib.sha256(b'fork-blukecoin-funding').digest()
77 key = ECKey()
78 key.set(seed, compressed=True)
79 priv = key.get_bytes()
80 wif = byte_to_base58(priv + b'\x01', 128)
81 desc = descsum_create(f"wpkh({wif})")
82 import_res = n.importdescriptors([{
83 "desc": desc,
84 "timestamp": GENESIS_TIME + 600,
85 "active": False,
86 "internal": False,
87 }])
88 assert all(r['success'] for r in import_res)
89
90 def run_test(self):
91 node_a = self.nodes[0]
92 node_b = self.nodes[1]
93
94 # A peer that transmits blake2b-style headers (valid difficulty,
95 # SHA256d hash fails) connects to node A.
96 blukecoin_peer = node_a.add_p2p_connection(BlukecoinInterface())
97
98 # Send the offending header.
99 blukecoin_peer.send_message(msg_headers([blake2b_style_header()]))
100
101 # Node A must tag the peer as blukecoin and record it in the DB.
102 self.wait_until(lambda: any(
103 p.get('blukecoin') for p in node_a.getpeerinfo()))
104 self.wait_until(lambda: len(node_a.getblukecoinpeers()) >= 1)
105 detected = node_a.getblukecoinpeers()
106 self.log.info('Node A detected blukecoin peers: %s' % detected)
107 assert len(detected) >= 1
108 assert any(p.get('blukecoin') for p in node_a.getpeerinfo())
109
110 # Connect the second fork node so the detection can be gossiped.
111 node_a.addnode(f'127.0.0.1:{p2p_port(1)}', 'onetry')
112 self.wait_until(lambda: len(node_a.getblukecoinpeers()) >= 1)
113
114 # The address must be gossiped to node B via the BLUKECOIN message.
115 self.wait_until(lambda: len(node_b.getblukecoinpeers()) >= 1)
116 self.log.info('Node B learned blukecoin peers: %s' % node_b.getblukecoinpeers())
117 assert len(node_b.getblukecoinpeers()) >= 1
118
119
120 if __name__ == '__main__':
121 ForkBlukecoinTest(__file__).main()
122