Coverage for tests/ledger/hsm2dongle_cmds/test_hsm2dongle_advance_blockchain.py: 100%
78 statements
« prev ^ index » next coverage.py v7.5.3, created at 2025-07-10 13:43 +0000
1# The MIT License (MIT)
2#
3# Copyright (c) 2021 RSK Labs Ltd
4#
5# Permission is hereby granted, free of charge, to any person obtaining a copy of
6# this software and associated documentation files (the "Software"), to deal in
7# the Software without restriction, including without limitation the rights to
8# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
9# of the Software, and to permit persons to whom the Software is furnished to do
10# so, subject to the following conditions:
11#
12# The above copyright notice and this permission notice shall be included in all
13# copies or substantial portions of the Software.
14#
15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21# SOFTWARE.
23from unittest.mock import patch, call
24from parameterized import parameterized
25from ..test_hsm2dongle import TestHSM2DongleBase
26from ledgerblue.commException import CommException
28import logging
30logging.disable(logging.CRITICAL)
33class TestHSM2DongleAdvanceBlockchain(TestHSM2DongleBase):
34 def setup_mocks(self,
35 mmplsize_mock,
36 get_cb_txn_mock,
37 cb_txn_get_hash_mock,
38 gbh_mock):
39 mmplsize_mock.side_effect = lambda h: len(h)//8
40 get_cb_txn_mock.side_effect = lambda h: {"cb_txn": h}
41 cb_txn_get_hash_mock.side_effect = lambda h: \
42 (bytes([len(h["cb_txn"])//5])*4).hex()
43 gbh_mock.return_value = "00"
45 @parameterized.expand([
46 ("partial_v2.0.x", 0x05, 2),
47 ("total_v2.0.x", 0x06, 1),
48 ("partial_v2.1.x", 0x05, 2),
49 ("total_v2.1.x", 0x06, 1),
50 ])
51 @patch("ledger.hsm2dongle.get_block_hash")
52 @patch("ledger.hsm2dongle.coinbase_tx_get_hash")
53 @patch("ledger.hsm2dongle.get_coinbase_txn")
54 @patch("ledger.hsm2dongle.rlp_mm_payload_size")
55 def test_advance_blockchain_ok(
56 self,
57 _,
58 device_response,
59 expected_response,
60 mmplsize_mock,
61 get_cb_txn_mock,
62 cb_txn_get_hash_mock,
63 gbh_mock,
64 ):
65 self.setup_mocks(mmplsize_mock,
66 get_cb_txn_mock,
67 cb_txn_get_hash_mock,
68 gbh_mock)
69 brothers_spec = [
70 # (brother list of brother bytes, chunk size)
71 ([self.buf(190), self.buf(100)], 90),
72 None, # 2nd block has no brothers
73 ([self.buf(130)], 60),
74 ]
75 blocks_spec = [
76 # (block bytes, chunk size, brothers)
77 (self.buf(300), 80, brothers_spec[0]),
78 (self.buf(250), 100, brothers_spec[1]),
79 (self.buf(140), 50, brothers_spec[2]),
80 ]
82 self.dongle.exchange.side_effect = [
83 bs for excs in map(self.spec_to_exchange, blocks_spec)
84 for bs in excs
85 ] + [bytes([0, 0, device_response])] # Success response
87 blocks_hex = list(map(lambda bs: bs[0].hex(), blocks_spec))
88 brothers_list = list(map(
89 lambda bs: list(map(
90 lambda b: b.hex(), bs[0])) if bs else [],
91 brothers_spec))
92 self.assertEqual(
93 (True, expected_response),
94 self.hsm2dongle.advance_blockchain(blocks_hex, brothers_list),
95 )
97 self.assert_exchange([
98 [0x10, 0x02, 0x00, 0x00, 0x00, 0x03], # Init, 3 blocks
99 [0x10, 0x03, 0x00, 0x4B] +
100 [0x78, 0x78, 0x78, 0x78], # Blk #1 meta
101 [0x10, 0x04] + list(blocks_spec[0][0][80*0:80*1]), # Blk #1 chunk
102 [0x10, 0x04] + list(blocks_spec[0][0][80*1:80*2]), # Blk #1 chunk
103 [0x10, 0x04] + list(blocks_spec[0][0][80*2:80*3]), # Blk #1 chunk
104 [0x10, 0x04] + list(blocks_spec[0][0][80*3:80*4]), # Blk #1 chunk
105 [0x10, 0x07, 0x02], # Blk #1 brother count
106 [0x10, 0x08, 0x00, 0x2f, 0x4c, 0x4c, 0x4c, 0x4c], # Blk #1 bro #1 meta
107 [0x10, 0x09] + list(brothers_spec[0][0][0][90*0:90*1]), # Blk #1 bro #1 chunk
108 [0x10, 0x09] + list(brothers_spec[0][0][0][90*1:90*2]), # Blk #1 bro #1 chunk
109 [0x10, 0x09] + list(brothers_spec[0][0][0][90*2:90*3]), # Blk #1 bro #1 chunk
110 [0x10, 0x08, 0x00, 0x19, 0x28, 0x28, 0x28, 0x28], # Blk #1 bro #2 meta
111 [0x10, 0x09] + list(brothers_spec[0][0][1][90*0:90*1]), # Blk #1 bro #2 chunk
112 [0x10, 0x09] + list(brothers_spec[0][0][1][90*1:90*2]), # Blk #1 bro #2 chunk
113 [0x10, 0x03, 0x00, 0x3E] +
114 [0x64, 0x64, 0x64, 0x64], # Blk #2 meta
115 [0x10, 0x04] + list(blocks_spec[1][0][100*0:100*1]), # Blk #2 chunk
116 [0x10, 0x04] + list(blocks_spec[1][0][100*1:100*2]), # Blk #2 chunk
117 [0x10, 0x04] + list(blocks_spec[1][0][100*2:100*3]), # Blk #2 chunk
118 [0x10, 0x07, 0x00], # Blk #2 brother count
119 [0x10, 0x03, 0x00, 0x23] +
120 [0x38, 0x38, 0x38, 0x38], # Blk #3 meta
121 [0x10, 0x04] + list(blocks_spec[2][0][50*0:50*1]), # Blk #3 chunk
122 [0x10, 0x04] + list(blocks_spec[2][0][50*1:50*2]), # Blk #3 chunk
123 [0x10, 0x04] + list(blocks_spec[2][0][50*2:50*3]), # Blk #3 chunk
124 [0x10, 0x07, 0x01], # Blk #3 brother count
125 [0x10, 0x08, 0x00, 0x20, 0x34, 0x34, 0x34, 0x34], # Blk #3 bro #1 meta
126 [0x10, 0x09] + list(brothers_spec[2][0][0][60*0:60*1]), # Blk #3 bro #1 chunk
127 [0x10, 0x09] + list(brothers_spec[2][0][0][60*1:60*2]), # Blk #3 bro #1 chunk
128 [0x10, 0x09] + list(brothers_spec[2][0][0][60*2:60*3]), # Blk #3 bro #1 chunk
129 ])
131 @parameterized.expand(TestHSM2DongleBase.CHUNK_ERROR_MAPPINGS)
132 @patch("ledger.hsm2dongle.get_block_hash")
133 @patch("ledger.hsm2dongle.coinbase_tx_get_hash")
134 @patch("ledger.hsm2dongle.get_coinbase_txn")
135 @patch("ledger.hsm2dongle.rlp_mm_payload_size")
136 def test_advance_blockchain_chunk_error_result(
137 self,
138 _,
139 error_code,
140 response,
141 mmplsize_mock,
142 get_cb_txn_mock,
143 cb_txn_get_hash_mock,
144 gbh_mock,
145 ):
146 self.setup_mocks(mmplsize_mock,
147 get_cb_txn_mock,
148 cb_txn_get_hash_mock,
149 gbh_mock)
150 brothers_spec = [
151 # (brother list of brother bytes, chunk size)
152 ([self.buf(190), self.buf(100)], 90),
153 None, # 2nd block has no brothers
154 ([self.buf(130)], 60),
155 ]
156 blocks_spec = [
157 # (block bytes, chunk size, brothers)
158 (self.buf(300), 80, brothers_spec[0]),
159 (self.buf(250), 100, brothers_spec[1]),
160 (self.buf(140), 50, brothers_spec[2]),
161 ]
163 side_effect = [
164 bs for excs in map(self.spec_to_exchange, blocks_spec)
165 for bs in excs
166 ]
168 # Make the second chunk of the second block fail
169 # First block meta & chunks & bro metas & chunks
170 # + second block meta & first & second chunk
171 exchange_index = (
172 (1 + 300//80 + 1) + 1 + (1 + 190//90 + 1) + (1 + 100//90 + 1) + 3
173 )
175 if type(error_code) == bytes:
176 side_effect[exchange_index] = error_code
177 else:
178 side_effect[exchange_index] = CommException("a-message", error_code)
179 side_effect = side_effect[:exchange_index + 1]
180 self.dongle.exchange.side_effect = side_effect
182 blocks_hex = list(map(lambda bs: bs[0].hex(), blocks_spec))
183 brothers_list = list(map(
184 lambda bs: list(map(
185 lambda b: b.hex(), bs[0])) if bs else [],
186 brothers_spec))
188 self.assertEqual(
189 (False, response),
190 self.hsm2dongle.advance_blockchain(blocks_hex, brothers_list),
191 )
193 self.assert_exchange([
194 [0x10, 0x02, 0x00, 0x00, 0x00, 0x03], # Init, 3 blocks
195 [0x10, 0x03, 0x00, 0x4B] +
196 [0x78, 0x78, 0x78, 0x78], # Blk #1 meta
197 [0x10, 0x04] + list(blocks_spec[0][0][80*0:80*1]), # Blk #1 chunk
198 [0x10, 0x04] + list(blocks_spec[0][0][80*1:80*2]), # Blk #1 chunk
199 [0x10, 0x04] + list(blocks_spec[0][0][80*2:80*3]), # Blk #1 chunk
200 [0x10, 0x04] + list(blocks_spec[0][0][80*3:80*4]), # Blk #1 chunk
201 [0x10, 0x07, 0x02], # Blk #1 brother count
202 [0x10, 0x08, 0x00, 0x2f, 0x4c, 0x4c, 0x4c, 0x4c], # Blk #1 bro #1 meta
203 [0x10, 0x09] + list(brothers_spec[0][0][0][90*0:90*1]), # Blk #1 bro #1 chunk
204 [0x10, 0x09] + list(brothers_spec[0][0][0][90*1:90*2]), # Blk #1 bro #1 chunk
205 [0x10, 0x09] + list(brothers_spec[0][0][0][90*2:90*3]), # Blk #1 bro #1 chunk
206 [0x10, 0x08, 0x00, 0x19, 0x28, 0x28, 0x28, 0x28], # Blk #1 bro #2 meta
207 [0x10, 0x09] + list(brothers_spec[0][0][1][90*0:90*1]), # Blk #1 bro #2 chunk
208 [0x10, 0x09] + list(brothers_spec[0][0][1][90*1:90*2]), # Blk #1 bro #2 chunk
209 [0x10, 0x03, 0x00, 0x3E] +
210 [0x64, 0x64, 0x64, 0x64], # Blk #2 meta
211 [0x10, 0x04] + list(blocks_spec[1][0][100*0:100*1]), # Blk #2 chunk
212 [0x10, 0x04] + list(blocks_spec[1][0][100*1:100*2]), # Blk #2 chunk
213 ])
215 @parameterized.expand([
216 ("prot_invalid", 0x6B87, -3),
217 ("unexpected", 0x6BFF, -10),
218 ("error_response", bytes([0, 0, 0xFF]), -10),
219 ])
220 @patch("ledger.hsm2dongle.get_block_hash")
221 @patch("ledger.hsm2dongle.coinbase_tx_get_hash")
222 @patch("ledger.hsm2dongle.get_coinbase_txn")
223 @patch("ledger.hsm2dongle.rlp_mm_payload_size")
224 def test_advance_blockchain_metadata_error_result(
225 self,
226 _,
227 error_code,
228 response,
229 mmplsize_mock,
230 get_cb_txn_mock,
231 cb_txn_get_hash_mock,
232 gbh_mock,
233 ):
234 self.setup_mocks(mmplsize_mock,
235 get_cb_txn_mock,
236 cb_txn_get_hash_mock,
237 gbh_mock)
238 brothers_spec = [
239 # (brother list of brother bytes, chunk size)
240 ([self.buf(190), self.buf(100)], 90),
241 None, # 2nd block has no brothers
242 ([self.buf(130)], 60),
243 ]
244 blocks_spec = [
245 # (block bytes, chunk size, brothers)
246 (self.buf(300), 80, brothers_spec[0]),
247 (self.buf(250), 100, brothers_spec[1]),
248 (self.buf(140), 50, brothers_spec[2]),
249 ]
251 side_effect = [
252 bs for excs in map(self.spec_to_exchange, blocks_spec)
253 for bs in excs
254 ]
256 # Make the metadata of the third block fail
257 # First block meta & chunks & bro metas & chunks
258 # + second block meta & chunks & bro meta
259 # + third block meta
260 exchange_index = (
261 (1 + 300//80 + 1) + 1 + (1 + 190//90 + 1) + (1 + 100//90 + 1) +
262 (1 + 250//100 + 1) + 1 +
263 1
264 )
266 if type(error_code) == bytes:
267 side_effect[exchange_index] = error_code
268 else:
269 side_effect[exchange_index] = CommException("a-message", error_code)
270 side_effect = side_effect[:exchange_index + 1]
271 self.dongle.exchange.side_effect = side_effect
273 blocks_hex = list(map(lambda bs: bs[0].hex(), blocks_spec))
275 brothers_list = list(map(
276 lambda bs: list(map(
277 lambda b: b.hex(), bs[0])) if bs else [],
278 brothers_spec))
280 self.assertEqual(
281 (False, response),
282 self.hsm2dongle.advance_blockchain(blocks_hex, brothers_list),
283 )
285 self.assert_exchange([
286 [0x10, 0x02, 0x00, 0x00, 0x00, 0x03], # Init, 3 blocks
287 [0x10, 0x03, 0x00, 0x4B] +
288 [0x78, 0x78, 0x78, 0x78], # Blk #1 meta
289 [0x10, 0x04] + list(blocks_spec[0][0][80*0:80*1]), # Blk #1 chunk
290 [0x10, 0x04] + list(blocks_spec[0][0][80*1:80*2]), # Blk #1 chunk
291 [0x10, 0x04] + list(blocks_spec[0][0][80*2:80*3]), # Blk #1 chunk
292 [0x10, 0x04] + list(blocks_spec[0][0][80*3:80*4]), # Blk #1 chunk
293 [0x10, 0x07, 0x02], # Blk #1 brother count
294 [0x10, 0x08, 0x00, 0x2f, 0x4c, 0x4c, 0x4c, 0x4c], # Blk #1 bro #1 meta
295 [0x10, 0x09] + list(brothers_spec[0][0][0][90*0:90*1]), # Blk #1 bro #1 chunk
296 [0x10, 0x09] + list(brothers_spec[0][0][0][90*1:90*2]), # Blk #1 bro #1 chunk
297 [0x10, 0x09] + list(brothers_spec[0][0][0][90*2:90*3]), # Blk #1 bro #1 chunk
298 [0x10, 0x08, 0x00, 0x19, 0x28, 0x28, 0x28, 0x28], # Blk #1 bro #2 meta
299 [0x10, 0x09] + list(brothers_spec[0][0][1][90*0:90*1]), # Blk #1 bro #2 chunk
300 [0x10, 0x09] + list(brothers_spec[0][0][1][90*1:90*2]), # Blk #1 bro #2 chunk
301 [0x10, 0x03, 0x00, 0x3E] +
302 [0x64, 0x64, 0x64, 0x64], # Blk #2 meta
303 [0x10, 0x04] + list(blocks_spec[1][0][100*0:100*1]), # Blk #2 chunk
304 [0x10, 0x04] + list(blocks_spec[1][0][100*1:100*2]), # Blk #2 chunk
305 [0x10, 0x04] + list(blocks_spec[1][0][100*2:100*3]), # Blk #2 chunk
306 [0x10, 0x07, 0x00], # Blk #2 brother count
307 [0x10, 0x03, 0x00, 0x23] +
308 [0x38, 0x38, 0x38, 0x38], # Blk #3 meta
309 ])
311 @patch("ledger.hsm2dongle.rlp_mm_payload_size")
312 def test_advance_blockchain_metadata_error_generating(self, mmplsize_mock):
313 mmplsize_mock.side_effect = ValueError()
314 self.dongle.exchange.side_effect = [bytes([0, 0, 0x03])]
316 self.assertEqual(
317 (False, -2),
318 self.hsm2dongle.advance_blockchain(["first-block", "second-block"],
319 [[], []]),
320 )
322 self.assert_exchange([
323 [0x10, 0x02, 0x00, 0x00, 0x00, 0x02], # Init, 2 blocks
324 ])
325 self.assertEqual([call("first-block")], mmplsize_mock.call_args_list)
327 @parameterized.expand([
328 ("prot_invalid", CommException("a-message", 0x6B87), -1),
329 ("unexpected", CommException("a-message", 0x6BFF), -10),
330 ("invalid_response", bytes([0, 0, 0xFF]), -10),
331 ])
332 def test_advance_blockchain_init_error(self, _, error, response):
333 self.dongle.exchange.side_effect = [error]
335 self.assertEqual(
336 (False, response),
337 self.hsm2dongle.advance_blockchain(["first-block", "second-block"],
338 [[], []]),
339 )
341 self.assert_exchange([
342 [0x10, 0x02, 0x00, 0x00, 0x00, 0x02], # Init, 2 blocks
343 ])