test(hive): device tests for the phase-3 clear-sign op table

Covers the 11 ops added to the firmware clear-sign table: transfer_to_vesting,
withdraw_vesting, limit_order_create/cancel, convert, comment_options,
transfer_to/from_savings, claim_reward_balance, delegate_vesting_shares,
account_update2.

Transactions are built by this file's own Graphene serializer (the _op_*
helpers), never from firmware-emitted bytes, so a parser bug and a serializer
bug cannot cancel out.

Beyond happy-path sign-and-recover, the negative cases pin the invariants that
actually protect the user:

  - asset symbol AND precision are pinned per op: a swapped symbol hides a
    ~2000x value difference behind an identical-looking number, and a wrong
    precision moves the decimal point relative to what the chain applies
  - negative int64 amounts are refused (they would render as enormous
    positive values)
  - comment_options only binds immediately after a comment with a matching
    author and permlink; detached it could redirect the payout of a post
    published earlier that the user is not reviewing on screen
  - account_update2 refuses any authority field
  - beneficiaries must be strictly ascending, unique, and sum to <= 100%
  - zero amounts are refused where meaningless and accepted where meaningful
    (withdraw_vesting 0 stops a power-down, delegate 0 removes a delegation)
  - truncated op bodies are refused rather than partially parsed

Two fixes to existing tests:

  - Three assertions matched rejection strings that the firmware has since
    grouped by cause ("malformed op count", "weight out of range",
    "mixed active+posting auths"). The text is diagnostic, not contractual —
    the protection is that the device refuses — so the assertions follow the
    grouping.
  - rejects_excluded_and_unknown_ops used op type 3 as its "unknown op",
    mislabelled in a comment as comment_options. Op 3 is transfer_to_vesting
    and is now clear-signed, so it no longer reached the unknown-op path.
    Switched to 49 (recurrent_transfer), which is deliberately out of the
    table.

38/38 pass against an emulator built from keepkey-firmware
feat/hive-clearsign-ops-phase3.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: BitHighlander

tests/test_msg_hive.py

@@ -127,6 +127,89 @@ def _op_custom_json(required_auths, required_posting_auths, id_, json_):
     return out + _string(id_) + _string(json_)
 
 
+def _asset(amount, symbol):
+    """int64 LE amount + uint8 precision + 7-byte NUL-padded symbol.
+
+    Precision is pinned per symbol exactly as firmware's cur_asset requires;
+    passing the wrong one is what the negative tests below exercise.
+    """
+    precision = 6 if symbol == "VESTS" else 3
+    return (struct.pack("<q", amount) + bytes([precision]) +
+            symbol.encode("ascii").ljust(7, b"\x00"))
+
+
+def _asset_raw(amount, precision, symbol):
+    """Asset with a caller-chosen precision, for malformed-input tests."""
+    return (struct.pack("<q", amount) + bytes([precision]) +
+            symbol.encode("ascii").ljust(7, b"\x00"))
+
+
+def _op_transfer_to_vesting(from_, to, amount):
+    return (_varint(3) + _string(from_) + _string(to) + _asset(amount, "HIVE"))
+
+
+def _op_withdraw_vesting(account, vesting_shares):
+    return _varint(4) + _string(account) + _asset(vesting_shares, "VESTS")
+
+
+def _op_limit_order_create(owner, orderid, sell, sell_sym, recv, recv_sym,
+                           fill_or_kill, expiration):
+    return (_varint(5) + _string(owner) + struct.pack("<I", orderid) +
+            _asset(sell, sell_sym) + _asset(recv, recv_sym) +
+            bytes([1 if fill_or_kill else 0]) + struct.pack("<I", expiration))
+
+
+def _op_limit_order_cancel(owner, orderid):
+    return _varint(6) + _string(owner) + struct.pack("<I", orderid)
+
+
+def _op_convert(owner, requestid, amount):
+    return (_varint(8) + _string(owner) + struct.pack("<I", requestid) +
+            _asset(amount, "HBD"))
+
+
+def _op_comment_options(author, permlink, max_payout, percent_hbd,
+                        allow_votes=True, allow_curation=True,
+                        beneficiaries=None):
+    out = (_varint(19) + _string(author) + _string(permlink) +
+           _asset(max_payout, "HBD") + struct.pack("<H", percent_hbd) +
+           bytes([1 if allow_votes else 0, 1 if allow_curation else 0]))
+    if not beneficiaries:
+        return out + _varint(0)
+    out += _varint(1) + _varint(0) + _varint(len(beneficiaries))
+    for account, weight in beneficiaries:
+        out += _string(account) + struct.pack("<H", weight)
+    return out
+
+
+def _op_transfer_to_savings(from_, to, amount, symbol, memo=""):
+    return (_varint(32) + _string(from_) + _string(to) +
+            _asset(amount, symbol) + _string(memo))
+
+
+def _op_transfer_from_savings(from_, request_id, to, amount, symbol, memo=""):
+    return (_varint(33) + _string(from_) + struct.pack("<I", request_id) +
+            _string(to) + _asset(amount, symbol) + _string(memo))
+
+
+def _op_claim_reward_balance(account, hive_amt, hbd_amt, vests_amt):
+    return (_varint(39) + _string(account) + _asset(hive_amt, "HIVE") +
+            _asset(hbd_amt, "HBD") + _asset(vests_amt, "VESTS"))
+
+
+def _op_delegate_vesting_shares(delegator, delegatee, vesting_shares):
+    return (_varint(40) + _string(delegator) + _string(delegatee) +
+            _asset(vesting_shares, "VESTS"))
+
+
+def _op_account_update2(account, json_metadata, posting_json_metadata,
+                        authority_present=False):
+    return (_varint(43) + _string(account) +
+            bytes([1 if authority_present else 0, 0, 0, 0]) +
+            _string(json_metadata) + _string(posting_json_metadata) +
+            _varint(0))
+
+
 class _Reader:
     """Cursor over the device-emitted Graphene bytes. Matches firmware
     serialization exactly (see hive.c append_* helpers)."""
