Port sign_message_by_id

Author: Quexington

chia/_tests/cmds/cmd_test_utils.py

@@ -48,6 +48,8 @@
     SendTransactionMultiResponse,
     SignMessageByAddress,
     SignMessageByAddressResponse,
+    SignMessageByID,
+    SignMessageByIDResponse,
     WalletInfoResponse,
 )
 from chia.wallet.wallet_rpc_client import WalletRpcClient
@@ -167,12 +169,21 @@ async def sign_message_by_address(self, request: SignMessageByAddress) -> SignMe
         signing_mode = SigningMode.CHIP_0002.value
         return SignMessageByAddressResponse(pubkey, signature, signing_mode)
 
-    async def sign_message_by_id(self, id: str, message: str) -> tuple[str, str, str]:
-        self.add_to_log("sign_message_by_id", (id, message))
-        pubkey = bytes([4] * 48).hex()
-        signature = bytes([7] * 576).hex()
+    async def sign_message_by_id(self, request: SignMessageByID) -> SignMessageByIDResponse:
+        self.add_to_log("sign_message_by_id", (request.id, request.message))
+        pubkey = G1Element.from_bytes(
+            bytes.fromhex(
+                "a9e652cb551d5978a9ee4b7aa52a4e826078a54b08a3d903c38611cb8a804a9a29c926e4f8549314a079e04ecde10cc1"
+            )
+        )
+        signature = G2Element.from_bytes(
+            bytes.fromhex(
+                "c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+                "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+            )
+        )
         signing_mode = SigningMode.CHIP_0002.value
