[LABS-305] Extract signing functionality from wallet RPC to its own library

chia/_tests/wallet/nft_wallet/test_nft_wallet.py

@@ -665,6 +665,124 @@ async def test_nft_wallet_rpc_creation_and_list(wallet_environments: WalletTestF
         await env.rpc_client.count_nfts(NFTCountNFTs(uint32(50)))
 
 
+@pytest.mark.limit_consensus_modes(allowed=[ConsensusMode.PLAIN], reason="irrelevant")
+@pytest.mark.parametrize("wallet_environments", [{"num_environments": 1, "blocks_needed": [1]}], indirect=True)
+@pytest.mark.anyio
+async def test_sign_message_by_nft_id(wallet_environments: WalletTestFramework) -> None:
+    env = wallet_environments.environments[0]
+    wallet_node = env.node
+    wallet = env.xch_wallet
+
+    env.wallet_aliases = {
+        "xch": 1,
+        "nft": 2,
+    }
+
+    nft_wallet_0 = await env.rpc_client.fetch("create_new_wallet", dict(wallet_type="nft_wallet", name="NFT WALLET 1"))
+    assert isinstance(nft_wallet_0, dict)
+    assert nft_wallet_0.get("success")
+    assert env.wallet_aliases["nft"] == nft_wallet_0["wallet_id"]
+
+    async with wallet.wallet_state_manager.new_action_scope(wallet_environments.tx_config, push=True) as action_scope:
+        wallet_ph = await action_scope.get_puzzle_hash(wallet.wallet_state_manager)
+    await env.rpc_client.mint_nft(
+        request=NFTMintNFTRequest(
+            wallet_id=uint32(env.wallet_aliases["nft"]),
+            royalty_address=encode_puzzle_hash(wallet_ph, AddressType.NFT.hrp(wallet_node.config)),
+            target_address=None,
+            hash=bytes32.from_hexstr("0xD4584AD463139FA8C0D9F68F4B59F185D4584AD463139FA8C0D9F68F4B59F185"),
+            uris=["https://www.chia.net/img/branding/chia-logo.svg"],
+            push=True,
+        ),
+        tx_config=wallet_environments.tx_config,
+    )
+
+    await wallet_environments.process_pending_states(
+        [
+            WalletStateTransition(
+                pre_block_balance_updates={
+                    "xch": {"set_remainder": True},  # tested above
+                    "nft": {"init": True, "pending_coin_removal_count": 1},
+                },
+                post_block_balance_updates={
+                    "xch": {"set_remainder": True},  # tested above
+                    "nft": {
+                        "pending_coin_removal_count": -1,
+                        "unspent_coin_count": 1,
+                    },
+                },
+            )
+        ]
+    )
+
+    nft_list = await env.rpc_client.list_nfts(NFTGetNFTs(uint32(env.wallet_aliases["nft"])))
+    nft_id = nft_list.nft_list[0].nft_id
+
+    # Test general string
+    message = "Hello World"
+    response = await env.rpc_client.sign_message_by_id(
+        SignMessageByID(
+            id=nft_id,
+            message=message,
+        )
+    )
+    puzzle: Program = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, message))
+    assert AugSchemeMPL.verify(
+        response.pubkey,
+        puzzle.get_tree_hash(),
+        response.signature,
+    )
+    # Test hex string
+    message = "0123456789ABCDEF"
+    response = await env.rpc_client.sign_message_by_id(
+        SignMessageByID(
+            id=nft_id,
+            message=message,
+            is_hex=True,
+        )
+    )
+    puzzle = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, bytes.fromhex(message)))
+
+    assert AugSchemeMPL.verify(
+        response.pubkey,
+        puzzle.get_tree_hash(),
+        response.signature,
+    )
+
+    # Test BLS sign string
+    message = "Hello World"
+    response = await env.rpc_client.sign_message_by_id(
+        SignMessageByID(
+            id=nft_id,
+            message=message,
+            is_hex=False,
+            safe_mode=False,
+        )
+    )
+
+    assert AugSchemeMPL.verify(
+        response.pubkey,
+        bytes(message, "utf-8"),
+        response.signature,
+    )
+    # Test BLS sign hex
+    message = "0123456789ABCDEF"
+    response = await env.rpc_client.sign_message_by_id(
+        SignMessageByID(
+            id=nft_id,
+            message=message,
+            is_hex=True,
+            safe_mode=False,
+        )
+    )
+
+    assert AugSchemeMPL.verify(
+        response.pubkey,
+        hexstr_to_bytes(message),
+        response.signature,
+    )
+
+
 @pytest.mark.limit_consensus_modes(allowed=[ConsensusMode.PLAIN], reason="irrelevant")
 @pytest.mark.parametrize("wallet_environments", [{"num_environments": 1, "blocks_needed": [1]}], indirect=True)
 @pytest.mark.anyio

