test(hive): parse serialized_tx and bind every field by position

Addresses review: substring-presence was too weak — a role swap (both keys
present), a creator rewrite, or an amount change could still pass.

Add a cursor-based Graphene reader matching the firmware append_* layout exactly
(incl. account_update's 0x01 optional-present flags, asset symbol padding, and
the no-wrapper memo_key) and rewrite all three signing tests to parse and assert
each field at its expected position + assert_end() for no trailing bytes:
- transfer: from / to / amount / precision / symbol / memo
- account_create: fee / creator / name / owner|active|posting authority slots / memo_key
- account_update: account / each replacement key in its slot / memo_key

Recovery assertions retained. Parser validated offline against hand-built
firmware-format bytes.

Author: BitHighlander

tests/test_msg_hive.py

@@ -75,13 +75,74 @@ def recover_compressed(serialized_tx, sig65):
     return candidates[recid].to_string("compressed")
 
 
-class TestMsgHive(common.KeepKeyTest):
+class _Reader:
+    """Cursor over the device-emitted Graphene bytes. Matches firmware
+    serialization exactly (see hive.c append_* helpers)."""
+
+    def __init__(self, data):
+        self.d = data
+        self.i = 0
+
+    def take(self, n):
+        v = self.d[self.i:self.i + n]
+        assert len(v) == n, "truncated serialized_tx"
+        self.i += n
+        return v
+
+    def u8(self):
+        return self.take(1)[0]
+
+    def u16le(self):
+        return int.from_bytes(self.take(2), "little")
+
+    def u32le(self):
+        return int.from_bytes(self.take(4), "little")
+
+    def u64le(self):
+        return int.from_bytes(self.take(8), "little")
+
+    def varint(self):
+        shift = result = 0
+        while True:
+            b = self.u8()
+            result |= (b & 0x7F) << shift
+            if not (b & 0x80):
+                return result
+            shift += 7
+
+    def string(self):
+        return self.take(self.varint())
+
+    def asset(self):
+        amount = self.u64le()
+        precision = self.u8()
+        symbol = self.take(7).rstrip(b"\x00").decode()
+        return amount, precision, symbol
+
+    def authority(self):
+        # weight_threshold=1, 0 account auths, 1 key auth, key(33), weight=1
+        assert self.u32le() == 1, "weight_threshold must be 1"
+        assert self.varint() == 0, "expected 0 account_auths"
+        assert self.varint() == 1, "expected 1 key_auth"
+        key = self.take(33)
+        assert self.u16le() == 1, "key weight must be 1"
+        return key
+
+    def assert_end(self):
+        assert self.i == len(self.d), "trailing bytes after operation (offset %d/%d)" % (self.i, len(self.d))
+
+
+def _parse_header(r, expected_op):
+    ref_block_num = r.u16le()
+    ref_block_prefix = r.u32le()
+    expiration = r.u32le()
+    assert r.varint() == 1, "expected exactly one operation"
+    op_type = r.varint()
+    assert op_type == expected_op, "op_type %d != expected %d" % (op_type, expected_op)
+    return ref_block_num, ref_block_prefix, expiration
 
-    def _owner_raw(self):
-        """Device-derived owner key (33-byte compressed) at account 0."""
-        resp = hive.get_public_key(self.client, hive_path(ROLE_OWNER), show_display=False)
-        self.assertEqual(len(resp.raw_public_key), 33)
-        return resp.raw_public_key
+
+class TestMsgHive(common.KeepKeyTest):
 
     def test_hive_get_public_key_active(self):
         """Active-role key derives and returns an STM-prefixed key + 33-byte raw."""
@@ -133,10 +194,19 @@ def test_hive_sign_transfer(self):
         )
         self.assertEqual(len(resp.signature), 65)
         self.assertIn(resp.signature[0], (31, 32))
-        self.assertTrue(len(resp.serialized_tx) > 0)
         self.assertEqual(recover_compressed(resp.serialized_tx, resp.signature), active.raw_public_key)