@@ -697,8 +780,12 @@ def test_hive_sign_ops_rejects_excluded_and_unknown_ops(self):
         for op_type in (2, 9, 10):
             self._assert_ops_fails("dedicated message",
                                    _ops_tx([_varint(op_type)]))
+        # 49 = recurrent_transfer: a real Hive op deliberately kept out of the
+        # table. (This previously used op 3, mislabelled "comment_options";
+        # op 3 is transfer_to_vesting and is now clear-signed, so it no longer
+        # exercises the unknown-op path.)
         self._assert_ops_fails("unsupported operation",
-                               _ops_tx([_varint(3)]))  # comment_options
+                               _ops_tx([_varint(49)]))
 
     def test_hive_sign_ops_rejects_malformed_structure(self):
         """Zero ops, >4 ops, nonzero extensions, trailing bytes, overlong
@@ -715,9 +802,9 @@ def test_hive_sign_ops_rejects_malformed_structure(self):
         self._assert_ops_fails("trailing bytes", _ops_tx([vote]) + b"\x00")
         # op_count as an overlong 6-byte varint encoding of 1
         head = struct.pack("<HII", 12345, 67890, 1700000000)
-        self._assert_ops_fails("malformed op count",
+        self._assert_ops_fails("malformed operation",
                                head + b"\x81\x80\x80\x80\x80\x00" + vote + b"\x00")
-        self._assert_ops_fails("weight out of range",
+        self._assert_ops_fails("value out of range",
                                _ops_tx([_op_vote("kkvoter", "author", "permlink", 10001)]))
 
     def test_hive_sign_ops_role_fences(self):
@@ -742,7 +829,7 @@ def test_hive_sign_ops_role_fences(self):
         self._assert_ops_fails("mixed posting/active", mixed, path=hive_path(ROLE_ACTIVE))
 
         both_auths = _ops_tx([_op_custom_json(["kkadmin"], ["kkplayer"], "x-id", '{"a":1}')])
-        self._assert_ops_fails("mixed active+posting auths", both_auths,
+        self._assert_ops_fails("mixed posting/active", both_auths,
                                path=hive_path(ROLE_ACTIVE))
 
     def test_hive_sign_ops_rejects_oversize(self):
@@ -758,6 +845,205 @@ def test_hive_sign_ops_rejects_oversize(self):
             hive.sign_operations(self.client, hive_path(ROLE_POSTING), tx,
                                  chain_id=HIVE_CHAIN_ID)
 
+    # ── Phase-3 op table ─────────────────────────────────────────────────
+    # Every tx below is built by THIS file's serializer, never by firmware, so
+    # a parser bug and a serializer bug cannot cancel out.
+
+    def _ops_signs_with(self, tx, role):
+        """Sign tx with `role` and assert the signature recovers to that key."""
+        key = hive.get_public_key(self.client, hive_path(role), show_display=False)
+        resp = hive.sign_operations(self.client, hive_path(role), tx,
+                                    chain_id=HIVE_CHAIN_ID)
+        self.assertEqual(self._recover_ops_signer(tx, resp.signature),
+                         key.raw_public_key)
+
+    def test_hive_sign_ops_limit_order_create(self):
+        """The op that motivated phase 3: a HIVE->HBD internal-market swap.
+        Active tier, since it moves funds."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        tx = _ops_tx([_op_limit_order_create("kktrader", 42, 1500, "HIVE",
+                                             400, "HBD", True, 1700003600)])
+        self._ops_signs_with(tx, ROLE_ACTIVE)
+
+    def test_hive_sign_ops_limit_order_cancel(self):
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        self._ops_signs_with(_ops_tx([_op_limit_order_cancel("kktrader", 42)]),
+                             ROLE_ACTIVE)
+
+    def test_hive_sign_ops_active_tier_value_ops(self):
+        """The active-tier ops that move or lock value all sign with active."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        for op in (
+            _op_transfer_to_vesting("kkuser", "kkuser", 1000),
+            _op_convert("kkuser", 7, 2500),
+            _op_transfer_to_savings("kkuser", "kkfriend", 1500, "HBD", "rent"),
+            _op_transfer_from_savings("kkuser", 7, "kkfriend", 1500, "HIVE"),
+            _op_delegate_vesting_shares("kkuser", "kkfriend", 1000000),
+            _op_withdraw_vesting("kkuser", 5000000),
+        ):
+            self._ops_signs_with(_ops_tx([op]), ROLE_ACTIVE)
+
+    def test_hive_sign_ops_posting_tier_ops(self):
+        """claim_reward_balance is posting tier — claiming is not spending."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        tx = _ops_tx([_op_claim_reward_balance("kkuser", 1234, 5678, 90123456)])
+        self._ops_signs_with(tx, ROLE_POSTING)
+
+    def test_hive_sign_ops_zero_amount_semantics(self):
+        """Zero means something for these two and nothing for the rest, so the
+        parser must not apply one blanket rule."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        # 0 VESTS withdraw_vesting cancels an in-progress power-down.
+        self._ops_signs_with(_ops_tx([_op_withdraw_vesting("kkuser", 0)]),
+                             ROLE_ACTIVE)
+        # 0 VESTS delegation removes an existing delegation.
+        self._ops_signs_with(
+            _ops_tx([_op_delegate_vesting_shares("kkuser", "kkfriend", 0)]),
+            ROLE_ACTIVE)
+        # A zero power-up, by contrast, does nothing and is refused.
+        self._assert_ops_fails("amount must be greater than zero",
+                               _ops_tx([_op_transfer_to_vesting("kkuser", "kkuser", 0)]),
+                               path=hive_path(ROLE_ACTIVE))
+        # Nothing to claim.
+        self._assert_ops_fails("no effect",
+                               _ops_tx([_op_claim_reward_balance("kkuser", 0, 0, 0)]))
+
+    def test_hive_sign_ops_asset_symbol_and_precision_pinned(self):
+        """A swapped symbol hides a ~2000x value difference behind an
+        identical-looking number; a wrong precision moves the decimal point
+        relative to what the chain applies. Both must be refused."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+        active = hive_path(ROLE_ACTIVE)
+
+        # transfer_to_vesting is HIVE-only.
+        wrong_symbol = (_varint(3) + _string("kkuser") + _string("kkuser") +
+                        _asset(1000, "HBD"))
+        self._assert_ops_fails("malformed operation", _ops_tx([wrong_symbol]),
+                               path=active)
+        # Right symbol, wrong precision.
+        wrong_precision = (_varint(3) + _string("kkuser") + _string("kkuser") +
+                           _asset_raw(1000, 6, "HIVE"))
+        self._assert_ops_fails("malformed operation", _ops_tx([wrong_precision]),
+                               path=active)
+        # Negative int64 would render as an enormous positive amount.
+        negative = (_varint(3) + _string("kkuser") + _string("kkuser") +
+                    _asset_raw(-1000, 3, "HIVE"))
+        self._assert_ops_fails("malformed operation", _ops_tx([negative]),
+                               path=active)
+        # An order priced VESTS-for-HBD is not a market that exists.
+        vests_order = (_varint(5) + _string("kktrader") + struct.pack("<I", 1) +
+                       _asset(100, "VESTS") + _asset(100, "HBD") +
+                       bytes([0]) + struct.pack("<I", 1))
+        self._assert_ops_fails("malformed operation", _ops_tx([vests_order]),
+                               path=active)
+        # Same symbol on both sides of an order is a no-op trade that still fills.
+        same_symbol = _op_limit_order_create("kktrader", 1, 100, "HIVE",
+                                             100, "HIVE", False, 1)
+        self._assert_ops_fails("symbols must differ", _ops_tx([same_symbol]),
+                               path=active)
+
+    def test_hive_sign_ops_comment_options_binds_to_its_comment(self):
+        """comment_options redirects a post's payout. Detached from its
+        comment it could retarget a post published earlier that the user is
+        not reviewing on screen, so firmware requires it to follow one with a
+        matching author and permlink."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        comment = _op_comment("", "hive-100", "kkauthor", "my-post",
+                              "Title", b"Body", "{}")
+        options = _op_comment_options("kkauthor", "my-post", 1000000, 10000)
+
+        # Standing alone: refused.
+        self._assert_ops_fails("must follow its comment", _ops_tx([options]))
+        # Following a comment for a DIFFERENT post: refused.
+        other = _op_comment_options("kkauthor", "other-post", 1000000, 10000)
+        self._assert_ops_fails("must follow its comment",
+                               _ops_tx([comment, other]))
+        # Correctly paired, with beneficiaries: signs on the posting key.
+        paired = _ops_tx([comment, _op_comment_options(
+            "kkauthor", "my-post", 1000000, 10000,
+            beneficiaries=[("aaron", 1000), ("zoe", 500)])])
+        self._ops_signs_with(paired, ROLE_POSTING)
+
+    def test_hive_sign_ops_comment_options_beneficiary_rules(self):
+        """hived requires strictly ascending, unique beneficiaries summing to
+        <= 100%; signing anything else only wastes a device confirmation."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        comment = _op_comment("", "hive-100", "kkauthor", "my-post",
+                              "Title", b"Body", "{}")
+        for bens in ([("zoe", 500), ("aaron", 1000)],      # unsorted
+                     [("aaron", 500), ("aaron", 500)],     # duplicate
+                     [("aaron", 6000), ("zoe", 5000)]):    # > 100%
+            tx = _ops_tx([comment, _op_comment_options(
+                "kkauthor", "my-post", 1000000, 10000, beneficiaries=bens)])
+            self._assert_ops_fails("beneficiaries", tx)
+
+    def test_hive_sign_ops_account_update2_rejects_authority_change(self):
+        """account_update2 can rotate account keys. Only the profile-metadata
+        form is in the table — the same device-derived-keys invariant that
+        keeps ops 9/10 out, applied field-level."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        self._assert_ops_fails(
+            "authority changes",
+            _ops_tx([_op_account_update2("kkuser", '{"profile":{}}', "",
+                                         authority_present=True)]),
+            path=hive_path(ROLE_ACTIVE))
+
+        # json_metadata is an active-key field...
+        self._ops_signs_with(
+            _ops_tx([_op_account_update2("kkuser", '{"profile":{}}', "")]),
+            ROLE_ACTIVE)
+        # ...while a posting-metadata-only profile edit stays posting tier.
+        self._ops_signs_with(
+            _ops_tx([_op_account_update2("kkuser", "", '{"profile":{}}')]),
+            ROLE_POSTING)
+
+    def test_hive_sign_ops_truncated_bodies_rejected(self):
+        """The signature covers the whole buffer, so a short read would mean
+        signing bytes the device never displayed. Every truncation must be
+        refused rather than partially parsed."""
+        self.requires_firmware("7.15.0")
+        self.requires_message("HiveSignOperations")
+        self.setup_mnemonic_nopin_nopassphrase()
+
+        for op in (_op_limit_order_create("kktrader", 1, 100, "HIVE", 50,
+                                          "HBD", False, 9),
+                   _op_claim_reward_balance("kkuser", 1, 1, 1),
+                   _op_transfer_from_savings("kkuser", 7, "kkfriend", 1500,
+                                             "HBD", "memo")):
+            # One byte short is the boundary case; a deeper cut exercises the
+            # length-prefixed string readers.
+            for cut in (1, 5):
+                if cut >= len(op):
+                    continue
+                self._assert_ops_fails(None, _ops_tx([op[:-cut]]),
+                                       path=hive_path(ROLE_ACTIVE))
+
     def test_hive_sign_message_rejects_chain_id_prefix(self):
         """A 'message' that begins with the mainnet chain id would hash to a
         broadcastable TRANSACTION digest (tx digest = SHA256(chain_id || tx)).