Port sign_message_by_address

Author: Quexington

chia/_tests/cmds/cmd_test_utils.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 from typing import Any, Optional, cast
 
-from chia_rs import BlockRecord, Coin, G2Element
+from chia_rs import BlockRecord, Coin, G1Element, G2Element
 from chia_rs.sized_bytes import bytes32
 from chia_rs.sized_ints import uint8, uint16, uint32, uint64
 
@@ -46,6 +46,8 @@
     NFTGetInfo,
     NFTGetInfoResponse,
     SendTransactionMultiResponse,
+    SignMessageByAddress,
+    SignMessageByAddressResponse,
     WalletInfoResponse,
 )
 from chia.wallet.wallet_rpc_client import WalletRpcClient
@@ -149,12 +151,12 @@ async def get_cat_name(self, wallet_id: int) -> str:
         self.add_to_log("get_cat_name", (wallet_id,))
         return "test" + str(wallet_id)
 
-    async def sign_message_by_address(self, address: str, message: str) -> tuple[str, str, str]:
-        self.add_to_log("sign_message_by_address", (address, message))
-        pubkey = bytes([3] * 48).hex()
-        signature = bytes([6] * 576).hex()
+    async def sign_message_by_address(self, request: SignMessageByAddress) -> SignMessageByAddressResponse:
+        self.add_to_log("sign_message_by_address", (request.address, request.message))
+        pubkey = G1Element.from_bytes(bytes([3] * 48))
+        signature = G2Element.from_bytes(bytes([6] * 576))
         signing_mode = SigningMode.CHIP_0002.value
-        return pubkey, signature, signing_mode
+        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))

chia/cmds/wallet_funcs.py

@@ -73,6 +73,7 @@
     NFTTransferNFT,
     RoyaltyAsset,
     SendTransactionResponse,
+    SignMessageByAddress,
     VCAddProofs,
     VCGet,
     VCGetList,
@@ -1626,9 +1627,12 @@ async def sign_message(
             if address is None:
                 print("Address is required for XCH address type.")
                 return
-            pubkey, signature, signing_mode = await wallet_client.sign_message_by_address(
-                address.original_address, message
+            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
         elif addr_type == AddressType.DID:
             if did_id is None:
                 print("DID id is required for DID address type.")

chia/wallet/wallet_request_types.py

@@ -376,6 +376,23 @@ class VerifySignatureResponse(Streamable):
     error: Optional[str] = None
 
 
+@streamable
+@dataclass(frozen=True)
+class SignMessageByAddress(Streamable):
+    address: str
+    message: str
+    is_hex: bool = False
+    safe_mode: bool = True
+
+
+@streamable
+@dataclass(frozen=True)
+class SignMessageByAddressResponse(Streamable):
+    pubkey: G1Element
+    signature: G2Element
+    signing_mode: str
+
+
 @streamable
 @dataclass(frozen=True)
 class GetTransactionMemo(Streamable):

chia/wallet/wallet_rpc_api.py

@@ -243,6 +243,8 @@
     PWStatus,
     PWStatusResponse,
     SetWalletResyncOnStartup,
+    SignMessageByAddress,
+    SignMessageByAddressResponse,
     SplitCoins,
     SplitCoinsResponse,
     SubmitTransactions,
@@ -2021,35 +2023,29 @@ async def verify_signature(self, request: VerifySignature) -> VerifySignatureRes
         else:
             return VerifySignatureResponse(isValid=False, error="Signature is invalid.")
 
-    async def sign_message_by_address(self, request: dict[str, Any]) -> EndpointResult:
+    @marshal
+    async def sign_message_by_address(self, request: SignMessageByAddress) -> SignMessageByAddressResponse:
         """
         Given a derived P2 address, sign the message by its private key.
         :param request:
         :return:
         """
-        puzzle_hash: bytes32 = decode_puzzle_hash(request["address"])
-        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
+        puzzle_hash: bytes32 = decode_puzzle_hash(request.address)
         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
         pubkey, signature = await self.service.wallet_state_manager.main_wallet.sign_message(
-            request["message"], puzzle_hash, mode
+            request.message, puzzle_hash, mode
+        )
+        return SignMessageByAddressResponse(
+            pubkey=pubkey,
+            signature=signature,
+            signing_mode=mode.value,
         )
-        return {
-            "success": True,
-            "pubkey": str(pubkey),
-            "signature": str(signature),
-            "signing_mode": mode.value,
-        }
 
     async def sign_message_by_id(self, request: dict[str, Any]) -> EndpointResult:
         """

chia/wallet/wallet_rpc_client.py

@@ -160,6 +160,8 @@
     SendTransactionMultiResponse,
     SendTransactionResponse,
     SetWalletResyncOnStartup,
+    SignMessageByAddress,
+    SignMessageByAddressResponse,
     SplitCoins,
     SplitCoinsResponse,
     SubmitTransactions,
@@ -1144,9 +1146,10 @@ async def send_notification(
         )
         return TransactionRecord.from_json_dict(response["tx"])
 
-    async def sign_message_by_address(self, address: str, message: str) -> tuple[str, str, str]:
-        response = await self.fetch("sign_message_by_address", {"address": address, "message": message})
-        return response["pubkey"], response["signature"], response["signing_mode"]
+    async def sign_message_by_address(self, request: SignMessageByAddress) -> SignMessageByAddressResponse:
+        return SignMessageByAddressResponse.from_json_dict(
+            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