chia/_tests/wallet/rpc/test_wallet_rpc.py

@@ -154,6 +154,7 @@
     SendTransaction,
     SendTransactionMulti,
     SetWalletResyncOnStartup,
+    SignMessageByAddress,
     SpendClawbackCoins,
     SplitCoins,
     TakeOffer,
@@ -3129,6 +3130,37 @@ async def test_verify_signature(
     assert res == rpc_response
 
 
+@pytest.mark.parametrize(
+    "wallet_environments",
+    [
+        {
+            "num_environments": 1,
+            "blocks_needed": [1],
+            "reuse_puzhash": True,
+            "trusted": True,
+        }
+    ],
+    indirect=True,
+)
+@pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="irrelevant")
+async def test_sign_message_by_address(wallet_environments: WalletTestFramework) -> None:
+    client: WalletRpcClient = wallet_environments.environments[0].rpc_client
+
+    message = "foo"
+    address = await client.get_next_address(GetNextAddress(uint32(1)))
+    signed_message = await client.sign_message_by_address(SignMessageByAddress(address.address, message))
+
+    await wallet_environments.environments[0].rpc_client.verify_signature(
+        VerifySignature(
+            message=message,
+            pubkey=signed_message.pubkey,
+            signature=signed_message.signature,
+            signing_mode=signed_message.signing_mode,
+        )
+    )
+
+
 @pytest.mark.parametrize(
     "wallet_environments",
     [{"num_environments": 2, "blocks_needed": [1, 0]}],

chia/wallet/did_wallet/did_wallet.py

@@ -16,7 +16,6 @@
 from chia.types.blockchain_format.coin import Coin
 from chia.types.blockchain_format.program import Program
 from chia.types.coin_spend import make_spend
-from chia.types.signing_mode import CHIP_0002_SIGN_MESSAGE_PREFIX, SigningMode
 from chia.wallet.conditions import (
     AssertCoinAnnouncement,
     Condition,
@@ -30,8 +29,6 @@
 from chia.wallet.did_wallet.did_wallet_puzzles import match_did_puzzle, uncurry_innerpuz
 from chia.wallet.lineage_proof import LineageProof
 from chia.wallet.puzzles.p2_delegated_puzzle_or_hidden_puzzle import (
-    DEFAULT_HIDDEN_PUZZLE_HASH,
-    calculate_synthetic_secret_key,
     puzzle_for_pk,
     puzzle_hash_for_pk,
 )
@@ -898,27 +895,15 @@ def get_parent_for_coin(self, coin) -> LineageProof | None:
 
         return parent_info
 
-    async def sign_message(self, message: str, mode: SigningMode) -> tuple[G1Element, G2Element]:
+    async def current_p2_puzzle_hash(self) -> bytes32:
         if self.did_info.current_inner is None:
             raise ValueError("Missing DID inner puzzle.")
         puzzle_args = did_wallet_puzzles.uncurry_innerpuz(self.did_info.current_inner)
         if puzzle_args is not None:
             p2_puzzle, _, _, _, _ = puzzle_args
-            puzzle_hash = p2_puzzle.get_tree_hash()
-            private = await self.wallet_state_manager.get_private_key(puzzle_hash)
-            synthetic_secret_key = calculate_synthetic_secret_key(private, DEFAULT_HIDDEN_PUZZLE_HASH)
-            synthetic_pk = synthetic_secret_key.get_g1()
-            if mode == SigningMode.CHIP_0002_HEX_INPUT:
-                hex_message: bytes = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, bytes.fromhex(message))).get_tree_hash()
-            elif mode == SigningMode.BLS_MESSAGE_AUGMENTATION_UTF8_INPUT:
-                hex_message = bytes(message, "utf-8")
-            elif mode == SigningMode.BLS_MESSAGE_AUGMENTATION_HEX_INPUT:
-                hex_message = bytes.fromhex(message)
-            else:
-                hex_message = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, message)).get_tree_hash()
-            return synthetic_pk, AugSchemeMPL.sign(synthetic_secret_key, hex_message)
+            return p2_puzzle.get_tree_hash()
         else:
