mirror of
https://github.com/Chia-Network/chia-blockchain.git
synced 2026-08-24 10:05:29 -05:00
[LABS-189] Port chia wallet did to @chia_command (#21114)
* Port `chia wallet notifications` to `@chia_command` framework * Comments by @cursor * Add parsing tests * Fix tests * Add context for AddressParamType * Comments by @cursor * Port `chia wallet vcs` to `@chia_command` framework * Add context for AddressParamType * Remove old CLI test * [LABS-189] Port `chia wallet did` to `@chia_command` * Fix tests * Fix tests * Comments by @cursor * Comments by @cursor * Comments by @cursor * Test coverage * Fix tests * Comments by @cursor * bad merge
This commit is contained in:
Matt Hauff
@@ -1,447 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from chia_rs import G2Element
|
||||
from chia_rs.sized_bytes import bytes32, bytes48
|
||||
from chia_rs.sized_ints import uint16, uint32, uint64
|
||||
|
||||
from chia._tests.cmds.cmd_test_utils import TestRpcClients, TestWalletRpcClient, logType, run_cli_command_and_assert
|
||||
from chia._tests.cmds.wallet.test_consts import FINGERPRINT_ARG, STD_TX, STD_UTX, get_bytes32
|
||||
from chia.types.blockchain_format.program import Program
|
||||
from chia.types.signing_mode import SigningMode
|
||||
from chia.util.bech32m import encode_puzzle_hash
|
||||
from chia.wallet.conditions import Condition, ConditionValidTimes, CreateCoinAnnouncement, CreatePuzzleAnnouncement
|
||||
from chia.wallet.did_wallet.did_info import did_recovery_is_nil
|
||||
from chia.wallet.util.curry_and_treehash import NIL_TREEHASH
|
||||
from chia.wallet.util.tx_config import DEFAULT_TX_CONFIG, TXConfig
|
||||
from chia.wallet.wallet_request_types import (
|
||||
CreateNewWallet,
|
||||
CreateNewWalletType,
|
||||
DIDFindLostDID,
|
||||
DIDFindLostDIDResponse,
|
||||
DIDGetDID,
|
||||
DIDGetDIDResponse,
|
||||
DIDGetInfo,
|
||||
DIDGetInfoResponse,
|
||||
DIDMessageSpend,
|
||||
DIDMessageSpendResponse,
|
||||
DIDSetWalletName,
|
||||
DIDSetWalletNameResponse,
|
||||
DIDTransferDID,
|
||||
DIDTransferDIDResponse,
|
||||
DIDType,
|
||||
DIDUpdateMetadata,
|
||||
DIDUpdateMetadataResponse,
|
||||
)
|
||||
from chia.wallet.wallet_spend_bundle import WalletSpendBundle
|
||||
|
||||
test_condition_valid_times: ConditionValidTimes = ConditionValidTimes(min_time=uint64(100), max_time=uint64(150))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
argnames=["program", "result"],
|
||||
argvalues=[
|
||||
(Program.to(NIL_TREEHASH), True),
|
||||
(Program.NIL, True),
|
||||
(Program.to(bytes32([1] * 32)), False),
|
||||
],
|
||||
)
|
||||
def test_did_recovery_is_nil(program: Program, result: bool) -> None:
|
||||
# test that the alternate wallet nil recovery list bytes are used
|
||||
assert did_recovery_is_nil(program) is result
|
||||
|
||||
|
||||
# DID Commands
|
||||
def test_did_create(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
inst_rpc_client = TestWalletRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
command_args = [
|
||||
"wallet",
|
||||
"did",
|
||||
"create",
|
||||
FINGERPRINT_ARG,
|
||||
"-ntest",
|
||||
"-a3",
|
||||
"-m0.1",
|
||||
"--valid-at",
|
||||
"100",
|
||||
"--expires-at",
|
||||
"150",
|
||||
]
|
||||
# these are various things that should be in the output
|
||||
assert_list = [
|
||||
"Successfully created a DID wallet with name test and id 3 on key 123456",
|
||||
(
|
||||
"Successfully created a DID did:chia:1qgpqyqszqgpqyqszqgpqyqszqgpqyqszqgpqyqszqgpqyqszqgpq4msw0c"
|
||||
" in the newly created DID wallet"
|
||||
),
|
||||
]
|
||||
run_cli_command_and_assert(capsys, root_dir, command_args, assert_list)
|
||||
expected_calls: logType = {
|
||||
"create_new_wallet": [
|
||||
(
|
||||
CreateNewWallet(
|
||||
wallet_type=CreateNewWalletType.DID_WALLET,
|
||||
did_type=DIDType.NEW,
|
||||
amount=uint64(3),
|
||||
wallet_name="test",
|
||||
fee=uint64(100_000_000_000),
|
||||
push=True,
|
||||
),
|
||||
DEFAULT_TX_CONFIG,
|
||||
tuple(),
|
||||
test_condition_valid_times,
|
||||
)
|
||||
],
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
|
||||
|
||||
def test_did_sign_message(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
# set RPC Client
|
||||
inst_rpc_client = TestWalletRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
did_id = encode_puzzle_hash(get_bytes32(1), "did:chia:")
|
||||
message = b"hello did world!!"
|
||||
command_args = ["wallet", "did", "sign_message", FINGERPRINT_ARG, f"-m{message.hex()}"]
|
||||
# these are various things that should be in the output
|
||||
assert_list = [
|
||||
f"Message: {message.hex()}",
|
||||
"Public Key: a9e652cb551d5978a9ee4b7aa52a4e826078a54b08a3d903c38611cb8a804a9a29c926e4f8549314a079e04ecde10cc1",
|
||||
"Signature: c0" + "00" * (42 - 1),
|
||||
f"Signing Mode: {SigningMode.CHIP_0002.value}",
|
||||
]
|
||||
run_cli_command_and_assert(capsys, root_dir, [*command_args, f"-i{did_id}"], assert_list)
|
||||
expected_calls: logType = {
|
||||
"sign_message_by_id": [(did_id, message.hex())], # xch std
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
|
||||
|
||||
def test_did_set_name(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
# set RPC Client
|
||||
class DidSetNameRpcClient(TestWalletRpcClient):
|
||||
async def did_set_wallet_name(self, request: DIDSetWalletName) -> DIDSetWalletNameResponse:
|
||||
self.add_to_log("did_set_wallet_name", (request.wallet_id, request.name))
|
||||
return DIDSetWalletNameResponse(wallet_id=request.wallet_id)
|
||||
|
||||
inst_rpc_client = DidSetNameRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
w_id = 3
|
||||
did_name = "testdid"
|
||||
command_args = ["wallet", "did", "set_name", FINGERPRINT_ARG, f"-i{w_id}", f"-n{did_name}"]
|
||||
# these are various things that should be in the output
|
||||
assert_list = [f"Successfully set a new name for DID wallet with id {w_id}: {did_name}"]
|
||||
run_cli_command_and_assert(capsys, root_dir, command_args, assert_list)
|
||||
expected_calls: logType = {
|
||||
"did_set_wallet_name": [(w_id, did_name)],
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
|
||||
|
||||
def test_did_get_did(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
# set RPC Client
|
||||
class DidGetDidRpcClient(TestWalletRpcClient):
|
||||
async def get_did_id(self, request: DIDGetDID) -> DIDGetDIDResponse:
|
||||
self.add_to_log("get_did_id", (request.wallet_id,))
|
||||
return DIDGetDIDResponse(
|
||||
wallet_id=request.wallet_id,
|
||||
my_did=encode_puzzle_hash(get_bytes32(1), "did:chia:"),
|
||||
coin_id=get_bytes32(2),
|
||||
)
|
||||
|
||||
inst_rpc_client = DidGetDidRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
w_id = 3
|
||||
expected_did = encode_puzzle_hash(get_bytes32(1), "did:chia:")
|
||||
command_args = ["wallet", "did", "get_did", FINGERPRINT_ARG, f"-i{w_id}"]
|
||||
# these are various things that should be in the output
|
||||
assert_list = [f"DID: {expected_did}", f"Coin ID: {get_bytes32(2)}"]
|
||||
run_cli_command_and_assert(capsys, root_dir, command_args, assert_list)
|
||||
expected_calls: logType = {
|
||||
"get_did_id": [(w_id,)],
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
|
||||
|
||||
def test_did_get_details(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
# set RPC Client
|
||||
class DidGetDetailsRpcClient(TestWalletRpcClient):
|
||||
async def get_did_info(self, request: DIDGetInfo) -> DIDGetInfoResponse:
|
||||
self.add_to_log("get_did_info", (request.coin_id, request.latest))
|
||||
response = DIDGetInfoResponse(
|
||||
did_id=encode_puzzle_hash(get_bytes32(2), "did:chia:"),
|
||||
latest_coin=get_bytes32(3),
|
||||
p2_address=encode_puzzle_hash(get_bytes32(4), "xch"),
|
||||
public_key=bytes48([5] * 48),
|
||||
launcher_id=get_bytes32(6),
|
||||
metadata={"did metadata": "yes"},
|
||||
recovery_list_hash=get_bytes32(7),
|
||||
num_verification=uint16(8),
|
||||
full_puzzle=Program.to(9),
|
||||
solution=Program.to(10),
|
||||
hints=[get_bytes32(11), get_bytes32(12)],
|
||||
)
|
||||
return response
|
||||
|
||||
inst_rpc_client = DidGetDetailsRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
did_coin_id_hex = get_bytes32(1).hex()
|
||||
command_args = ["wallet", "did", "get_details", FINGERPRINT_ARG, "--coin_id", did_coin_id_hex]
|
||||
# these are various things that should be in the output
|
||||
assert_list = [
|
||||
f"DID: {encode_puzzle_hash(get_bytes32(2), 'did:chia:')}",
|
||||
f"Coin ID: {get_bytes32(3).hex()}",
|
||||
"Inner P2 Address: xch1qszqgpqyqszqgpqyqszqgpqyqszqgpqyqszqgpqyqszqgpqyqszqkxck8d",
|
||||
f"Public Key: {bytes48([5] * 48).hex()}",
|
||||
f"Launcher ID: {get_bytes32(6).hex()}",
|
||||
"DID Metadata: {'did metadata': 'yes'}",
|
||||
f"Recovery List Hash: {get_bytes32(7).hex()}",
|
||||
"Recovery Required Verifications: 8",
|
||||
"Last Spend Puzzle: 09",
|
||||
"Last Spend Solution: 0a",
|
||||
(
|
||||
"Last Spend Hints: ['0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b', "
|
||||
"'0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c']"
|
||||
),
|
||||
]
|
||||
run_cli_command_and_assert(capsys, root_dir, command_args, assert_list)
|
||||
expected_calls: logType = {
|
||||
"get_did_info": [(did_coin_id_hex, True)],
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
|
||||
|
||||
def test_did_update_metadata(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
# set RPC Client
|
||||
class DidUpdateMetadataRpcClient(TestWalletRpcClient):
|
||||
async def update_did_metadata(
|
||||
self,
|
||||
request: DIDUpdateMetadata,
|
||||
tx_config: TXConfig,
|
||||
extra_conditions: tuple[Condition, ...] = tuple(),
|
||||
timelock_info: ConditionValidTimes = ConditionValidTimes(),
|
||||
) -> DIDUpdateMetadataResponse:
|
||||
self.add_to_log(
|
||||
"update_did_metadata",
|
||||
(request.wallet_id, request.metadata, tx_config, request.push, extra_conditions, timelock_info),
|
||||
)
|
||||
return DIDUpdateMetadataResponse(
|
||||
unsigned_transactions=[STD_UTX],
|
||||
transactions=[STD_TX],
|
||||
spend_bundle=WalletSpendBundle([], G2Element()),
|
||||
wallet_id=uint32(request.wallet_id),
|
||||
)
|
||||
|
||||
inst_rpc_client = DidUpdateMetadataRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
w_id = 3
|
||||
json_mdata = '{"foo": "bar"}'
|
||||
command_args = [
|
||||
"wallet",
|
||||
"did",
|
||||
"update_metadata",
|
||||
FINGERPRINT_ARG,
|
||||
f"-i{w_id}",
|
||||
"--metadata",
|
||||
json_mdata,
|
||||
"--reuse",
|
||||
"--valid-at",
|
||||
"100",
|
||||
"--expires-at",
|
||||
"150",
|
||||
]
|
||||
# these are various things that should be in the output
|
||||
assert STD_TX.spend_bundle is not None
|
||||
assert_list = [f"Successfully updated DID wallet ID: {w_id}, Spend Bundle: {STD_TX.spend_bundle.to_json_dict()}"]
|
||||
run_cli_command_and_assert(capsys, root_dir, command_args, assert_list)
|
||||
expected_calls: logType = {
|
||||
"update_did_metadata": [
|
||||
(w_id, {"foo": "bar"}, DEFAULT_TX_CONFIG.override(reuse_puzhash=True), True, (), test_condition_valid_times)
|
||||
],
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
|
||||
|
||||
def test_did_find_lost(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
# set RPC Client
|
||||
class DidFindLostRpcClient(TestWalletRpcClient):
|
||||
async def find_lost_did(self, request: DIDFindLostDID) -> DIDFindLostDIDResponse:
|
||||
self.add_to_log(
|
||||
"find_lost_did",
|
||||
(request.coin_id, request.recovery_list_hash, request.metadata, request.num_verification),
|
||||
)
|
||||
return DIDFindLostDIDResponse(latest_coin_id=get_bytes32(2))
|
||||
|
||||
inst_rpc_client = DidFindLostRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
c_id = get_bytes32(1)
|
||||
json_mdata = '{"foo": "bar"}'
|
||||
command_args = [
|
||||
"wallet",
|
||||
"did",
|
||||
"find_lost",
|
||||
FINGERPRINT_ARG,
|
||||
"--coin_id",
|
||||
c_id.hex(),
|
||||
"--metadata",
|
||||
json_mdata,
|
||||
]
|
||||
# these are various things that should be in the output
|
||||
assert_list = [f"Successfully found lost DID {c_id.hex()}, latest coin ID: {get_bytes32(2).hex()}"]
|
||||
run_cli_command_and_assert(capsys, root_dir, command_args, assert_list)
|
||||
expected_calls: logType = {
|
||||
"find_lost_did": [(c_id.hex(), None, {"foo": "bar"}, None)],
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
|
||||
|
||||
def test_did_message_spend(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
# set RPC Client
|
||||
class DidMessageSpendRpcClient(TestWalletRpcClient):
|
||||
async def did_message_spend(
|
||||
self,
|
||||
request: DIDMessageSpend,
|
||||
tx_config: TXConfig,
|
||||
extra_conditions: tuple[Condition, ...] = tuple(),
|
||||
timelock_info: ConditionValidTimes = ConditionValidTimes(),
|
||||
) -> DIDMessageSpendResponse:
|
||||
self.add_to_log(
|
||||
"did_message_spend", (request.wallet_id, tx_config, extra_conditions, request.push, timelock_info)
|
||||
)
|
||||
return DIDMessageSpendResponse(
|
||||
unsigned_transactions=[STD_UTX], transactions=[STD_TX], spend_bundle=WalletSpendBundle([], G2Element())
|
||||
)
|
||||
|
||||
inst_rpc_client = DidMessageSpendRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
w_id = 3
|
||||
c_announcements = [get_bytes32(1), get_bytes32(2)]
|
||||
puz_announcements = [get_bytes32(3), get_bytes32(4)]
|
||||
command_args = [
|
||||
"wallet",
|
||||
"did",
|
||||
"message_spend",
|
||||
FINGERPRINT_ARG,
|
||||
f"-i{w_id}",
|
||||
"--coin_announcements",
|
||||
",".join([announcement.hex() for announcement in c_announcements]),
|
||||
"--puzzle_announcements",
|
||||
",".join([announcement.hex() for announcement in puz_announcements]),
|
||||
"--valid-at",
|
||||
"100",
|
||||
"--expires-at",
|
||||
"150",
|
||||
]
|
||||
# these are various things that should be in the output
|
||||
assert STD_TX.spend_bundle is not None
|
||||
assert_list = [f"Message Spend Bundle: {STD_TX.spend_bundle.to_json_dict()}"]
|
||||
run_cli_command_and_assert(capsys, root_dir, command_args, assert_list)
|
||||
expected_calls: logType = {
|
||||
"did_message_spend": [
|
||||
(
|
||||
w_id,
|
||||
DEFAULT_TX_CONFIG,
|
||||
(
|
||||
*(CreateCoinAnnouncement(ann) for ann in c_announcements),
|
||||
*(CreatePuzzleAnnouncement(ann) for ann in puz_announcements),
|
||||
),
|
||||
True,
|
||||
test_condition_valid_times,
|
||||
)
|
||||
],
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
|
||||
|
||||
def test_did_transfer(capsys: object, get_test_cli_clients: tuple[TestRpcClients, Path]) -> None:
|
||||
test_rpc_clients, root_dir = get_test_cli_clients
|
||||
|
||||
# set RPC Client
|
||||
class DidTransferRpcClient(TestWalletRpcClient):
|
||||
async def did_transfer_did(
|
||||
self,
|
||||
request: DIDTransferDID,
|
||||
tx_config: TXConfig,
|
||||
extra_conditions: tuple[Condition, ...] = tuple(),
|
||||
timelock_info: ConditionValidTimes = ConditionValidTimes(),
|
||||
) -> DIDTransferDIDResponse:
|
||||
self.add_to_log(
|
||||
"did_transfer_did",
|
||||
(
|
||||
request.wallet_id,
|
||||
request.inner_address,
|
||||
request.fee,
|
||||
request.with_recovery_info,
|
||||
tx_config,
|
||||
request.push,
|
||||
extra_conditions,
|
||||
timelock_info,
|
||||
),
|
||||
)
|
||||
return DIDTransferDIDResponse(
|
||||
unsigned_transactions=[STD_UTX],
|
||||
transactions=[STD_TX],
|
||||
transaction=STD_TX,
|
||||
transaction_id=STD_TX.name,
|
||||
)
|
||||
|
||||
inst_rpc_client = DidTransferRpcClient()
|
||||
test_rpc_clients.wallet_rpc_client = inst_rpc_client
|
||||
w_id = 3
|
||||
t_address = encode_puzzle_hash(get_bytes32(1), "xch")
|
||||
command_args = [
|
||||
"wallet",
|
||||
"did",
|
||||
"transfer",
|
||||
FINGERPRINT_ARG,
|
||||
f"-i{w_id}",
|
||||
"-m0.5",
|
||||
"--reuse",
|
||||
"--target-address",
|
||||
t_address,
|
||||
"--valid-at",
|
||||
"100",
|
||||
"--expires-at",
|
||||
"150",
|
||||
]
|
||||
# these are various things that should be in the output
|
||||
assert_list = [
|
||||
f"Successfully transferred DID to {t_address}",
|
||||
f"Transaction ID: {get_bytes32(2).hex()}",
|
||||
f"Transaction: {STD_TX.to_json_dict()}",
|
||||
]
|
||||
run_cli_command_and_assert(capsys, root_dir, command_args, assert_list)
|
||||
expected_calls: logType = {
|
||||
"did_transfer_did": [
|
||||
(
|
||||
w_id,
|
||||
t_address,
|
||||
500000000000,
|
||||
True,
|
||||
DEFAULT_TX_CONFIG.override(reuse_puzhash=True),
|
||||
True,
|
||||
(),
|
||||
test_condition_valid_times,
|
||||
)
|
||||
],
|
||||
}
|
||||
test_rpc_clients.wallet_rpc_client.check_log(expected_calls)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -46,8 +46,10 @@ from chia._tests.wallet.test_wallet_coin_store import (
|
||||
record_8,
|
||||
record_9,
|
||||
)
|
||||
from chia.cmds.cmd_helpers import NeedsWalletRPC, WalletClientInfo
|
||||
from chia.cmds.coins import CombineCMD, SplitCMD
|
||||
from chia.cmds.param_types import CliAmount
|
||||
from chia.cmds.wallet import DidGetDidCMD, DidSetWalletNameCMD
|
||||
from chia.full_node.full_node_rpc_client import FullNodeRpcClient
|
||||
from chia.pools.pool_wallet_info import NewPoolWalletInitialTargetState
|
||||
from chia.protocols.fee_estimate import FeeEstimate, FeeEstimateGroup
|
||||
@@ -126,7 +128,6 @@ from chia.wallet.wallet_request_types import (
|
||||
DeleteNotifications,
|
||||
DeleteUnconfirmedTransactions,
|
||||
DIDCreateBackupFile,
|
||||
DIDGetDID,
|
||||
DIDGetMetadata,
|
||||
DIDGetPubkey,
|
||||
DIDGetWalletName,
|
||||
@@ -2156,7 +2157,7 @@ async def test_get_coin_records_by_names(wallet_environments: WalletTestFramewor
|
||||
)
|
||||
@pytest.mark.limit_consensus_modes(reason="irrelevant")
|
||||
@pytest.mark.anyio
|
||||
async def test_did_endpoints(wallet_environments: WalletTestFramework) -> None:
|
||||
async def test_did_endpoints(wallet_environments: WalletTestFramework, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
env = wallet_environments.environments[0]
|
||||
env_2 = wallet_environments.environments[1]
|
||||
|
||||
@@ -2203,15 +2204,51 @@ async def test_did_endpoints(wallet_environments: WalletTestFramework) -> None:
|
||||
|
||||
# Set wallet name
|
||||
new_wallet_name = "test name"
|
||||
await wallet_1_rpc.did_set_wallet_name(DIDSetWalletName(wallet_id=did_wallet_id_0, name=new_wallet_name))
|
||||
await DidSetWalletNameCMD(
|
||||
rpc_info=NeedsWalletRPC(
|
||||
client_info=WalletClientInfo(
|
||||
wallet_1_rpc, env.wallet_state_manager.root_pubkey.get_fingerprint(), env.wallet_state_manager.config
|
||||
)
|
||||
),
|
||||
wallet_id=42,
|
||||
name=new_wallet_name,
|
||||
).run()
|
||||
assert "Failed to set DID wallet name" in capsys.readouterr().out
|
||||
await DidSetWalletNameCMD(
|
||||
rpc_info=NeedsWalletRPC(
|
||||
client_info=WalletClientInfo(
|
||||
wallet_1_rpc, env.wallet_state_manager.root_pubkey.get_fingerprint(), env.wallet_state_manager.config
|
||||
)
|
||||
),
|
||||
wallet_id=did_wallet_id_0,
|
||||
name=new_wallet_name,
|
||||
).run()
|
||||
|
||||
get_name_res = await wallet_1_rpc.did_get_wallet_name(DIDGetWalletName(wallet_id=did_wallet_id_0))
|
||||
assert get_name_res.name == new_wallet_name
|
||||
with pytest.raises(ValueError, match="wallet id 1 is of type Wallet but type DIDWallet is required"):
|
||||
await wallet_1_rpc.did_set_wallet_name(DIDSetWalletName(wallet_id=wallet_1_id, name=new_wallet_name))
|
||||
|
||||
# Check DID ID
|
||||
did_id_res = await wallet_1_rpc.get_did_id(DIDGetDID(wallet_id=did_wallet_id_0))
|
||||
assert did_id_0 == did_id_res.my_did
|
||||
await DidGetDidCMD(
|
||||
rpc_info=NeedsWalletRPC(
|
||||
client_info=WalletClientInfo(
|
||||
wallet_1_rpc, env.wallet_state_manager.root_pubkey.get_fingerprint(), env.wallet_state_manager.config
|
||||
)
|
||||
),
|
||||
wallet_id=42,
|
||||
).run()
|
||||
assert "Failed to get DID" in capsys.readouterr().out
|
||||
await DidGetDidCMD(
|
||||
rpc_info=NeedsWalletRPC(
|
||||
client_info=WalletClientInfo(
|
||||
wallet_1_rpc, env.wallet_state_manager.root_pubkey.get_fingerprint(), env.wallet_state_manager.config
|
||||
)
|
||||
),
|
||||
wallet_id=did_wallet_id_0,
|
||||
).run()
|
||||
assert did_id_0 is not None
|
||||
assert did_id_0 in capsys.readouterr().out
|
||||
# Create backup file
|
||||
await wallet_1_rpc.create_did_backup_file(DIDCreateBackupFile(wallet_id=did_wallet_id_0))
|
||||
|
||||
|
||||
+269
-336
@@ -759,362 +759,295 @@ def check_wallet_cmd(ctx: click.Context, db_path: str, verbose: bool) -> None:
|
||||
|
||||
|
||||
@wallet_cmd.group("did", help="DID related actions")
|
||||
def did_cmd() -> None:
|
||||
def did_cmd() -> None: # pragma: no cover
|
||||
pass
|
||||
|
||||
|
||||
@did_cmd.command("create", help="Create DID wallet")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="create",
|
||||
short_help="Create DID wallet",
|
||||
help="Create DID wallet",
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-n", "--name", help="Set the DID wallet name", type=str)
|
||||
@click.option(
|
||||
"-a",
|
||||
"--amount",
|
||||
help="Set the DID amount in mojos. Value must be an odd number.",
|
||||
type=int,
|
||||
default=1,
|
||||
show_default=True,
|
||||
)
|
||||
@options.create_fee()
|
||||
@tx_out_cmd()
|
||||
@click.pass_context
|
||||
def did_create_wallet_cmd(
|
||||
ctx: click.Context,
|
||||
wallet_rpc_port: int | None,
|
||||
fingerprint: int,
|
||||
name: str | None,
|
||||
amount: int,
|
||||
fee: uint64,
|
||||
push: bool,
|
||||
condition_valid_times: ConditionValidTimes,
|
||||
) -> list[TransactionRecord]:
|
||||
from chia.cmds.wallet_funcs import create_did_wallet
|
||||
|
||||
return asyncio.run(
|
||||
create_did_wallet(
|
||||
ChiaCliContext.set_default(ctx).root_path,
|
||||
wallet_rpc_port,
|
||||
fingerprint,
|
||||
fee,
|
||||
name,
|
||||
amount,
|
||||
push,
|
||||
condition_valid_times=condition_valid_times,
|
||||
)
|
||||
class CreateDidWalletCMD(TransactionEndpointWithTimelocks):
|
||||
name: str | None = option("-n", "--name", help="Set the DID wallet name", type=str, default=None)
|
||||
amount: int = option(
|
||||
"-a",
|
||||
"--amount",
|
||||
help="Set the DID amount in mojos. Value must be an odd number.",
|
||||
type=int,
|
||||
default=1,
|
||||
show_default=True,
|
||||
)
|
||||
metadata: Sequence[str] = option(
|
||||
"--metadata",
|
||||
help="A key value pair of metadata to set on the created DID (format == key:value)",
|
||||
type=str,
|
||||
multiple=True,
|
||||
default=tuple(),
|
||||
)
|
||||
|
||||
@transaction_endpoint_runner
|
||||
async def run(self) -> list[TransactionRecord]:
|
||||
from chia.cmds.wallet_funcs import create_did_wallet
|
||||
|
||||
@did_cmd.command("sign_message", help="Sign a message by a DID")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
return await create_did_wallet(
|
||||
wallet_info,
|
||||
self.fee,
|
||||
self.name,
|
||||
self.amount,
|
||||
self.push,
|
||||
self.metadata,
|
||||
condition_valid_times=self.load_condition_valid_times(),
|
||||
tx_config=self.tx_config_loader.load_tx_config(
|
||||
units["chia"], wallet_info.config, wallet_info.fingerprint
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="sign_message",
|
||||
short_help="Sign a message by a DID",
|
||||
help="Sign a message by a DID",
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-i", "--did_id", help="DID ID you want to use for signing", type=AddressParamType(), required=True)
|
||||
@click.option("-m", "--hex_message", help="The hex message you want to sign", type=str, required=True)
|
||||
@click.pass_context
|
||||
def did_sign_message(
|
||||
ctx: click.Context, wallet_rpc_port: int | None, fingerprint: int, did_id: CliAddress, hex_message: str
|
||||
) -> None:
|
||||
from chia.cmds.wallet_funcs import sign_message
|
||||
class DidSignMessageCMD:
|
||||
rpc_info: NeedsWalletRPC
|
||||
did_id: CliAddress = option(
|
||||
"-i", "--did_id", help="DID ID you want to use for signing", type=AddressParamType(), required=True
|
||||
)
|
||||
hex_message: str = option("-m", "--hex_message", help="The hex message you want to sign", type=str, required=True)
|
||||
|
||||
asyncio.run(
|
||||
sign_message(
|
||||
root_path=ChiaCliContext.set_default(ctx).root_path,
|
||||
wallet_rpc_port=wallet_rpc_port,
|
||||
fp=fingerprint,
|
||||
addr_type=AddressType.DID,
|
||||
message=hex_message,
|
||||
did_id=did_id,
|
||||
)
|
||||
async def run(self) -> None:
|
||||
from chia.cmds.wallet_funcs import sign_message
|
||||
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
await sign_message(
|
||||
wallet_info=wallet_info,
|
||||
addr_type=AddressType.DID,
|
||||
message=self.hex_message,
|
||||
did_id=self.did_id,
|
||||
)
|
||||
|
||||
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="set_name",
|
||||
short_help="Set DID wallet name",
|
||||
help="Set DID wallet name",
|
||||
)
|
||||
class DidSetWalletNameCMD:
|
||||
rpc_info: NeedsWalletRPC
|
||||
wallet_id: int = option("-i", "--id", help="Id of the wallet to use", type=int, required=True)
|
||||
name: str = option("-n", "--name", help="Set the DID wallet name", type=str, required=True)
|
||||
|
||||
async def run(self) -> None:
|
||||
from chia.cmds.wallet_funcs import did_set_wallet_name
|
||||
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
await did_set_wallet_name(wallet_info, self.wallet_id, self.name)
|
||||
|
||||
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="get_did",
|
||||
short_help="Get DID from wallet",
|
||||
help="Get DID from wallet",
|
||||
)
|
||||
class DidGetDidCMD:
|
||||
rpc_info: NeedsWalletRPC
|
||||
wallet_id: int = option("-i", "--id", help="Id of the wallet to use", type=int, required=True)
|
||||
|
||||
async def run(self) -> None:
|
||||
from chia.cmds.wallet_funcs import get_did
|
||||
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
await get_did(wallet_info, self.wallet_id)
|
||||
|
||||
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="get_details",
|
||||
short_help="Get more details of any DID",
|
||||
help="Get more details of any DID",
|
||||
)
|
||||
class DidGetDetailsCMD:
|
||||
rpc_info: NeedsWalletRPC
|
||||
coin_id: str = option("-id", "--coin_id", help="Id of the DID or any coin ID of the DID", type=str, required=True)
|
||||
latest: bool = option("-l", "--latest", help="Return latest DID information", is_flag=True, default=True)
|
||||
|
||||
async def run(self) -> None:
|
||||
from chia.cmds.wallet_funcs import get_did_info
|
||||
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
await get_did_info(wallet_info, self.coin_id, self.latest)
|
||||
|
||||
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="update_metadata",
|
||||
short_help="Update the metadata of a DID",
|
||||
help="Update the metadata of a DID",
|
||||
)
|
||||
class DidUpdateMetadataCMD(TransactionEndpointWithTimelocks):
|
||||
wallet_id: int = option("-i", "--id", help="Id of the DID wallet to use", type=int, required=True)
|
||||
metadata: str = option("-d", "--metadata", help="The new whole metadata in json format", type=str, required=True)
|
||||
|
||||
@transaction_endpoint_runner
|
||||
async def run(self) -> list[TransactionRecord]:
|
||||
from chia.cmds.wallet_funcs import update_did_metadata
|
||||
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
return await update_did_metadata(
|
||||
wallet_info,
|
||||
self.wallet_id,
|
||||
self.metadata,
|
||||
self.fee,
|
||||
self.push,
|
||||
condition_valid_times=self.load_condition_valid_times(),
|
||||
tx_config=self.tx_config_loader.load_tx_config(
|
||||
units["chia"], wallet_info.config, wallet_info.fingerprint
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="find_lost",
|
||||
short_help="Find the did you should own and recovery the DID wallet",
|
||||
help="Find the did you should own and recovery the DID wallet",
|
||||
)
|
||||
class DidFindLostCMD:
|
||||
rpc_info: NeedsWalletRPC
|
||||
coin_id: str = option("-id", "--coin_id", help="Id of the DID or any coin ID of the DID", type=str, required=True)
|
||||
metadata: str | None = option(
|
||||
"-m", "--metadata", help="The new whole metadata in json format", type=str, required=False
|
||||
)
|
||||
recovery_list_hash: str | None = option(
|
||||
"-r",
|
||||
"--recovery_list_hash",
|
||||
help="Override the recovery list hash of the DID. Only set this "
|
||||
"if your last DID spend updated the recovery list",
|
||||
type=str,
|
||||
required=False,
|
||||
)
|
||||
num_verification: int | None = option(
|
||||
"-n",
|
||||
"--num_verification",
|
||||
help="Override the required verification number of the DID."
|
||||
" Only set this if your last DID spend updated the required verification number",
|
||||
type=int,
|
||||
required=False,
|
||||
)
|
||||
|
||||
async def run(self) -> None:
|
||||
from chia.cmds.wallet_funcs import find_lost_did
|
||||
|
||||
@did_cmd.command("set_name", help="Set DID wallet name")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
await find_lost_did(
|
||||
wallet_info,
|
||||
self.coin_id,
|
||||
self.metadata,
|
||||
self.recovery_list_hash,
|
||||
self.num_verification,
|
||||
)
|
||||
|
||||
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="message_spend",
|
||||
short_help="Generate a DID spend bundle for announcements",
|
||||
help="Generate a DID spend bundle for announcements",
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-i", "--id", help="Id of the wallet to use", type=int, required=True)
|
||||
@click.option("-n", "--name", help="Set the DID wallet name", type=str, required=True)
|
||||
@click.pass_context
|
||||
def did_wallet_name_cmd(ctx: click.Context, wallet_rpc_port: int | None, fingerprint: int, id: int, name: str) -> None:
|
||||
from chia.cmds.wallet_funcs import did_set_wallet_name
|
||||
|
||||
asyncio.run(did_set_wallet_name(ChiaCliContext.set_default(ctx).root_path, wallet_rpc_port, fingerprint, id, name))
|
||||
|
||||
|
||||
@did_cmd.command("get_did", help="Get DID from wallet")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-i", "--id", help="Id of the wallet to use", type=int, required=True)
|
||||
@click.pass_context
|
||||
def did_get_did_cmd(ctx: click.Context, wallet_rpc_port: int | None, fingerprint: int, id: int) -> None:
|
||||
from chia.cmds.wallet_funcs import get_did
|
||||
|
||||
asyncio.run(get_did(ChiaCliContext.set_default(ctx).root_path, wallet_rpc_port, fingerprint, id))
|
||||
|
||||
|
||||
@did_cmd.command("get_details", help="Get more details of any DID")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-id", "--coin_id", help="Id of the DID or any coin ID of the DID", type=str, required=True)
|
||||
@click.option("-l", "--latest", help="Return latest DID information", is_flag=True, default=True)
|
||||
@click.pass_context
|
||||
def did_get_details_cmd(
|
||||
ctx: click.Context, wallet_rpc_port: int | None, fingerprint: int, coin_id: str, latest: bool
|
||||
) -> None:
|
||||
from chia.cmds.wallet_funcs import get_did_info
|
||||
|
||||
asyncio.run(get_did_info(ChiaCliContext.set_default(ctx).root_path, wallet_rpc_port, fingerprint, coin_id, latest))
|
||||
|
||||
|
||||
@did_cmd.command("update_metadata", help="Update the metadata of a DID")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-i", "--id", help="Id of the DID wallet to use", type=int, required=True)
|
||||
@click.option("-d", "--metadata", help="The new whole metadata in json format", type=str, required=True)
|
||||
@click.option(
|
||||
"--reuse",
|
||||
help="Reuse existing address for the change.",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
)
|
||||
@tx_out_cmd()
|
||||
@click.pass_context
|
||||
def did_update_metadata_cmd(
|
||||
ctx: click.Context,
|
||||
wallet_rpc_port: int | None,
|
||||
fingerprint: int,
|
||||
id: int,
|
||||
metadata: str,
|
||||
reuse: bool,
|
||||
push: bool,
|
||||
condition_valid_times: ConditionValidTimes,
|
||||
) -> list[TransactionRecord]:
|
||||
from chia.cmds.wallet_funcs import update_did_metadata
|
||||
|
||||
return asyncio.run(
|
||||
update_did_metadata(
|
||||
ChiaCliContext.set_default(ctx).root_path,
|
||||
wallet_rpc_port,
|
||||
fingerprint,
|
||||
id,
|
||||
metadata,
|
||||
reuse,
|
||||
push=push,
|
||||
condition_valid_times=condition_valid_times,
|
||||
)
|
||||
class DidMessageSpendCMD(TransactionEndpointWithTimelocks):
|
||||
wallet_id: int = option("-i", "--id", help="Id of the DID wallet to use", type=int, required=True)
|
||||
puzzle_announcements: str | None = option(
|
||||
"-pa",
|
||||
"--puzzle_announcements",
|
||||
help="The list of puzzle announcement hex strings, split by comma (,)",
|
||||
type=str,
|
||||
required=False,
|
||||
)
|
||||
coin_announcements: str | None = option(
|
||||
"-ca",
|
||||
"--coin_announcements",
|
||||
help="The list of coin announcement hex strings, split by comma (,)",
|
||||
type=str,
|
||||
required=False,
|
||||
)
|
||||
|
||||
@transaction_endpoint_runner
|
||||
async def run(self) -> list[TransactionRecord]:
|
||||
from chia.cmds.wallet_funcs import did_message_spend
|
||||
|
||||
@did_cmd.command("find_lost", help="Find the did you should own and recovery the DID wallet")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-id", "--coin_id", help="Id of the DID or any coin ID of the DID", type=str, required=True)
|
||||
@click.option("-m", "--metadata", help="The new whole metadata in json format", type=str, required=False)
|
||||
@click.option(
|
||||
"-r",
|
||||
"--recovery_list_hash",
|
||||
help="Override the recovery list hash of the DID. Only set this if your last DID spend updated the recovery list",
|
||||
type=str,
|
||||
required=False,
|
||||
)
|
||||
@click.option(
|
||||
"-n",
|
||||
"--num_verification",
|
||||
help="Override the required verification number of the DID."
|
||||
" Only set this if your last DID spend updated the required verification number",
|
||||
type=int,
|
||||
required=False,
|
||||
)
|
||||
@click.pass_context
|
||||
def did_find_lost_cmd(
|
||||
ctx: click.Context,
|
||||
wallet_rpc_port: int | None,
|
||||
fingerprint: int,
|
||||
coin_id: str,
|
||||
metadata: str | None,
|
||||
recovery_list_hash: str | None,
|
||||
num_verification: int | None,
|
||||
) -> None:
|
||||
from chia.cmds.wallet_funcs import find_lost_did
|
||||
puzzle_list: list[str] = []
|
||||
coin_list: list[str] = []
|
||||
if self.puzzle_announcements is not None:
|
||||
try:
|
||||
puzzle_list = self.puzzle_announcements.split(",")
|
||||
for announcement in puzzle_list:
|
||||
bytes.fromhex(announcement)
|
||||
except ValueError:
|
||||
print("Invalid puzzle announcement format, should be a list of hex strings.")
|
||||
return []
|
||||
if self.coin_announcements is not None:
|
||||
try:
|
||||
coin_list = self.coin_announcements.split(",")
|
||||
for announcement in coin_list:
|
||||
bytes.fromhex(announcement)
|
||||
except ValueError:
|
||||
print("Invalid coin announcement format, should be a list of hex strings.")
|
||||
return []
|
||||
|
||||
asyncio.run(
|
||||
find_lost_did(
|
||||
root_path=ChiaCliContext.set_default(ctx).root_path,
|
||||
wallet_rpc_port=wallet_rpc_port,
|
||||
fp=fingerprint,
|
||||
coin_id=coin_id,
|
||||
metadata=metadata,
|
||||
recovery_list_hash=recovery_list_hash,
|
||||
num_verification=num_verification,
|
||||
)
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
return await did_message_spend(
|
||||
wallet_info,
|
||||
self.wallet_id,
|
||||
puzzle_list,
|
||||
coin_list,
|
||||
self.fee,
|
||||
self.push,
|
||||
condition_valid_times=self.load_condition_valid_times(),
|
||||
tx_config=self.tx_config_loader.load_tx_config(
|
||||
units["chia"], wallet_info.config, wallet_info.fingerprint
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@chia_command(
|
||||
group=did_cmd,
|
||||
name="transfer",
|
||||
short_help="Transfer a DID",
|
||||
help="Transfer a DID",
|
||||
)
|
||||
class DidTransferDidCMD(TransactionEndpointWithTimelocks):
|
||||
wallet_id: int = option("-i", "--id", help="Id of the DID wallet to use", type=int, required=True)
|
||||
# TODO: Change RPC to use puzzlehash instead of address
|
||||
target_address: CliAddress = option(
|
||||
"-ta", "--target-address", help="Target recipient wallet address", type=AddressParamType(), required=True
|
||||
)
|
||||
reset_recovery: bool = option(
|
||||
"-rr", "--reset_recovery", help="If you want to reset the recovery DID settings.", is_flag=True, default=False
|
||||
)
|
||||
|
||||
@transaction_endpoint_runner
|
||||
async def run(self) -> list[TransactionRecord]:
|
||||
from chia.cmds.wallet_funcs import transfer_did
|
||||
|
||||
@did_cmd.command("message_spend", help="Generate a DID spend bundle for announcements")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-i", "--id", help="Id of the DID wallet to use", type=int, required=True)
|
||||
@click.option(
|
||||
"-pa",
|
||||
"--puzzle_announcements",
|
||||
help="The list of puzzle announcement hex strings, split by comma (,)",
|
||||
type=str,
|
||||
required=False,
|
||||
)
|
||||
@click.option(
|
||||
"-ca",
|
||||
"--coin_announcements",
|
||||
help="The list of coin announcement hex strings, split by comma (,)",
|
||||
type=str,
|
||||
required=False,
|
||||
)
|
||||
@tx_out_cmd()
|
||||
@click.pass_context
|
||||
def did_message_spend_cmd(
|
||||
ctx: click.Context,
|
||||
wallet_rpc_port: int | None,
|
||||
fingerprint: int,
|
||||
id: int,
|
||||
puzzle_announcements: str | None,
|
||||
coin_announcements: str | None,
|
||||
push: bool,
|
||||
condition_valid_times: ConditionValidTimes,
|
||||
) -> list[TransactionRecord]:
|
||||
from chia.cmds.wallet_funcs import did_message_spend
|
||||
|
||||
puzzle_list: list[str] = []
|
||||
coin_list: list[str] = []
|
||||
if puzzle_announcements is not None:
|
||||
try:
|
||||
puzzle_list = puzzle_announcements.split(",")
|
||||
# validate puzzle announcements is list of hex strings
|
||||
for announcement in puzzle_list:
|
||||
bytes.fromhex(announcement)
|
||||
except ValueError:
|
||||
print("Invalid puzzle announcement format, should be a list of hex strings.")
|
||||
return []
|
||||
if coin_announcements is not None:
|
||||
try:
|
||||
coin_list = coin_announcements.split(",")
|
||||
# validate that coin announcements is a list of hex strings
|
||||
for announcement in coin_list:
|
||||
bytes.fromhex(announcement)
|
||||
except ValueError:
|
||||
print("Invalid coin announcement format, should be a list of hex strings.")
|
||||
return []
|
||||
|
||||
return asyncio.run(
|
||||
did_message_spend(
|
||||
ChiaCliContext.set_default(ctx).root_path,
|
||||
wallet_rpc_port,
|
||||
fingerprint,
|
||||
id,
|
||||
puzzle_list,
|
||||
coin_list,
|
||||
push=push,
|
||||
condition_valid_times=condition_valid_times,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@did_cmd.command("transfer", help="Transfer a DID")
|
||||
@click.option(
|
||||
"-wp",
|
||||
"--wallet-rpc-port",
|
||||
help="Set the port where the Wallet is hosting the RPC interface. See the rpc_port under wallet in config.yaml",
|
||||
type=int,
|
||||
default=None,
|
||||
)
|
||||
@options.create_fingerprint()
|
||||
@click.option("-i", "--id", help="Id of the DID wallet to use", type=int, required=True)
|
||||
# TODO: Change RPC to use puzzlehash instead of address
|
||||
@click.option("-ta", "--target-address", help="Target recipient wallet address", type=AddressParamType(), required=True)
|
||||
@click.option(
|
||||
"-rr", "--reset_recovery", help="If you want to reset the recovery DID settings.", is_flag=True, default=False
|
||||
)
|
||||
@options.create_fee()
|
||||
@click.option(
|
||||
"--reuse",
|
||||
help="Reuse existing address for the change.",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
)
|
||||
@tx_out_cmd()
|
||||
@click.pass_context
|
||||
def did_transfer_did(
|
||||
ctx: click.Context,
|
||||
wallet_rpc_port: int | None,
|
||||
fingerprint: int,
|
||||
id: int,
|
||||
target_address: CliAddress,
|
||||
reset_recovery: bool,
|
||||
fee: uint64,
|
||||
reuse: bool,
|
||||
push: bool,
|
||||
condition_valid_times: ConditionValidTimes,
|
||||
) -> list[TransactionRecord]:
|
||||
from chia.cmds.wallet_funcs import transfer_did
|
||||
|
||||
return asyncio.run(
|
||||
transfer_did(
|
||||
ChiaCliContext.set_default(ctx).root_path,
|
||||
wallet_rpc_port,
|
||||
fingerprint,
|
||||
id,
|
||||
fee,
|
||||
target_address,
|
||||
reset_recovery is False,
|
||||
True if reuse else None,
|
||||
push=push,
|
||||
condition_valid_times=condition_valid_times,
|
||||
)
|
||||
)
|
||||
async with self.rpc_info.wallet_rpc() as wallet_info:
|
||||
return await transfer_did(
|
||||
wallet_info,
|
||||
self.wallet_id,
|
||||
self.fee,
|
||||
self.target_address,
|
||||
not self.reset_recovery,
|
||||
self.push,
|
||||
condition_valid_times=self.load_condition_valid_times(),
|
||||
tx_config=self.tx_config_loader.load_tx_config(
|
||||
units["chia"], wallet_info.config, wallet_info.fingerprint
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@wallet_cmd.group("nft", help="NFT related actions")
|
||||
|
||||
+182
-191
@@ -1074,211 +1074,190 @@ async def print_balances(
|
||||
|
||||
|
||||
async def create_did_wallet(
|
||||
root_path: pathlib.Path,
|
||||
wallet_rpc_port: int | None,
|
||||
fp: int | None,
|
||||
wallet_info: WalletClientInfo,
|
||||
fee: uint64,
|
||||
name: str | None,
|
||||
amount: int,
|
||||
push: bool,
|
||||
metadata: Sequence[str],
|
||||
condition_valid_times: ConditionValidTimes,
|
||||
tx_config: TXConfig,
|
||||
) -> list[TransactionRecord]:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, fingerprint, config):
|
||||
try:
|
||||
response = await wallet_client.create_new_wallet(
|
||||
CreateNewWallet(
|
||||
wallet_type=CreateNewWalletType.DID_WALLET,
|
||||
did_type=DIDType.NEW,
|
||||
amount=uint64(amount),
|
||||
fee=fee,
|
||||
wallet_name=name,
|
||||
push=push,
|
||||
),
|
||||
tx_config=CMDTXConfigLoader().to_tx_config(units["chia"], config, fingerprint),
|
||||
timelock_info=condition_valid_times,
|
||||
)
|
||||
wallet_id = response.wallet_id
|
||||
my_did = response.my_did
|
||||
print(f"Successfully created a DID wallet with name {name} and id {wallet_id} on key {fingerprint}")
|
||||
print(f"Successfully created a DID {my_did} in the newly created DID wallet")
|
||||
return [] # TODO: fix this endpoint to return transactions
|
||||
except Exception as e:
|
||||
print(f"Failed to create DID wallet: {e}")
|
||||
return []
|
||||
try:
|
||||
response = await wallet_info.client.create_new_wallet(
|
||||
CreateNewWallet(
|
||||
wallet_type=CreateNewWalletType.DID_WALLET,
|
||||
did_type=DIDType.NEW,
|
||||
amount=uint64(amount),
|
||||
fee=fee,
|
||||
wallet_name=name,
|
||||
push=push,
|
||||
metadata={
|
||||
split_args[0]: ":".join(split_args[1:]) for split_args in (pair.split(":") for pair in metadata)
|
||||
},
|
||||
),
|
||||
tx_config=tx_config,
|
||||
timelock_info=condition_valid_times,
|
||||
)
|
||||
wallet_id = response.wallet_id
|
||||
my_did = response.my_did
|
||||
print(f"Successfully created a DID wallet with name {name} and id {wallet_id} on key {wallet_info.fingerprint}")
|
||||
print(f"Successfully created a DID {my_did} in the newly created DID wallet")
|
||||
return [] # TODO: fix this endpoint to return transactions
|
||||
except Exception as e:
|
||||
print(f"Failed to create DID wallet: {e}")
|
||||
return []
|
||||
|
||||
|
||||
async def did_set_wallet_name(
|
||||
root_path: pathlib.Path, wallet_rpc_port: int | None, fp: int | None, wallet_id: int, name: str
|
||||
) -> None:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, _, _):
|
||||
try:
|
||||
await wallet_client.did_set_wallet_name(DIDSetWalletName(wallet_id=uint32(wallet_id), name=name))
|
||||
print(f"Successfully set a new name for DID wallet with id {wallet_id}: {name}")
|
||||
except Exception as e:
|
||||
print(f"Failed to set DID wallet name: {e}")
|
||||
async def did_set_wallet_name(wallet_info: WalletClientInfo, wallet_id: int, name: str) -> None:
|
||||
try:
|
||||
await wallet_info.client.did_set_wallet_name(DIDSetWalletName(wallet_id=uint32(wallet_id), name=name))
|
||||
print(f"Successfully set a new name for DID wallet with id {wallet_id}: {name}")
|
||||
except Exception as e:
|
||||
print(f"Failed to set DID wallet name: {e}")
|
||||
|
||||
|
||||
async def get_did(root_path: pathlib.Path, wallet_rpc_port: int | None, fp: int | None, did_wallet_id: int) -> None:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, _, _):
|
||||
try:
|
||||
response = await wallet_client.get_did_id(DIDGetDID(wallet_id=uint32(did_wallet_id)))
|
||||
print(f"{'DID:'.ljust(23)} {response.my_did}")
|
||||
print(f"{'Coin ID:'.ljust(23)} {response.coin_id.hex() if response.coin_id is not None else 'Unknown'}")
|
||||
except Exception as e:
|
||||
print(f"Failed to get DID: {e}")
|
||||
async def get_did(wallet_info: WalletClientInfo, did_wallet_id: int) -> None:
|
||||
try:
|
||||
response = await wallet_info.client.get_did_id(DIDGetDID(wallet_id=uint32(did_wallet_id)))
|
||||
print(f"{'DID:'.ljust(23)} {response.my_did}")
|
||||
print(f"{'Coin ID:'.ljust(23)} {response.coin_id.hex() if response.coin_id is not None else 'Unknown'}")
|
||||
except Exception as e:
|
||||
print(f"Failed to get DID: {e}")
|
||||
|
||||
|
||||
async def get_did_info(
|
||||
root_path: pathlib.Path, wallet_rpc_port: int | None, fp: int | None, coin_id: str, latest: bool
|
||||
) -> None:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, _, _):
|
||||
did_padding_length = 23
|
||||
try:
|
||||
response = await wallet_client.get_did_info(DIDGetInfo(coin_id=coin_id, latest=latest))
|
||||
print(f"{'DID:'.ljust(did_padding_length)} {response.did_id}")
|
||||
print(f"{'Coin ID:'.ljust(did_padding_length)} {response.latest_coin.hex()}")
|
||||
print(f"{'Inner P2 Address:'.ljust(did_padding_length)} {response.p2_address}")
|
||||
print(f"{'Public Key:'.ljust(did_padding_length)} {response.public_key.hex()}")
|
||||
print(f"{'Launcher ID:'.ljust(did_padding_length)} {response.launcher_id.hex()}")
|
||||
print(f"{'DID Metadata:'.ljust(did_padding_length)} {response.metadata}")
|
||||
print(
|
||||
f"{'Recovery List Hash:'.ljust(did_padding_length)} "
|
||||
+ (response.recovery_list_hash.hex() if response.recovery_list_hash is not None else "")
|
||||
)
|
||||
print(f"{'Recovery Required Verifications:'.ljust(did_padding_length)} {response.num_verification}")
|
||||
print(f"{'Last Spend Puzzle:'.ljust(did_padding_length)} {bytes(response.full_puzzle).hex()}")
|
||||
print(f"{'Last Spend Solution:'.ljust(did_padding_length)} {bytes(response.solution).hex()}")
|
||||
print(f"{'Last Spend Hints:'.ljust(did_padding_length)} {[hint.hex() for hint in response.hints]}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Failed to get DID details: {e}")
|
||||
async def get_did_info(wallet_info: WalletClientInfo, coin_id: str, latest: bool) -> None:
|
||||
did_padding_length = 23
|
||||
try:
|
||||
response = await wallet_info.client.get_did_info(DIDGetInfo(coin_id=coin_id, latest=latest))
|
||||
print(f"{'DID:'.ljust(did_padding_length)} {response.did_id}")
|
||||
print(f"{'Coin ID:'.ljust(did_padding_length)} {response.latest_coin.hex()}")
|
||||
print(f"{'Inner P2 Address:'.ljust(did_padding_length)} {response.p2_address}")
|
||||
print(f"{'Public Key:'.ljust(did_padding_length)} {response.public_key.hex()}")
|
||||
print(f"{'Launcher ID:'.ljust(did_padding_length)} {response.launcher_id.hex()}")
|
||||
print(f"{'DID Metadata:'.ljust(did_padding_length)} {response.metadata}")
|
||||
print(
|
||||
f"{'Recovery List Hash:'.ljust(did_padding_length)} "
|
||||
+ (response.recovery_list_hash.hex() if response.recovery_list_hash is not None else "")
|
||||
)
|
||||
print(f"{'Recovery Required Verifications:'.ljust(did_padding_length)} {response.num_verification}")
|
||||
print(f"{'Last Spend Puzzle:'.ljust(did_padding_length)} {bytes(response.full_puzzle).hex()}")
|
||||
print(f"{'Last Spend Solution:'.ljust(did_padding_length)} {bytes(response.solution).hex()}")
|
||||
print(f"{'Last Spend Hints:'.ljust(did_padding_length)} {[hint.hex() for hint in response.hints]}")
|
||||
except Exception as e:
|
||||
print(f"Failed to get DID details: {e}")
|
||||
|
||||
|
||||
async def update_did_metadata(
|
||||
root_path: pathlib.Path,
|
||||
wallet_rpc_port: int | None,
|
||||
fp: int | None,
|
||||
wallet_info: WalletClientInfo,
|
||||
did_wallet_id: int,
|
||||
metadata: str,
|
||||
reuse_puzhash: bool,
|
||||
fee: uint64,
|
||||
push: bool,
|
||||
condition_valid_times: ConditionValidTimes,
|
||||
tx_config: TXConfig,
|
||||
) -> list[TransactionRecord]:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, fingerprint, config):
|
||||
try:
|
||||
response = await wallet_client.update_did_metadata(
|
||||
DIDUpdateMetadata(
|
||||
wallet_id=uint32(did_wallet_id),
|
||||
metadata=json.loads(metadata),
|
||||
push=push,
|
||||
),
|
||||
tx_config=CMDTXConfigLoader(
|
||||
reuse_puzhash=reuse_puzhash,
|
||||
).to_tx_config(units["chia"], config, fingerprint),
|
||||
timelock_info=condition_valid_times,
|
||||
try:
|
||||
response = await wallet_info.client.update_did_metadata(
|
||||
DIDUpdateMetadata(
|
||||
wallet_id=uint32(did_wallet_id),
|
||||
metadata=json.loads(metadata),
|
||||
fee=fee,
|
||||
push=push,
|
||||
),
|
||||
tx_config=tx_config,
|
||||
timelock_info=condition_valid_times,
|
||||
)
|
||||
if push:
|
||||
print(
|
||||
f"Successfully updated DID wallet ID: {response.wallet_id}, "
|
||||
f"Spend Bundle: {response.spend_bundle.to_json_dict()}"
|
||||
)
|
||||
if push:
|
||||
print(
|
||||
f"Successfully updated DID wallet ID: {response.wallet_id}, "
|
||||
f"Spend Bundle: {response.spend_bundle.to_json_dict()}"
|
||||
)
|
||||
return response.transactions
|
||||
except Exception as e:
|
||||
print(f"Failed to update DID metadata: {e}")
|
||||
return []
|
||||
return response.transactions
|
||||
except Exception as e:
|
||||
print(f"Failed to update DID metadata: {e}")
|
||||
return []
|
||||
|
||||
|
||||
async def did_message_spend(
|
||||
root_path: pathlib.Path,
|
||||
wallet_rpc_port: int | None,
|
||||
fp: int | None,
|
||||
wallet_info: WalletClientInfo,
|
||||
did_wallet_id: int,
|
||||
puzzle_announcements: list[str],
|
||||
coin_announcements: list[str],
|
||||
fee: uint64,
|
||||
push: bool,
|
||||
condition_valid_times: ConditionValidTimes,
|
||||
tx_config: TXConfig,
|
||||
) -> list[TransactionRecord]:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, fingerprint, config):
|
||||
try:
|
||||
response = await wallet_client.did_message_spend(
|
||||
DIDMessageSpend(wallet_id=uint32(did_wallet_id), push=push),
|
||||
CMDTXConfigLoader().to_tx_config(units["chia"], config, fingerprint),
|
||||
extra_conditions=(
|
||||
*(CreateCoinAnnouncement(hexstr_to_bytes(ca)) for ca in coin_announcements),
|
||||
*(CreatePuzzleAnnouncement(hexstr_to_bytes(pa)) for pa in puzzle_announcements),
|
||||
),
|
||||
timelock_info=condition_valid_times,
|
||||
)
|
||||
print(f"Message Spend Bundle: {response.spend_bundle.to_json_dict()}")
|
||||
return response.transactions
|
||||
except Exception as e:
|
||||
print(f"Failed to update DID metadata: {e}")
|
||||
return []
|
||||
try:
|
||||
response = await wallet_info.client.did_message_spend(
|
||||
DIDMessageSpend(wallet_id=uint32(did_wallet_id), fee=fee, push=push),
|
||||
tx_config,
|
||||
extra_conditions=(
|
||||
*(CreateCoinAnnouncement(hexstr_to_bytes(ca)) for ca in coin_announcements),
|
||||
*(CreatePuzzleAnnouncement(hexstr_to_bytes(pa)) for pa in puzzle_announcements),
|
||||
),
|
||||
timelock_info=condition_valid_times,
|
||||
)
|
||||
print(f"Message Spend Bundle: {json.dumps(response.spend_bundle.to_json_dict())}")
|
||||
return response.transactions
|
||||
except Exception as e:
|
||||
print(f"Failed to create DID message spend: {e}")
|
||||
return []
|
||||
|
||||
|
||||
async def transfer_did(
|
||||
root_path: pathlib.Path,
|
||||
wallet_rpc_port: int | None,
|
||||
fp: int | None,
|
||||
wallet_info: WalletClientInfo,
|
||||
did_wallet_id: int,
|
||||
fee: uint64,
|
||||
target_cli_address: CliAddress,
|
||||
with_recovery: bool,
|
||||
reuse_puzhash: bool | None,
|
||||
push: bool,
|
||||
condition_valid_times: ConditionValidTimes,
|
||||
tx_config: TXConfig,
|
||||
) -> list[TransactionRecord]:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, fingerprint, config):
|
||||
try:
|
||||
target_address = target_cli_address.original_address
|
||||
response = await wallet_client.did_transfer_did(
|
||||
DIDTransferDID(
|
||||
wallet_id=uint32(did_wallet_id),
|
||||
inner_address=target_address,
|
||||
fee=fee,
|
||||
with_recovery_info=with_recovery,
|
||||
push=push,
|
||||
),
|
||||
tx_config=CMDTXConfigLoader(
|
||||
reuse_puzhash=reuse_puzhash,
|
||||
).to_tx_config(units["chia"], config, fingerprint),
|
||||
timelock_info=condition_valid_times,
|
||||
)
|
||||
if push:
|
||||
print(f"Successfully transferred DID to {target_address}")
|
||||
print(f"Transaction ID: {response.transaction_id.hex()}")
|
||||
print(f"Transaction: {response.transaction.to_json_dict()}")
|
||||
return response.transactions
|
||||
except Exception as e:
|
||||
print(f"Failed to transfer DID: {e}")
|
||||
return []
|
||||
try:
|
||||
target_address = target_cli_address.original_address
|
||||
response = await wallet_info.client.did_transfer_did(
|
||||
DIDTransferDID(
|
||||
wallet_id=uint32(did_wallet_id),
|
||||
inner_address=target_address,
|
||||
fee=fee,
|
||||
with_recovery_info=with_recovery,
|
||||
push=push,
|
||||
),
|
||||
tx_config=tx_config,
|
||||
timelock_info=condition_valid_times,
|
||||
)
|
||||
if push:
|
||||
print(f"Successfully transferred DID to {target_address}")
|
||||
print(f"Transaction ID: {response.transaction_id.hex()}")
|
||||
print(f"Transaction: {response.transaction.to_json_dict()}")
|
||||
return response.transactions
|
||||
except Exception as e:
|
||||
print(f"Failed to transfer DID: {e}")
|
||||
return []
|
||||
|
||||
|
||||
async def find_lost_did(
|
||||
*,
|
||||
root_path: pathlib.Path,
|
||||
wallet_rpc_port: int | None,
|
||||
fp: int | None,
|
||||
wallet_info: WalletClientInfo,
|
||||
coin_id: str,
|
||||
metadata: str | None,
|
||||
recovery_list_hash: str | None,
|
||||
num_verification: int | None,
|
||||
) -> None:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, _, _):
|
||||
try:
|
||||
response = await wallet_client.find_lost_did(
|
||||
DIDFindLostDID(
|
||||
coin_id=coin_id,
|
||||
recovery_list_hash=bytes32.from_hexstr(recovery_list_hash)
|
||||
if recovery_list_hash is not None
|
||||
else None,
|
||||
num_verification=uint16.construct_optional(num_verification),
|
||||
metadata=json.loads(metadata) if metadata is not None else None,
|
||||
)
|
||||
try:
|
||||
response = await wallet_info.client.find_lost_did(
|
||||
DIDFindLostDID(
|
||||
coin_id=coin_id,
|
||||
recovery_list_hash=bytes32.from_hexstr(recovery_list_hash) if recovery_list_hash is not None else None,
|
||||
num_verification=uint16.construct_optional(num_verification),
|
||||
metadata=json.loads(metadata) if metadata is not None else None,
|
||||
)
|
||||
print(f"Successfully found lost DID {coin_id}, latest coin ID: {response.latest_coin_id.hex()}")
|
||||
except Exception as e:
|
||||
print(f"Failed to find lost DID: {e}")
|
||||
)
|
||||
print(f"Successfully found lost DID {coin_id}, latest coin ID: {response.latest_coin_id.hex()}")
|
||||
except Exception as e:
|
||||
print(f"Failed to find lost DID: {e}")
|
||||
|
||||
|
||||
async def create_nft_wallet(
|
||||
@@ -1690,48 +1669,60 @@ async def delete_notifications(wallet_info: WalletClientInfo, ids: Sequence[byte
|
||||
print("Success!")
|
||||
|
||||
|
||||
async def sign_message(
|
||||
async def sign_message(*, wallet_info: WalletClientInfo | None = None, **kwargs: Any) -> None:
|
||||
if wallet_info is not None:
|
||||
await _sign_message(wallet_info=wallet_info, **kwargs)
|
||||
else:
|
||||
async with get_wallet_client(kwargs["root_path"], kwargs.get("wallet_rpc_port"), kwargs.get("fp")) as (
|
||||
wallet_client,
|
||||
fp,
|
||||
config,
|
||||
):
|
||||
del kwargs["root_path"]
|
||||
del kwargs["wallet_rpc_port"]
|
||||
del kwargs["fp"]
|
||||
await _sign_message(
|
||||
wallet_info=WalletClientInfo(client=wallet_client, fingerprint=fp, config=config), **kwargs
|
||||
)
|
||||
|
||||
|
||||
async def _sign_message(
|
||||
*,
|
||||
root_path: pathlib.Path,
|
||||
wallet_rpc_port: int | None,
|
||||
fp: int | None,
|
||||
wallet_info: WalletClientInfo,
|
||||
addr_type: AddressType,
|
||||
message: str,
|
||||
address: CliAddress | None = None,
|
||||
did_id: CliAddress | None = None,
|
||||
nft_id: CliAddress | None = None,
|
||||
) -> None:
|
||||
async with get_wallet_client(root_path, wallet_rpc_port, fp) as (wallet_client, _, _):
|
||||
response: SignMessageByAddressResponse | SignMessageByIDResponse
|
||||
if addr_type == AddressType.XCH:
|
||||
if address is None:
|
||||
print("Address is required for XCH address type.")
|
||||
return
|
||||
response = await wallet_client.sign_message_by_address(
|
||||
SignMessageByAddress(address=address.original_address, message=message)
|
||||
)
|
||||
elif addr_type == AddressType.DID:
|
||||
if did_id is None:
|
||||
print("DID id is required for DID address type.")
|
||||
return
|
||||
response = await wallet_client.sign_message_by_id(
|
||||
SignMessageByID(id=did_id.original_address, message=message)
|
||||
)
|
||||
elif addr_type == AddressType.NFT:
|
||||
if nft_id is None:
|
||||
print("NFT id is required for NFT address type.")
|
||||
return
|
||||
response = await wallet_client.sign_message_by_id(
|
||||
SignMessageByID(id=nft_id.original_address, message=message)
|
||||
)
|
||||
else:
|
||||
print("Invalid wallet type.")
|
||||
return
|
||||
print("")
|
||||
print(f"Message: {message}")
|
||||
print(f"Public Key: {response.pubkey!s}")
|
||||
print(f"Signature: {response.signature!s}")
|
||||
print(f"Signing Mode: {response.signing_mode}")
|
||||
response: SignMessageByAddressResponse | SignMessageByIDResponse
|
||||
# It's the responsiblity of our code to make sure the correct address/did/nft_id is provided for the address type
|
||||
if addr_type == AddressType.XCH:
|
||||
assert address is not None
|
||||
response = await wallet_info.client.sign_message_by_address(
|
||||
SignMessageByAddress(address=address.original_address, message=message)
|
||||
)
|
||||
elif addr_type == AddressType.DID:
|
||||
assert did_id is not None
|
||||
response = await wallet_info.client.sign_message_by_id(
|
||||
SignMessageByID(id=did_id.original_address, message=message)
|
||||
)
|
||||
elif addr_type == AddressType.NFT:
|
||||
assert nft_id is not None
|
||||
response = await wallet_info.client.sign_message_by_id(
|
||||
SignMessageByID(id=nft_id.original_address, message=message)
|
||||
)
|
||||
# This should be impossible because all address types are in the if/else chain
|
||||
# but just in case one is added in the future, we leave this in as a fallback
|
||||
else: # pragma: no cover
|
||||
print("Invalid wallet type.")
|
||||
return
|
||||
|
||||
print("")
|
||||
print(f"Message: {message}")
|
||||
print(f"Public Key: {response.pubkey!s}")
|
||||
print(f"Signature: {response.signature!s}")
|
||||
print(f"Signing Mode: {response.signing_mode}")
|
||||
|
||||
|
||||
async def spend_clawback(
|
||||
|
||||
Reference in New Issue
Block a user