-
Notifications
You must be signed in to change notification settings - Fork 8
/
stub_invalid_block.py
100 lines (78 loc) · 3.22 KB
/
stub_invalid_block.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
#!/usr/bin/env python3
import socket
import time
from commander import Commander
# The entire Bitcoin Core test_framework directory is available as a library
from test_framework.messages import (
hash256,
msg_block,
msg_ping,
)
from test_framework.p2p import MAGIC_BYTES, P2PInterface
from test_framework.blocktools import create_block, create_coinbase
def get_signet_network_magic_from_node(node):
template = node.getblocktemplate({"rules": ["segwit", "signet"]})
challenge = template["signet_challenge"]
challenge_bytes = bytes.fromhex(challenge)
data = len(challenge_bytes).to_bytes() + challenge_bytes
digest = hash256(data)
return digest[0:4]
# The actual scenario is a class like a Bitcoin Core functional test.
# Commander is a subclass of BitcoinTestFramework instide Warnet
# that allows to operate on containerized nodes instead of local nodes.
class InvalidBlock(Commander):
def set_test_params(self):
# This setting is ignored but still required as
# a sub-class of BitcoinTestFramework
self.num_nodes = 1
def add_options(self, parser):
parser.description = "Send an invalid block to a node"
parser.usage = "warnet run stub_invalid_block.py --debug"
# Scenario entrypoint
def run_test(self):
def get_msg(message):
if message:
[print(f"Peer: {addr.ip}, Port: {addr.port}") for addr in message.addrs]
victim = "TARGET_TANK_NAME.default.svc"
# regtest or signet
chain = self.nodes[0].chain
# The victim's address could be an explicit IP address
# OR a kubernetes hostname (use default chain p2p port)
dstaddr = socket.gethostbyname(victim)
if chain == "regtest":
dstport = 18444
if chain == "signet":
dstport = 38333
MAGIC_BYTES["signet"] = get_signet_network_magic_from_node(self.nodes[0])
# Now we will use a python-based Bitcoin p2p node to send very specific,
# unusual or non-standard messages to a "victim" node.
self.log.info(f"Attacking {victim}")
attacker = P2PInterface()
attacker.peer_connect(
dstaddr=dstaddr, dstport=dstport, net="signet", timeout_factor=1
)()
attacker.wait_until(lambda: attacker.is_connected, check_connected=False)
best_block_hash = self.nodes[0].getbestblockhash()
best_block = self.nodes[0].getblock(best_block_hash)
best_block_time = best_block["time"]
coinbase = create_coinbase(height=123)
new_block = create_block(
hashprev=int(best_block_hash, 16),
coinbase=coinbase,
ntime=best_block_time + 1,
)
# new_block.solve()
new_block.hashMerkleRoot = 0xDEADBEEF
msg = msg_block(new_block)
# DEAR HACKERS: The invalid block msg has been made, send it now!
ping = msg_ping()
self.log.info(f"Ping: {attacker.ping_counter}")
attacker.send_message(ping)
time.sleep(3)
self.log.info(f"Ping: {attacker.ping_counter}")
attacker.wait_for_disconnect()
self.log.info("Disconnected or timed out")
def main():
InvalidBlock().main()
if __name__ == "__main__":
main()