-            raise ValueError("Invalid inner DID puzzle.")
+            raise ValueError("Invalid DID inner puzzle.")
 
     async def generate_new_decentralised_id(
         self,

chia/wallet/nft_wallet/nft_wallet.py

@@ -14,7 +14,6 @@
 from chia.types.blockchain_format.coin import Coin
 from chia.types.blockchain_format.program import Program
 from chia.types.coin_spend import make_spend
-from chia.types.signing_mode import CHIP_0002_SIGN_MESSAGE_PREFIX, SigningMode
 from chia.util.casts import int_from_bytes, int_to_bytes
 from chia.util.hash import std_hash
 from chia.wallet.conditions import (
@@ -38,8 +37,6 @@
 from chia.wallet.outer_puzzles import AssetType, construct_puzzle, match_puzzle, solve_puzzle
 from chia.wallet.puzzle_drivers import PuzzleInfo, Solver
 from chia.wallet.puzzles.p2_delegated_puzzle_or_hidden_puzzle import (
-    DEFAULT_HIDDEN_PUZZLE_HASH,
-    calculate_synthetic_secret_key,
     puzzle_for_pk,
 )
 from chia.wallet.singleton import SINGLETON_LAUNCHER_PUZZLE, SINGLETON_LAUNCHER_PUZZLE_HASH, create_singleton_puzzle
@@ -497,23 +494,11 @@ async def get_puzzle_info(self, nft_id: bytes32) -> PuzzleInfo:
         else:
             return puzzle_info
 
-    async def sign_message(self, message: str, nft: NFTCoinInfo, mode: SigningMode) -> tuple[G1Element, G2Element]:
+    async def current_p2_puzzle_hash(self, nft: NFTCoinInfo) -> bytes32:
         uncurried_nft = UncurriedNFT.uncurry(*nft.full_puzzle.uncurry())
         if uncurried_nft is not None:
             p2_puzzle = uncurried_nft.p2_puzzle
-            puzzle_hash = p2_puzzle.get_tree_hash()
-            private = await self.wallet_state_manager.get_private_key(puzzle_hash)
-            synthetic_secret_key = calculate_synthetic_secret_key(private, DEFAULT_HIDDEN_PUZZLE_HASH)
-            synthetic_pk = synthetic_secret_key.get_g1()
-            if mode == SigningMode.CHIP_0002_HEX_INPUT:
-                hex_message: bytes = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, bytes.fromhex(message))).get_tree_hash()
-            elif mode == SigningMode.BLS_MESSAGE_AUGMENTATION_UTF8_INPUT:
-                hex_message = bytes(message, "utf-8")
-            elif mode == SigningMode.BLS_MESSAGE_AUGMENTATION_HEX_INPUT:
-                hex_message = bytes.fromhex(message)
-            else:
-                hex_message = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, message)).get_tree_hash()
-            return synthetic_pk, AugSchemeMPL.sign(synthetic_secret_key, hex_message)
+            return p2_puzzle.get_tree_hash()
         else:
             raise ValueError("Invalid NFT puzzle.")
 

chia/wallet/util/signing.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from chia_rs import AugSchemeMPL, G1Element, G2Element, PrivateKey
+from chia_rs.sized_bytes import bytes32
+
+from chia.types.blockchain_format.program import Program
+from chia.types.signing_mode import CHIP_0002_SIGN_MESSAGE_PREFIX, SigningMode
+from chia.util.bech32m import decode_puzzle_hash
+from chia.util.byte_types import hexstr_to_bytes
+from chia.wallet.puzzles import p2_delegated_conditions
+from chia.wallet.puzzles.p2_delegated_puzzle_or_hidden_puzzle import puzzle_hash_for_synthetic_public_key
+from chia.wallet.wallet_request_types import VerifySignatureResponse
+
+# CHIP-0002 message signing as documented at:
+# https://github.com/Chia-Network/chips/blob/80e4611fe52b174bf1a0382b9dff73805b18b8c6/CHIPs/chip-0002.md
+
+
+def verify_signature(
+    *, signing_mode: SigningMode, public_key: G1Element, message: str, signature: G2Element, address: str | None
+) -> VerifySignatureResponse:
+    """
+    Given a public key, message and signature, verify if it is valid.
+    :param request:
+    :return:
+    """
+    if signing_mode in {SigningMode.CHIP_0002, SigningMode.CHIP_0002_P2_DELEGATED_CONDITIONS}:
+        # CHIP-0002 message signatures are made over the tree hash of:
+        #   ("Chia Signed Message", message)
+        message_to_verify: bytes = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, message)).get_tree_hash()
+    elif signing_mode == SigningMode.BLS_MESSAGE_AUGMENTATION_HEX_INPUT:
+        # Message is expected to be a hex string
+        message_to_verify = hexstr_to_bytes(message)
+    elif signing_mode == SigningMode.BLS_MESSAGE_AUGMENTATION_UTF8_INPUT:
+        # Message is expected to be a UTF-8 string
+        message_to_verify = bytes(message, "utf-8")
+    else:
+        raise ValueError(f"Unsupported signing mode: {signing_mode!r}")
+
+    # Verify using the BLS message augmentation scheme
+    is_valid = AugSchemeMPL.verify(
+        public_key,
+        message_to_verify,
+        signature,
+    )
+    if address is not None:
+        # For signatures made by the sign_message_by_address/sign_message_by_id
+        # endpoints, the "address" field should contain the p2_address of the NFT/DID
+        # that was used to sign the message.
+        puzzle_hash: bytes32 = decode_puzzle_hash(address)
+        expected_puzzle_hash: bytes32 | None = None
+        if signing_mode == SigningMode.CHIP_0002_P2_DELEGATED_CONDITIONS:
+            puzzle = p2_delegated_conditions.puzzle_for_pk(Program.to(public_key))
+            expected_puzzle_hash = bytes32(puzzle.get_tree_hash())
+        else:
+            expected_puzzle_hash = puzzle_hash_for_synthetic_public_key(public_key)
+        if puzzle_hash != expected_puzzle_hash:
+            return VerifySignatureResponse(isValid=False, error="Public key doesn't match the address")
+    if is_valid:
+        return VerifySignatureResponse(isValid=is_valid)
+    else:
+        return VerifySignatureResponse(isValid=False, error="Signature is invalid.")
+
+
+@dataclass(kw_only=True, frozen=True)
+class SignMessageResponse:
+    pubkey: G1Element
+    signature: G2Element
+
+
+def sign_message(secret_key: PrivateKey, message: str, mode: SigningMode) -> SignMessageResponse:
+    public_key = secret_key.get_g1()
+    if mode == SigningMode.CHIP_0002_HEX_INPUT:
+        hex_message: bytes = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, bytes.fromhex(message))).get_tree_hash()
+    elif mode == SigningMode.BLS_MESSAGE_AUGMENTATION_UTF8_INPUT:
+        hex_message = bytes(message, "utf-8")
+    elif mode == SigningMode.BLS_MESSAGE_AUGMENTATION_HEX_INPUT:
+        hex_message = bytes.fromhex(message)
+    else:
+        hex_message = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, message)).get_tree_hash()
+    return SignMessageResponse(
+        pubkey=public_key,
+        signature=AugSchemeMPL.sign(secret_key, hex_message),
+    )

