|
| 1 | +# This file is part of the KeepKey project. |
| 2 | +# |
| 3 | +# Copyright (C) 2026 KeepKey |
| 4 | +# |
| 5 | +# This library is free software: you can redistribute it and/or modify |
| 6 | +# it under the terms of the GNU Lesser General Public License as published by |
| 7 | +# the Free Software Foundation, either version 3 of the License, or |
| 8 | +# (at your option) any later version. |
| 9 | +# |
| 10 | +# This library is distributed in the hope that it will be useful, |
| 11 | +# but WITHOUT ANY WARRANTY; without even the implied warranty of |
| 12 | +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| 13 | +# GNU Lesser General Public License for more details. |
| 14 | +# |
| 15 | +# You should have received a copy of the GNU Lesser General Public License |
| 16 | +# along with this library. If not, see <http://www.gnu.org/licenses/>. |
| 17 | + |
| 18 | +"""Hive (SLIP-0048) device tests — multi-role keys + account operations. |
| 19 | +
|
| 20 | +Uses the standard 12-word test seed (mnemonic12, "alcohol ... aisle") via |
| 21 | +setup_mnemonic_nopin_nopassphrase(). |
| 22 | +
|
| 23 | +The account_create / account_update / transfer tests are self-validating: they |
| 24 | +recover the signer from the 65-byte device signature over |
| 25 | +SHA256(chain_id || serialized_tx) and assert it equals the device-derived |
| 26 | +signing key. This exercises the device AND validates the attestation-digest |
| 27 | +contract documented in keepkey-vault docs/HIVE-ATTESTATION-DIGEST-SPEC.md — |
| 28 | +no precomputed golden vector required, and not circular (recovery is an |
| 29 | +independent cryptographic check). |
| 30 | +""" |
| 31 | + |
| 32 | +import hashlib |
| 33 | +import unittest |
| 34 | + |
| 35 | +import common |
| 36 | + |
| 37 | +from ecdsa import SECP256k1, VerifyingKey |
| 38 | +from ecdsa.util import sigdecode_string |
| 39 | + |
| 40 | +from keepkeylib import hive |
| 41 | +from keepkeylib.tools import parse_path |
| 42 | + |
| 43 | +# Hive mainnet chain id: beeab0de followed by 28 zero bytes (32 bytes). |
| 44 | +HIVE_CHAIN_ID = bytes.fromhex("beeab0de" + "00" * 28) |
| 45 | + |
| 46 | +# SLIP-0048 roles (hardened offsets within the role component). |
| 47 | +ROLE_OWNER, ROLE_ACTIVE, ROLE_MEMO, ROLE_POSTING = 0, 1, 3, 4 |
| 48 | + |
| 49 | +HIVE_OP_TRANSFER = 2 |
| 50 | +HIVE_OP_ACCOUNT_CREATE = 9 |
| 51 | +HIVE_OP_ACCOUNT_UPDATE = 10 |
| 52 | + |
| 53 | + |
| 54 | +def hive_path(role, account_index=0): |
| 55 | + """m/48'/13'/role'/account'/0' — all five components hardened.""" |
| 56 | + h = 0x80000000 |
| 57 | + return [h + 48, h + 13, h + role, h + account_index, h] |
| 58 | + |
| 59 | + |
| 60 | +def recover_compressed(serialized_tx, sig65): |
| 61 | + """Recover the 33-byte compressed signer pubkey from a Hive device signature. |
| 62 | +
|
| 63 | + Mirrors HIVE-ATTESTATION-DIGEST-SPEC.md §1-2: |
| 64 | + digest = SHA256(chain_id || serialized_tx) |
| 65 | + sig[0] = 27 + recovery_id + 4 -> recovery_id = sig[0] - 31 |
| 66 | + sig[1:65] = r || s |
| 67 | + """ |
| 68 | + assert len(sig65) == 65, "Hive signature must be 65 bytes" |
| 69 | + recid = sig65[0] - 31 |
| 70 | + assert 0 <= recid <= 3, "unexpected recovery header byte %d" % sig65[0] |
| 71 | + digest = hashlib.sha256(HIVE_CHAIN_ID + serialized_tx).digest() |
| 72 | + candidates = VerifyingKey.from_public_key_recovery_with_digest( |
| 73 | + sig65[1:], digest, SECP256k1, hashfunc=hashlib.sha256, sigdecode=sigdecode_string |
| 74 | + ) |
| 75 | + return candidates[recid].to_string("compressed") |
| 76 | + |
| 77 | + |
| 78 | +class TestMsgHive(common.KeepKeyTest): |
| 79 | + |
| 80 | + def _owner_raw(self): |
| 81 | + """Device-derived owner key (33-byte compressed) at account 0.""" |
| 82 | + resp = hive.get_public_key(self.client, hive_path(ROLE_OWNER), show_display=False) |
| 83 | + self.assertEqual(len(resp.raw_public_key), 33) |
| 84 | + return resp.raw_public_key |
| 85 | + |
| 86 | + def test_hive_get_public_key_active(self): |
| 87 | + """Active-role key derives and returns an STM-prefixed key + 33-byte raw.""" |
| 88 | + self.requires_firmware("7.15.0") |
| 89 | + self.requires_message("HiveGetPublicKey") |
| 90 | + self.setup_mnemonic_nopin_nopassphrase() |
| 91 | + |
| 92 | + resp = hive.get_public_key(self.client, hive_path(ROLE_ACTIVE), show_display=False) |
| 93 | + self.assertTrue(resp.public_key.startswith("STM"), "expected STM-prefixed key") |
| 94 | + self.assertEqual(len(resp.raw_public_key), 33) |
| 95 | + self.assertIn(resp.raw_public_key[0], (2, 3), "compressed pubkey prefix") |
| 96 | + |
| 97 | + def test_hive_get_public_keys_all_roles(self): |
| 98 | + """All four role keys derive, are distinct, and STM-formatted.""" |
| 99 | + self.requires_firmware("7.15.0") |
| 100 | + self.requires_message("HiveGetPublicKeys") |
| 101 | + self.setup_mnemonic_nopin_nopassphrase() |
| 102 | + |
| 103 | + resp = hive.get_public_keys(self.client, account_index=0, show_display=False) |
| 104 | + keys = [resp.owner_key, resp.active_key, resp.memo_key, resp.posting_key] |
| 105 | + for k in keys: |
| 106 | + self.assertTrue(k.startswith("STM"), "expected STM-prefixed key, got %r" % k) |
| 107 | + self.assertEqual(len(set(keys)), 4, "the four role keys must be distinct") |
| 108 | + |
| 109 | + # The single-key path must agree with the bulk path for the active role. |
| 110 | + single = hive.get_public_key(self.client, hive_path(ROLE_ACTIVE), show_display=False) |
| 111 | + self.assertEqual(single.public_key, resp.active_key) |
| 112 | + |
| 113 | + def test_hive_sign_transfer(self): |
| 114 | + """Transfer (op 2) signs and the signature recovers to the active key.""" |
| 115 | + self.requires_firmware("7.15.0") |
| 116 | + self.requires_message("HiveSignTx") |
| 117 | + self.setup_mnemonic_nopin_nopassphrase() |
| 118 | + |
| 119 | + active = hive.get_public_key(self.client, hive_path(ROLE_ACTIVE), show_display=False) |
| 120 | + resp = hive.sign_tx( |
| 121 | + self.client, |
| 122 | + address_n=hive_path(ROLE_ACTIVE), |
| 123 | + chain_id=HIVE_CHAIN_ID, |
| 124 | + ref_block_num=12345, |
| 125 | + ref_block_prefix=67890, |
| 126 | + expiration=1700000000, |
| 127 | + sender="kktester", |
| 128 | + recipient="kkrecipient", |
| 129 | + amount=1000, # 1.000 HIVE |
| 130 | + decimals=3, |
| 131 | + asset_symbol="HIVE", |
| 132 | + memo="kktest", |
| 133 | + ) |
| 134 | + self.assertEqual(len(resp.signature), 65) |
| 135 | + self.assertIn(resp.signature[0], (31, 32)) |
| 136 | + self.assertTrue(len(resp.serialized_tx) > 0) |
| 137 | + self.assertEqual(recover_compressed(resp.serialized_tx, resp.signature), active.raw_public_key) |
| 138 | + # op byte sits right after header (u16 + u32 + u32) and the 0x01 op-count varint. |
| 139 | + self.assertEqual(resp.serialized_tx[11], HIVE_OP_TRANSFER) |
| 140 | + |
| 141 | + def test_hive_sign_account_create(self): |
| 142 | + """account_create (op 9): signs, recovers to owner key, binds the 4 keys + name. |
| 143 | +
|
| 144 | + This is the attestation a Pioneer sponsor verifies before spending an ACT. |
| 145 | + """ |
| 146 | + self.requires_firmware("7.15.0") |
| 147 | + self.requires_message("HiveSignAccountCreate") |
| 148 | + self.requires_message("HiveGetPublicKeys") |
| 149 | + self.setup_mnemonic_nopin_nopassphrase() |
| 150 | + |
| 151 | + owner_raw = self._owner_raw() |
| 152 | + keys = hive.get_public_keys(self.client, account_index=0, show_display=False) |
| 153 | + |
| 154 | + resp = hive.sign_account_create( |
| 155 | + self.client, |
| 156 | + address_n=hive_path(ROLE_OWNER), |
| 157 | + chain_id=HIVE_CHAIN_ID, |
| 158 | + ref_block_num=12345, |
| 159 | + ref_block_prefix=67890, |
| 160 | + expiration=1700000000, |
| 161 | + creator="kksponsor", |
| 162 | + new_account_name="kktestacct", |
| 163 | + fee_amount=3000, |
| 164 | + owner_key=keys.owner_key, |
| 165 | + active_key=keys.active_key, |
| 166 | + posting_key=keys.posting_key, |
| 167 | + memo_key=keys.memo_key, |
| 168 | + ) |
| 169 | + self.assertEqual(len(resp.signature), 65) |
| 170 | + self.assertIn(resp.signature[0], (31, 32)) |
| 171 | + |
| 172 | + # Attestation: signature recovers to the device owner key. |
| 173 | + self.assertEqual(recover_compressed(resp.serialized_tx, resp.signature), owner_raw) |
| 174 | + |
| 175 | + tx = resp.serialized_tx |
| 176 | + self.assertEqual(tx[11], HIVE_OP_ACCOUNT_CREATE) |
| 177 | + # The new account name and all four device-raw role keys are bound into the |
| 178 | + # signed bytes (per spec §3); a sponsor parses these to confirm what it creates. |
| 179 | + self.assertIn(b"kktestacct", tx) |
| 180 | + single = hive.get_public_key # local alias |
| 181 | + for role in (ROLE_OWNER, ROLE_ACTIVE, ROLE_POSTING, ROLE_MEMO): |
| 182 | + raw = single(self.client, hive_path(role), show_display=False).raw_public_key |
| 183 | + self.assertIn(raw, tx, "role %d key must be embedded in account_create" % role) |
| 184 | + |
| 185 | + def test_hive_sign_account_update(self): |
| 186 | + """account_update (op 10): signs and recovers to the owner key.""" |
| 187 | + self.requires_firmware("7.15.0") |
| 188 | + self.requires_message("HiveSignAccountUpdate") |
| 189 | + self.requires_message("HiveGetPublicKeys") |
| 190 | + self.setup_mnemonic_nopin_nopassphrase() |
| 191 | + |
| 192 | + owner_raw = self._owner_raw() |
| 193 | + keys = hive.get_public_keys(self.client, account_index=0, show_display=False) |
| 194 | + |
| 195 | + resp = hive.sign_account_update( |
| 196 | + self.client, |
| 197 | + address_n=hive_path(ROLE_OWNER), |
| 198 | + chain_id=HIVE_CHAIN_ID, |
| 199 | + ref_block_num=12345, |
| 200 | + ref_block_prefix=67890, |
| 201 | + expiration=1700000000, |
| 202 | + account="kktestacct", |
| 203 | + new_owner_key=keys.owner_key, |
| 204 | + new_active_key=keys.active_key, |
| 205 | + new_posting_key=keys.posting_key, |
| 206 | + new_memo_key=keys.memo_key, |
| 207 | + ) |
| 208 | + self.assertEqual(len(resp.signature), 65) |
| 209 | + self.assertIn(resp.signature[0], (31, 32)) |
| 210 | + self.assertEqual(recover_compressed(resp.serialized_tx, resp.signature), owner_raw) |
| 211 | + self.assertEqual(resp.serialized_tx[11], HIVE_OP_ACCOUNT_UPDATE) |
| 212 | + self.assertIn(b"kktestacct", resp.serialized_tx) |
| 213 | + |
| 214 | + |
| 215 | +if __name__ == "__main__": |
| 216 | + unittest.main() |
0 commit comments