Coverage for tests/sgx/hsm2dongle_cmds/test_hsm2dongle_advance_blockchain.py: 100%
89 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, HSM2DongleTestMode
26from ledgerblue.commException import CommException
28import logging
30logging.disable(logging.CRITICAL)
33class TestHSM2DongleSGXAdvanceBlockchain(TestHSM2DongleBase):
34 def get_test_mode(self):
35 return HSM2DongleTestMode.SGX
37 def setup_mocks(self,
38 mmplsize_mock,
39 get_cb_txn_mock,
40 cb_txn_get_hash_mock,
41 gbh_mock):
42 mmplsize_mock.side_effect = lambda h: len(h)//8
43 get_cb_txn_mock.side_effect = lambda h: {"cb_txn": h}
44 cb_txn_get_hash_mock.side_effect = lambda h: \
45 (bytes([len(h["cb_txn"])//5])*4).hex()
46 gbh_mock.return_value = "00"
48 def spec_to_exchange(self, spec, trim=False):
49 # Chunk size is ignored when requesting block data, full bh is always sent
51 # Block header
52 exchanges = [
53 bytes([0, 0, 0x03]),
54 bytes([0, 0, 0x04, 0xAA]),
55 ]
57 # Spec has brothers?
58 if len(spec) == 3:
59 exchanges += [bytes([0, 0, 0x07])] # Request brother list metadata
60 if len(spec) == 3 and spec[2] is not None:
61 brother_count = len(spec[2][0])
62 for i in range(brother_count):
63 exchanges += [
64 bytes([0, 0, 0x08]),
65 bytes([0, 0, 0x09, 0xBB])
66 ]
68 return exchanges
70 @parameterized.expand([
71 ("partial_v2.0.x", 0x05, 2),
72 ("total_v2.0.x", 0x06, 1),
73 ("partial_v2.1.x", 0x05, 2),
74 ("total_v2.1.x", 0x06, 1),
75 ])
76 @patch("ledger.hsm2dongle.get_block_hash")
77 @patch("ledger.hsm2dongle.coinbase_tx_get_hash")
78 @patch("ledger.hsm2dongle.get_coinbase_txn")
79 @patch("ledger.hsm2dongle.rlp_mm_payload_size")
80 def test_advance_blockchain_ok(
81 self,
82 _,
83 device_response,
84 expected_response,
85 mmplsize_mock,
86 get_cb_txn_mock,
87 cb_txn_get_hash_mock,
88 gbh_mock,
89 ):
90 self.setup_mocks(mmplsize_mock,
91 get_cb_txn_mock,
92 cb_txn_get_hash_mock,
93 gbh_mock)
94 brothers_spec = [
95 # (brother list of brother bytes, chunk size)
96 ([self.buf(190), self.buf(100)], None),
97 None, # 2nd block has no brothers
98 ([self.buf(130)], None),
99 ]
100 blocks_spec = [
101 # (block bytes, chunk size, brothers)
102 (self.buf(300), None, brothers_spec[0]),
103 (self.buf(250), None, brothers_spec[1]),
104 (self.buf(140), None, brothers_spec[2]),
105 ]
107 self.dongle.exchange.side_effect = [
108 bs for excs in map(self.spec_to_exchange, blocks_spec)
109 for bs in excs
110 ] + [bytes([0, 0, device_response])] # Success response
112 blocks_hex = list(map(lambda bs: bs[0].hex(), blocks_spec))
113 brothers_list = list(map(
114 lambda bs: list(map(
115 lambda b: b.hex(), bs[0])) if bs else [],
116 brothers_spec))
117 self.assertEqual(
118 (True, expected_response),
119 self.hsm2dongle.advance_blockchain(blocks_hex, brothers_list),
120 )
122 self.assert_exchange([
123 [0x10, 0x02, 0x00, 0x00, 0x00, 0x03], # Init, 3 blocks
124 [0x10, 0x03, 0x00, 0x4B] +
125 [0x78, 0x78, 0x78, 0x78], # Blk #1 meta
126 [0x10, 0x04] + list(blocks_spec[0][0]), # Blk #1
127 [0x10, 0x07, 0x02], # Blk #1 brother count
128 [0x10, 0x08, 0x00, 0x2f, 0x4c, 0x4c, 0x4c, 0x4c], # Blk #1 bro #1 meta
129 [0x10, 0x09] + list(brothers_spec[0][0][0]), # Blk #1 bro #1
130 [0x10, 0x08, 0x00, 0x19, 0x28, 0x28, 0x28, 0x28], # Blk #1 bro #2 meta
131 [0x10, 0x09] + list(brothers_spec[0][0][1]), # Blk #1 bro #2
132 [0x10, 0x03, 0x00, 0x3E] +
133 [0x64, 0x64, 0x64, 0x64], # Blk #2 meta
134 [0x10, 0x04] + list(blocks_spec[1][0]), # Blk #2
135 [0x10, 0x07, 0x00], # Blk #2 brother count
136 [0x10, 0x03, 0x00, 0x23] +
137 [0x38, 0x38, 0x38, 0x38], # Blk #3 meta
138 [0x10, 0x04] + list(blocks_spec[2][0]), # Blk #3
139 [0x10, 0x07, 0x01], # Blk #3 brother count
140 [0x10, 0x08, 0x00, 0x20, 0x34, 0x34, 0x34, 0x34], # Blk #3 bro #1 meta
141 [0x10, 0x09] + list(brothers_spec[2][0][0]), # Blk #3 bro #1
142 ])
144 @parameterized.expand(TestHSM2DongleBase.CHUNK_ERROR_MAPPINGS)
145 @patch("ledger.hsm2dongle.get_block_hash")
146 @patch("ledger.hsm2dongle.coinbase_tx_get_hash")
147 @patch("ledger.hsm2dongle.get_coinbase_txn")
148 @patch("ledger.hsm2dongle.rlp_mm_payload_size")
149 def test_advance_blockchain_bh_error_result(
150 self,
151 _,
152 error_code,
153 response,
154 mmplsize_mock,
155 get_cb_txn_mock,
156 cb_txn_get_hash_mock,
157 gbh_mock,
158 ):
159 self.setup_mocks(mmplsize_mock,
160 get_cb_txn_mock,
161 cb_txn_get_hash_mock,
162 gbh_mock)
163 brothers_spec = [
164 # (brother list of brother bytes, chunk size)
165 ([self.buf(190), self.buf(100)], None),
166 None, # 2nd block has no brothers
167 ([self.buf(130)], None),
168 ]
169 blocks_spec = [
170 # (block bytes, chunk size, brothers)
171 (self.buf(300), None, brothers_spec[0]),
172 (self.buf(250), None, brothers_spec[1]),
173 (self.buf(140), None, brothers_spec[2]),
174 ]
176 side_effect = [
177 bs for excs in map(self.spec_to_exchange, blocks_spec)
178 for bs in excs
179 ]
181 # Make the the second block fail
182 # First block meta & data & bro metas & data
183 # + second block meta & data
184 exchange_index = 2 + 1 + 4 + 2
186 if type(error_code) == bytes:
187 side_effect[exchange_index] = error_code
188 else:
189 side_effect[exchange_index] = CommException("a-message", error_code)
190 side_effect = side_effect[:exchange_index + 1]
191 self.dongle.exchange.side_effect = side_effect
193 blocks_hex = list(map(lambda bs: bs[0].hex(), blocks_spec))
194 brothers_list = list(map(
195 lambda bs: list(map(
196 lambda b: b.hex(), bs[0])) if bs else [],
197 brothers_spec))
199 self.assertEqual(
200 (False, response),
201 self.hsm2dongle.advance_blockchain(blocks_hex, brothers_list),
202 )
204 self.assert_exchange([
205 [0x10, 0x02, 0x00, 0x00, 0x00, 0x03], # Init, 3 blocks
206 [0x10, 0x03, 0x00, 0x4B] +
207 [0x78, 0x78, 0x78, 0x78], # Blk #1 meta
208 [0x10, 0x04] + list(blocks_spec[0][0]), # Blk #1
209 [0x10, 0x07, 0x02], # Blk #1 brother count
210 [0x10, 0x08, 0x00, 0x2f, 0x4c, 0x4c, 0x4c, 0x4c], # Blk #1 bro #1 meta
211 [0x10, 0x09] + list(brothers_spec[0][0][0]), # Blk #1 bro #1
212 [0x10, 0x08, 0x00, 0x19, 0x28, 0x28, 0x28, 0x28], # Blk #1 bro #2 meta
213 [0x10, 0x09] + list(brothers_spec[0][0][1]), # Blk #1 bro #2
214 [0x10, 0x03, 0x00, 0x3E] +
215 [0x64, 0x64, 0x64, 0x64], # Blk #2 meta
216 [0x10, 0x04] + list(blocks_spec[1][0]), # Blk #2
217 ])
219 @parameterized.expand([
220 ("prot_invalid", 0x6B87, -3),
221 ("unexpected", 0x6BFF, -10),
222 ("error_response", bytes([0, 0, 0xFF]), -10),
223 ])
224 @patch("ledger.hsm2dongle.get_block_hash")
225 @patch("ledger.hsm2dongle.coinbase_tx_get_hash")
226 @patch("ledger.hsm2dongle.get_coinbase_txn")
227 @patch("ledger.hsm2dongle.rlp_mm_payload_size")
228 def test_advance_blockchain_metadata_error_result(
229 self,
230 _,
231 error_code,
232 response,
233 mmplsize_mock,
234 get_cb_txn_mock,
235 cb_txn_get_hash_mock,
236 gbh_mock,
237 ):
238 self.setup_mocks(mmplsize_mock,
239 get_cb_txn_mock,
240 cb_txn_get_hash_mock,
241 gbh_mock)
242 brothers_spec = [
243 # (brother list of brother bytes, chunk size)
244 ([self.buf(190), self.buf(100)], None),
245 None, # 2nd block has no brothers
246 ([self.buf(130)], None),
247 ]
248 blocks_spec = [
249 # (block bytes, chunk size, brothers)
250 (self.buf(300), None, brothers_spec[0]),
251 (self.buf(250), None, brothers_spec[1]),
252 (self.buf(140), None, brothers_spec[2]),
253 ]
255 side_effect = [
256 bs for excs in map(self.spec_to_exchange, blocks_spec)
257 for bs in excs
258 ]
260 # Make the metadata of the third block fail
261 # First block meta & data + bro count + bro metas & data
262 # + second block meta & data & bro meta
263 # + third block meta
264 exchange_index = 2 + 1 + 4 + 2 + 1 + 1
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]), # Blk #1
290 [0x10, 0x07, 0x02], # Blk #1 brother count
291 [0x10, 0x08, 0x00, 0x2f, 0x4c, 0x4c, 0x4c, 0x4c], # Blk #1 bro #1 meta
292 [0x10, 0x09] + list(brothers_spec[0][0][0]), # Blk #1 bro #1
293 [0x10, 0x08, 0x00, 0x19, 0x28, 0x28, 0x28, 0x28], # Blk #1 bro #2 meta
294 [0x10, 0x09] + list(brothers_spec[0][0][1]), # Blk #1 bro #2
295 [0x10, 0x03, 0x00, 0x3E] +
296 [0x64, 0x64, 0x64, 0x64], # Blk #2 meta
297 [0x10, 0x04] + list(blocks_spec[1][0]), # Blk #2
298 [0x10, 0x07, 0x00], # Blk #2 brother count
299 [0x10, 0x03, 0x00, 0x23] +
300 [0x38, 0x38, 0x38, 0x38], # Blk #3 meta
301 ])
303 @patch("ledger.hsm2dongle.rlp_mm_payload_size")
304 def test_advance_blockchain_metadata_error_generating(self, mmplsize_mock):
305 mmplsize_mock.side_effect = ValueError()
306 self.dongle.exchange.side_effect = [bytes([0, 0, 0x03])]
308 self.assertEqual(
309 (False, -2),
310 self.hsm2dongle.advance_blockchain(["first-block", "second-block"],
311 [[], []]),
312 )
314 self.assert_exchange([
315 [0x10, 0x02, 0x00, 0x00, 0x00, 0x02], # Init, 2 blocks
316 ])
317 self.assertEqual([call("first-block")], mmplsize_mock.call_args_list)
319 @parameterized.expand([
320 ("prot_invalid", CommException("a-message", 0x6B87), -1),
321 ("unexpected", CommException("a-message", 0x6BFF), -10),
322 ("invalid_response", bytes([0, 0, 0xFF]), -10),
323 ])
324 def test_advance_blockchain_init_error(self, _, error, response):
325 self.dongle.exchange.side_effect = [error]
327 self.assertEqual(
328 (False, response),
329 self.hsm2dongle.advance_blockchain(["first-block", "second-block"],
330 [[], []]),
331 )
333 self.assert_exchange([
334 [0x10, 0x02, 0x00, 0x00, 0x00, 0x02], # Init, 2 blocks
335 ])