-        return pubkey, signature, signing_mode
+        return SignMessageByIDResponse(pubkey, signature, bytes32.zeros, signing_mode)
 
     async def cat_asset_id_to_name(self, asset_id: bytes32) -> Optional[tuple[Optional[uint32], str]]:
         """

chia/_tests/cmds/wallet/test_did.py

@@ -116,8 +116,9 @@ def test_did_sign_message(capsys: object, get_test_cli_clients: tuple[TestRpcCli
     # these are various things that should be in the output
     assert_list = [
         f"Message: {message.hex()}",
-        f"Public Key: {bytes([4] * 48).hex()}",
-        f"Signature: {bytes([7] * 576).hex()}",
+        "Public Key: a9e652cb551d5978a9ee4b7aa52a4e826078a54b08a3d903c38611cb8a804a9a29c926e4f8549314a079e04ecde10cc1",
+        "Signature: c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+        "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
         f"Signing Mode: {SigningMode.CHIP_0002.value}",
     ]
     run_cli_command_and_assert(capsys, root_dir, [*command_args, f"-i{did_id}"], assert_list)

chia/_tests/wallet/nft_wallet/test_nft_wallet.py

@@ -5,7 +5,7 @@
 from typing import Any, Callable
 
 import pytest
-from chia_rs import AugSchemeMPL, G1Element, G2Element
+from chia_rs import AugSchemeMPL
 from chia_rs.sized_bytes import bytes32
 from chia_rs.sized_ints import uint16, uint32, uint64
 from clvm_tools.binutils import disassemble
@@ -41,6 +41,7 @@
     NFTTransferBulk,
     NFTTransferNFT,
     NFTWalletWithDID,
+    SignMessageByID,
 )
 from chia.wallet.wallet_rpc_api import MAX_NFT_CHUNK_SIZE
 from chia.wallet.wallet_state_manager import WalletStateManager
@@ -2757,51 +2758,55 @@ async def test_nft_sign_message(wallet_environments: WalletTestFramework) -> Non
     assert not coin.pending_transaction
     # Test general string
     message = "Hello World"
-    pubkey, sig, _ = await env.rpc_client.sign_message_by_id(
-        id=encode_puzzle_hash(coin.launcher_id, AddressType.NFT.value), message=message
+    sign_by_id_res = await env.rpc_client.sign_message_by_id(
+        SignMessageByID(id=encode_puzzle_hash(coin.launcher_id, AddressType.NFT.value), message=message)
     )
     puzzle = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, message))
     assert AugSchemeMPL.verify(
-        G1Element.from_bytes(bytes.fromhex(pubkey)),
+        sign_by_id_res.pubkey,
         puzzle.get_tree_hash(),
-        G2Element.from_bytes(bytes.fromhex(sig)),
+        sign_by_id_res.signature,
     )
     # Test hex string
     message = "0123456789ABCDEF"
-    pubkey, sig, _ = await env.rpc_client.sign_message_by_id(
-        id=encode_puzzle_hash(coin.launcher_id, AddressType.NFT.value), message=message, is_hex=True
+    sign_by_id_res = await env.rpc_client.sign_message_by_id(
+        SignMessageByID(id=encode_puzzle_hash(coin.launcher_id, AddressType.NFT.value), message=message, is_hex=True)
     )
     puzzle = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, bytes.fromhex(message)))
     assert AugSchemeMPL.verify(
-        G1Element.from_bytes(bytes.fromhex(pubkey)),
+        sign_by_id_res.pubkey,
         puzzle.get_tree_hash(),
-        G2Element.from_bytes(bytes.fromhex(sig)),
+        sign_by_id_res.signature,
     )
     # Test BLS sign string
     message = "Hello World"
-    pubkey, sig, _ = await env.rpc_client.sign_message_by_id(
-        id=encode_puzzle_hash(coin.launcher_id, AddressType.NFT.value),
-        message=message,
-        is_hex=False,
-        safe_mode=False,
+    sign_by_id_res = await env.rpc_client.sign_message_by_id(
+        SignMessageByID(
+            id=encode_puzzle_hash(coin.launcher_id, AddressType.NFT.value),
+            message=message,
+            is_hex=False,
+            safe_mode=False,
+        )
     )
 
     assert AugSchemeMPL.verify(
-        G1Element.from_bytes(bytes.fromhex(pubkey)),
-        bytes(message, "utf-8"),
-        G2Element.from_bytes(bytes.fromhex(sig)),
+        sign_by_id_res.pubkey,
+        puzzle.get_tree_hash(),
+        sign_by_id_res.signature,
     )
     # Test BLS sign hex
     message = "0123456789ABCDEF"
-    pubkey, sig, _ = await env.rpc_client.sign_message_by_id(
-        id=encode_puzzle_hash(coin.launcher_id, AddressType.NFT.value),
-        message=message,
-        is_hex=True,
-        safe_mode=False,
+    sign_by_id_res = await env.rpc_client.sign_message_by_id(
+        SignMessageByID(
+            id=encode_puzzle_hash(coin.launcher_id, AddressType.NFT.value),
+            message=message,
+            is_hex=True,
+            safe_mode=False,
+        )
     )
 
     assert AugSchemeMPL.verify(
-        G1Element.from_bytes(bytes.fromhex(pubkey)),
-        bytes.fromhex(message),
-        G2Element.from_bytes(bytes.fromhex(sig)),
+        sign_by_id_res.pubkey,
+        puzzle.get_tree_hash(),
+        sign_by_id_res.signature,
     )

chia/cmds/wallet_funcs.py

@@ -74,6 +74,9 @@
     RoyaltyAsset,
     SendTransactionResponse,
     SignMessageByAddress,
+    SignMessageByAddressResponse,
+    SignMessageByID,
+    SignMessageByIDResponse,
     VCAddProofs,
     VCGet,
     VCGetList,
@@ -1627,30 +1630,27 @@ async def sign_message(
             if address is None:
                 print("Address is required for XCH address type.")
                 return
-            response = await wallet_client.sign_message_by_address(
-                SignMessageByAddress(address.original_address, message)
-            )
-            pubkey = str(response.pubkey)
-            signature = str(response.signature)
-            signing_mode = response.signing_mode
+            response: Union[
+                SignMessageByAddressResponse, SignMessageByIDResponse
+            ] = await wallet_client.sign_message_by_address(SignMessageByAddress(address.original_address, message))
         elif addr_type == AddressType.DID:
             if did_id is None:
                 print("DID id is required for DID address type.")
                 return
-            pubkey, signature, signing_mode = await wallet_client.sign_message_by_id(did_id.original_address, message)
+            response = await wallet_client.sign_message_by_id(SignMessageByID(did_id.original_address, message))
         elif addr_type == AddressType.NFT:
             if nft_id is None:
                 print("NFT id is required for NFT address type.")
                 return
-            pubkey, signature, signing_mode = await wallet_client.sign_message_by_id(nft_id.original_address, message)
+            response = await wallet_client.sign_message_by_id(SignMessageByID(nft_id.original_address, message))
         else:
             print("Invalid wallet type.")
             return
         print("")
         print(f"Message: {message}")
-        print(f"Public Key: {pubkey}")
-        print(f"Signature: {signature}")
-        print(f"Signing Mode: {signing_mode}")
+        print(f"Public Key: {response.pubkey!s}")
+        print(f"Signature: {response.signature!s}")
+        print(f"Signing Mode: {response.signing_mode}")
 
 
 async def spend_clawback(

chia/wallet/wallet_request_types.py

@@ -393,6 +393,24 @@ class SignMessageByAddressResponse(Streamable):
     signing_mode: str
 
 
+@streamable
+@dataclass(frozen=True)
+class SignMessageByID(Streamable):
+    id: str
+    message: str
+    is_hex: bool = False
+    safe_mode: bool = True
+
+
+@streamable
+@dataclass(frozen=True)
+class SignMessageByIDResponse(Streamable):
+    pubkey: G1Element
+    signature: G2Element
+    latest_coin_id: bytes32
+    signing_mode: str
+
+
 @streamable
 @dataclass(frozen=True)
 class GetTransactionMemo(Streamable):

chia/wallet/wallet_rpc_api.py

@@ -245,6 +245,8 @@
     SetWalletResyncOnStartup,
     SignMessageByAddress,
     SignMessageByAddressResponse,
+    SignMessageByID,
+    SignMessageByIDResponse,
     SplitCoins,
     SplitCoinsResponse,
     SubmitTransactions,
@@ -2047,28 +2049,23 @@ async def sign_message_by_address(self, request: SignMessageByAddress) -> SignMe
             signing_mode=mode.value,
         )
 
-    async def sign_message_by_id(self, request: dict[str, Any]) -> EndpointResult:
+    @marshal
+    async def sign_message_by_id(self, request: SignMessageByID) -> SignMessageByIDResponse:
         """
         Given a NFT/DID ID, sign the message by the P2 private key.
         :param request:
         :return:
         """
-        entity_id: bytes32 = decode_puzzle_hash(request["id"])
+        entity_id: bytes32 = decode_puzzle_hash(request.id)
         selected_wallet: Optional[WalletProtocol[Any]] = None
-        is_hex: bool = request.get("is_hex", False)
-        if isinstance(is_hex, str):
-            is_hex = True if is_hex.lower() == "true" else False
-        safe_mode: bool = request.get("safe_mode", True)
-        if isinstance(safe_mode, str):
-            safe_mode = True if safe_mode.lower() == "true" else False
         mode: SigningMode = SigningMode.CHIP_0002
-        if is_hex and safe_mode:
+        if request.is_hex and request.safe_mode:
             mode = SigningMode.CHIP_0002_HEX_INPUT
-        elif not is_hex and not safe_mode:
+        elif not request.is_hex and not request.safe_mode:
             mode = SigningMode.BLS_MESSAGE_AUGMENTATION_UTF8_INPUT
-        elif is_hex and not safe_mode:
+        elif request.is_hex and not request.safe_mode:
             mode = SigningMode.BLS_MESSAGE_AUGMENTATION_HEX_INPUT
-        if is_valid_address(request["id"], {AddressType.DID}, self.service.config):
+        if is_valid_address(request.id, {AddressType.DID}, self.service.config):
             for wallet in self.service.wallet_state_manager.wallets.values():
                 if wallet.type() == WalletType.DECENTRALIZED_ID.value:
                     assert isinstance(wallet, DIDWallet)
@@ -2077,11 +2074,11 @@ async def sign_message_by_id(self, request: dict[str, Any]) -> EndpointResult:
                         selected_wallet = wallet
                         break
             if selected_wallet is None:
-                return {"success": False, "error": f"DID for {entity_id.hex()} doesn't exist."}
+                raise ValueError(f"DID for {entity_id.hex()} doesn't exist.")
             assert isinstance(selected_wallet, DIDWallet)
-            pubkey, signature = await selected_wallet.sign_message(request["message"], mode)
+            pubkey, signature = await selected_wallet.sign_message(request.message, mode)
             latest_coin_id = (await selected_wallet.get_coin()).name()
-        elif is_valid_address(request["id"], {AddressType.NFT}, self.service.config):
+        elif is_valid_address(request.id, {AddressType.NFT}, self.service.config):
             target_nft: Optional[NFTCoinInfo] = None
             for wallet in self.service.wallet_state_manager.wallets.values():
                 if wallet.type() == WalletType.NFT.value:
@@ -2092,21 +2089,20 @@ async def sign_message_by_id(self, request: dict[str, Any]) -> EndpointResult:
                         target_nft = nft
                         break
             if selected_wallet is None or target_nft is None:
-                return {"success": False, "error": f"NFT for {entity_id.hex()} doesn't exist."}
+                raise ValueError(f"NFT for {entity_id.hex()} doesn't exist.")
 
             assert isinstance(selected_wallet, NFTWallet)
-            pubkey, signature = await selected_wallet.sign_message(request["message"], target_nft, mode)
+            pubkey, signature = await selected_wallet.sign_message(request.message, target_nft, mode)
             latest_coin_id = target_nft.coin.name()
         else:
-            return {"success": False, "error": f"Unknown ID type, {request['id']}"}
+            raise ValueError(f"Unknown ID type, {request.id}")
 
-        return {
-            "success": True,
-            "pubkey": str(pubkey),
-            "signature": str(signature),
-            "latest_coin_id": latest_coin_id.hex() if latest_coin_id is not None else None,
-            "signing_mode": mode.value,
-        }
+        return SignMessageByIDResponse(
+            pubkey=pubkey,
+            signature=signature,
+            latest_coin_id=latest_coin_id,
+            signing_mode=mode.value,
+        )
 
     ##########################################################################################
     # CATs and Trading

chia/wallet/wallet_rpc_client.py

@@ -162,6 +162,8 @@
     SetWalletResyncOnStartup,
     SignMessageByAddress,
     SignMessageByAddressResponse,
+    SignMessageByID,
+    SignMessageByIDResponse,
     SplitCoins,
     SplitCoinsResponse,
     SubmitTransactions,
@@ -1151,13 +1153,8 @@ async def sign_message_by_address(self, request: SignMessageByAddress) -> SignMe
             await self.fetch("sign_message_by_address", request.to_json_dict())
         )
 
-    async def sign_message_by_id(
-        self, id: str, message: str, is_hex: bool = False, safe_mode: bool = True
-    ) -> tuple[str, str, str]:
-        response = await self.fetch(
-            "sign_message_by_id", {"id": id, "message": message, "is_hex": is_hex, "safe_mode": safe_mode}
-        )
-        return response["pubkey"], response["signature"], response["signing_mode"]
+    async def sign_message_by_id(self, request: SignMessageByID) -> SignMessageByIDResponse:
+        return SignMessageByIDResponse.from_json_dict(await self.fetch("sign_message_by_id", request.to_json_dict()))
 
     async def verify_signature(self, request: VerifySignature) -> VerifySignatureResponse:
         return VerifySignatureResponse.from_json_dict(await self.fetch("verify_signature", {**request.to_json_dict()}))