chia/wallet/wallet.py

@@ -12,7 +12,6 @@
 from chia.types.blockchain_format.program import Program
 from chia.types.blockchain_format.serialized_program import SerializedProgram
 from chia.types.coin_spend import make_spend
-from chia.types.signing_mode import CHIP_0002_SIGN_MESSAGE_PREFIX, SigningMode
 from chia.util.hash import std_hash
 from chia.wallet.coin_selection import select_coins
 from chia.wallet.conditions import (
@@ -112,6 +111,9 @@ def type(cls) -> WalletType:
     def id(self) -> uint32:
         return self.wallet_id
 
+    def convert_secret_key_to_synthetic(self, secret_key: PrivateKey) -> PrivateKey:
+        return calculate_synthetic_secret_key(secret_key, DEFAULT_HIDDEN_PUZZLE_HASH)
+
     async def get_confirmed_balance(self, record_list: set[WalletCoinRecord] | None = None) -> uint128:
         return await self.wallet_state_manager.get_confirmed_balance_for_wallet(self.id(), record_list)
 
@@ -361,22 +363,6 @@ async def _generate_unsigned_transaction(
         self.log.debug(f"Spends is {spends}")
         return spends
 
-    async def sign_message(self, message: str, puzzle_hash: bytes32, mode: SigningMode) -> tuple[G1Element, G2Element]:
-        # CHIP-0002 message signing as documented at:
-        # https://github.com/Chia-Network/chips/blob/80e4611fe52b174bf1a0382b9dff73805b18b8c6/CHIPs/chip-0002.md#signmessage
-        private = await self.wallet_state_manager.get_private_key(puzzle_hash)
-        synthetic_secret_key = calculate_synthetic_secret_key(private, DEFAULT_HIDDEN_PUZZLE_HASH)
-        synthetic_pk = synthetic_secret_key.get_g1()
-        if mode == SigningMode.CHIP_0002_HEX_INPUT:
-            hex_message: bytes = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, bytes.fromhex(message))).get_tree_hash()
-        elif mode == SigningMode.BLS_MESSAGE_AUGMENTATION_UTF8_INPUT:
-            hex_message = bytes(message, "utf-8")
-        elif mode == SigningMode.BLS_MESSAGE_AUGMENTATION_HEX_INPUT:
-            hex_message = bytes.fromhex(message)
-        else:
-            hex_message = Program.to((CHIP_0002_SIGN_MESSAGE_PREFIX, message)).get_tree_hash()
-        return synthetic_pk, AugSchemeMPL.sign(synthetic_secret_key, hex_message)
-
     async def generate_signed_transaction(
         self,
         amounts: list[uint64],