-        # op byte sits right after header (u16 + u32 + u32) and the 0x01 op-count varint.
-        self.assertEqual(resp.serialized_tx[11], HIVE_OP_TRANSFER)
+
+        # Parse the transfer op and bind EVERY field — a rewritten recipient,
+        # amount, or asset must fail, not just a missing substring.
+        r = _Reader(resp.serialized_tx)
+        ref_num, ref_prefix, expiration = _parse_header(r, HIVE_OP_TRANSFER)
+        self.assertEqual((ref_num, ref_prefix, expiration), (12345, 67890, 1700000000))
+        self.assertEqual(r.string(), b"kktester")     # from
+        self.assertEqual(r.string(), b"kkrecipient")  # to
+        self.assertEqual(r.asset(), (1000, 3, "HIVE"))
+        self.assertEqual(r.string(), b"kktest")        # memo
+        self.assertEqual(r.varint(), 0)                # extensions
+        r.assert_end()
 
     def test_hive_sign_account_create(self):
         """account_create (op 9): signs, recovers to owner key, binds the 4 keys + name.
@@ -148,7 +218,9 @@ def test_hive_sign_account_create(self):
         self.requires_message("HiveGetPublicKeys")
         self.setup_mnemonic_nopin_nopassphrase()
 
-        owner_raw = self._owner_raw()
+        # Device-derived raw keys per role, for slot-exact comparison.
+        raw = {role: hive.get_public_key(self.client, hive_path(role), show_display=False).raw_public_key
+               for role in (ROLE_OWNER, ROLE_ACTIVE, ROLE_POSTING, ROLE_MEMO)}
         keys = hive.get_public_keys(self.client, account_index=0, show_display=False)
 
         resp = hive.sign_account_create(
@@ -170,17 +242,23 @@ def test_hive_sign_account_create(self):
         self.assertIn(resp.signature[0], (31, 32))
 
         # Attestation: signature recovers to the device owner key.
-        self.assertEqual(recover_compressed(resp.serialized_tx, resp.signature), owner_raw)
-
-        tx = resp.serialized_tx
-        self.assertEqual(tx[11], HIVE_OP_ACCOUNT_CREATE)
-        # The new account name and all four device-raw role keys are bound into the
-        # signed bytes (per spec §3); a sponsor parses these to confirm what it creates.
-        self.assertIn(b"kktestacct", tx)
-        single = hive.get_public_key  # local alias
-        for role in (ROLE_OWNER, ROLE_ACTIVE, ROLE_POSTING, ROLE_MEMO):
-            raw = single(self.client, hive_path(role), show_display=False).raw_public_key
-            self.assertIn(raw, tx, "role %d key must be embedded in account_create" % role)
+        self.assertEqual(recover_compressed(resp.serialized_tx, resp.signature), raw[ROLE_OWNER])
+
+        # Parse op 9 and bind EVERY field at its position. A firmware bug that
+        # swaps roles, rewrites the creator, or alters the fee must fail here.
+        r = _Reader(resp.serialized_tx)
+        ref_num, ref_prefix, expiration = _parse_header(r, HIVE_OP_ACCOUNT_CREATE)
+        self.assertEqual((ref_num, ref_prefix, expiration), (12345, 67890, 1700000000))
+        self.assertEqual(r.asset(), (3000, 3, "HIVE"))     # fee
+        self.assertEqual(r.string(), b"kksponsor")          # creator
+        self.assertEqual(r.string(), b"kktestacct")         # new_account_name
+        self.assertEqual(r.authority(), raw[ROLE_OWNER], "owner authority slot")
+        self.assertEqual(r.authority(), raw[ROLE_ACTIVE], "active authority slot")
+        self.assertEqual(r.authority(), raw[ROLE_POSTING], "posting authority slot")
+        self.assertEqual(r.take(33), raw[ROLE_MEMO], "memo_key slot")
+        self.assertEqual(r.string(), b"")                   # json_metadata
+        self.assertEqual(r.varint(), 0)                     # extensions
+        r.assert_end()
 
     def test_hive_sign_account_update(self):
         """account_update (op 10): signs and recovers to the owner key."""
@@ -189,7 +267,8 @@ def test_hive_sign_account_update(self):
         self.requires_message("HiveGetPublicKeys")
         self.setup_mnemonic_nopin_nopassphrase()
 
-        owner_raw = self._owner_raw()
+        raw = {role: hive.get_public_key(self.client, hive_path(role), show_display=False).raw_public_key
+               for role in (ROLE_OWNER, ROLE_ACTIVE, ROLE_POSTING, ROLE_MEMO)}
         keys = hive.get_public_keys(self.client, account_index=0, show_display=False)
 
         resp = hive.sign_account_update(
@@ -207,9 +286,21 @@ def test_hive_sign_account_update(self):
         )
         self.assertEqual(len(resp.signature), 65)
         self.assertIn(resp.signature[0], (31, 32))
-        self.assertEqual(recover_compressed(resp.serialized_tx, resp.signature), owner_raw)
-        self.assertEqual(resp.serialized_tx[11], HIVE_OP_ACCOUNT_UPDATE)
-        self.assertIn(b"kktestacct", resp.serialized_tx)
+        self.assertEqual(recover_compressed(resp.serialized_tx, resp.signature), raw[ROLE_OWNER])
+
+        # Parse op 10 and bind the replacement keys to their slots. A bad impl
+        # that updates the wrong authorities must fail even if op/name are right.
+        r = _Reader(resp.serialized_tx)
+        ref_num, ref_prefix, expiration = _parse_header(r, HIVE_OP_ACCOUNT_UPDATE)
+        self.assertEqual((ref_num, ref_prefix, expiration), (12345, 67890, 1700000000))
+        self.assertEqual(r.string(), b"kktestacct")          # account
+        for role, label in ((ROLE_OWNER, "owner"), (ROLE_ACTIVE, "active"), (ROLE_POSTING, "posting")):
+            self.assertEqual(r.u8(), 0x01, "%s optional-present flag" % label)
+            self.assertEqual(r.authority(), raw[role], "%s authority slot" % label)
+        self.assertEqual(r.take(33), raw[ROLE_MEMO], "memo_key slot")
+        self.assertEqual(r.string(), b"")                    # json_metadata
+        self.assertEqual(r.varint(), 0)                      # extensions
+        r.assert_end()
 
 
 if __name__ == "__main__":