Chia-Network/chia-blockchain
1
0
Fork 0
mirror of https://github.com/Chia-Network/chia-blockchain.git synced 2024-09-20 08:05:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Add DID CLI (#15065)

Browse Source 
* Add DID CLI

* minor fixes

* build fix

* resolving comments

---------

Co-authored-by: Sebastjan Trepca <trepca@gmail.com>
This commit is contained in:
Kronus91 2023-04-24 08:44:43 -10:00 committed by GitHub
parent e824332f35
commit 24019e35bb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 319 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

202
chia/cmds/wallet.py
View File

@ -678,6 +678,208 @@ def did_get_did_cmd(wallet_rpc_port: Optional[int], fingerprint: int, id: int) -
asyncio.run(execute_with_wallet(wallet_rpc_port, fingerprint, extra_params, get_did))
@did_cmd.command("get_details", short_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,
)
@click.option("-f", "--fingerprint", help="Set the fingerprint to specify which key to use", type=int)
@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)
def did_get_details_cmd(wallet_rpc_port: Optional[int], fingerprint: int, coin_id: str, latest: bool) -> None:
import asyncio
from .wallet_funcs import get_did_info
extra_params = {"coin_id": coin_id, "latest": latest}
asyncio.run(execute_with_wallet(wallet_rpc_port, fingerprint, extra_params, get_did_info))
@did_cmd.command("update_metadata", short_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,
)
@click.option("-f", "--fingerprint", help="Set the fingerprint to specify which key to use", type=int)
@click.option("-i", "--id", help="Id of the DID wallet to use", type=int, required=True)
@click.option("-m", "--metadata", help="The new whole metadata in json format", type=str, required=True)
@click.option("-r", "--reuse", help="Reuse existing address for the change.", is_flag=True, default=False)
def did_update_metadata_cmd(
wallet_rpc_port: Optional[int], fingerprint: int, id: int, metadata: str, reuse: bool
) -> None:
import asyncio
from .wallet_funcs import update_did_metadata
extra_params = {"did_wallet_id": id, "metadata": metadata, "reuse_puzhash": reuse}
asyncio.run(execute_with_wallet(wallet_rpc_port, fingerprint, extra_params, update_did_metadata))
@did_cmd.command("find_lost", short_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,
)
@click.option("-f", "--fingerprint", help="Set the fingerprint to specify which key to use", type=int)
@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,
)
def did_find_lost_cmd(
wallet_rpc_port: Optional[int],
fingerprint: int,
coin_id: str,
metadata: Optional[str],
recovery_list_hash: Optional[str],
num_verification: Optional[int],
) -> None:
import asyncio
from .wallet_funcs import find_lost_did
extra_params = {
"coin_id": coin_id,
"metadata": metadata,
"recovery_list_hash": recovery_list_hash,
"num_verification": num_verification,
}
asyncio.run(execute_with_wallet(wallet_rpc_port, fingerprint, extra_params, find_lost_did))
@did_cmd.command("message_spend", short_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,
)
@click.option("-f", "--fingerprint", help="Set the fingerprint to specify which key to use", type=int)
@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,
)
def did_message_spend_cmd(
wallet_rpc_port: Optional[int],
fingerprint: int,
id: int,
puzzle_announcements: Optional[str],
coin_announcements: Optional[str],
) -> None:
import asyncio
from .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
extra_params = {"did_wallet_id": id, "puzzle_announcements": puzzle_list, "coin_announcements": coin_list}
asyncio.run(execute_with_wallet(wallet_rpc_port, fingerprint, extra_params, did_message_spend))
@did_cmd.command("transfer", short_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,
)
@click.option("-f", "--fingerprint", help="Set the fingerprint to specify which key to use", type=int)
@click.option("-i", "--id", help="Id of the DID wallet to use", type=int, required=True)
@click.option("-ta", "--target-address", help="Target recipient wallet address", type=str, required=True)
@click.option(
"-r", "--reset_recovery", help="If you want to reset the recovery DID settings.", is_flag=True, default=False
)
@click.option(
"-m",
"--fee",
help="Set the fees per transaction, in XCH.",
type=str,
default="0",
show_default=True,
callback=validate_fee,
)
@click.option(
"-r",
"--reuse",
help="Reuse existing address for the change.",
is_flag=True,
default=False,
)
def did_trasnfer_did(
wallet_rpc_port: Optional[int],
fingerprint: int,
id: int,
target_address: str,
reset_recovery: bool,
fee: str,
reuse: bool,
) -> None:
import asyncio
from .wallet_funcs import transfer_did
extra_params = {
"did_wallet_id": id,
"with_recovery": reset_recovery is False,
"target_address": target_address,
"fee": fee,
"reuse_puzhash": True if reuse else None,
}
asyncio.run(execute_with_wallet(wallet_rpc_port, fingerprint, extra_params, transfer_did))
@wallet_cmd.group("nft", short_help="NFT related actions")
def nft_cmd():
pass

75
chia/cmds/wallet_funcs.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import json
import os
import pathlib
import sys
@ -830,6 +831,80 @@ async def get_did(args: Dict, wallet_client: WalletRpcClient, fingerprint: int)
print(f"Failed to get DID: {e}")
async def get_did_info(args: Dict, wallet_client: WalletRpcClient, fingerprint: int) -> None:
coin_id: str = args["coin_id"]
latest: bool = args["latest"]
did_padding_length = 23
try:
response = await wallet_client.get_did_info(coin_id, latest)
print(f"{'DID:'.ljust(did_padding_length)} {response['did_id']}")
print(f"{'Coin ID:'.ljust(did_padding_length)} {response['latest_coin']}")
print(f"{'Inner P2 Address:'.ljust(did_padding_length)} {response['p2_address']}")
print(f"{'Public Key:'.ljust(did_padding_length)} {response['public_key']}")
print(f"{'Launcher ID:'.ljust(did_padding_length)} {response['launcher_id']}")
print(f"{'DID Metadata:'.ljust(did_padding_length)} {response['metadata']}")
print(f"{'Recovery List Hash:'.ljust(did_padding_length)} {response['recovery_list_hash']}")
print(f"{'Recovery Required Verifications:'.ljust(did_padding_length)} {response['num_verification']}")
print(f"{'Last Spend Puzzle:'.ljust(did_padding_length)} {response['full_puzzle']}")
print(f"{'Last Spend Solution:'.ljust(did_padding_length)} {response['solution']}")
print(f"{'Last Spend Hints:'.ljust(did_padding_length)} {response['hints']}")
except Exception as e:
print(f"Failed to get DID details: {e}")
async def update_did_metadata(args: Dict, wallet_client: WalletRpcClient, fingerprint: int) -> None:
try:
response = await wallet_client.update_did_metadata(
args["did_wallet_id"], json.loads(args["metadata"]), args["reuse_puzhash"]
)
print(f"Successfully updated DID wallet ID: {response['wallet_id']}, Spend Bundle: {response['spend_bundle']}")
except Exception as e:
print(f"Failed to update DID metadata: {e}")
async def did_message_spend(args: Dict, wallet_client: WalletRpcClient, fingerprint: int) -> None:
try:
response = await wallet_client.did_message_spend(
args["did_wallet_id"], args["puzzle_announcements"], args["coin_announcements"]
)
print(f"Message Spend Bundle: {response['spend_bundle']}")
except Exception as e:
print(f"Failed to update DID metadata: {e}")
async def transfer_did(args: Dict, wallet_client: WalletRpcClient, fingerprint: int) -> None:
try:
response = await wallet_client.did_transfer_did(
args["did_wallet_id"],
args["target_address"],
args["fee"],
args["with_recovery"],
args["reuse_puzhash"],
)
print(f"Successfully transferred DID to {args['target_address']}")
print(f"Transaction ID: {response['transaction_id']}")
print(f"Transaction: {response['transaction']}")
except Exception as e:
print(f"Failed to transfer DID: {e}")
async def find_lost_did(args: Dict, wallet_client: WalletRpcClient, fingerprint: int) -> None:
try:
response = await wallet_client.find_lost_did(
args["coin_id"],
args["recovery_list_hash"],
args["metadata"],
args["num_verification"],
)
if response["success"]:
print(f"Successfully found lost DID {args['coin_id']}, latest coin ID: {response['latest_coin_id']}")
else:
print(f"Cannot find lost DID {args['coin_id']}: {response['error']}")
except Exception as e:
print(f"Failed to find lost DID: {e}")
async def create_nft_wallet(args: Dict, wallet_client: WalletRpcClient, fingerprint: int) -> None:
did_id = args["did_id"]
name = args["name"]

10
chia/rpc/wallet_rpc_api.py
View File

@ -1822,13 +1822,17 @@ class WalletRpcApi:
hints.append(memo.hex())
return {
"success": True,
"did_id": encode_puzzle_hash(
bytes32.from_hexstr(singleton_struct.rest().first().atom.hex()),
AddressType.DID.hrp(self.service.config),
),
"latest_coin": coin_state.coin.name().hex(),
"p2_address": encode_puzzle_hash(p2_puzzle.get_tree_hash(), AddressType.XCH.hrp(self.service.config)),
"public_key": public_key.as_python().hex(),
"recovery_list_hash": recovery_list_hash.as_python().hex(),
"public_key": public_key.atom.hex(),
"recovery_list_hash": recovery_list_hash.atom.hex(),
"num_verification": num_verification.as_int(),
"metadata": program_to_metadata(metadata),
"launcher_id": singleton_struct.rest().first().as_python().hex(),
"launcher_id": singleton_struct.rest().first().atom.hex(),
"full_puzzle": full_puzzle,
"solution": Program.from_bytes(bytes(coin_spend.solution)).as_python(),
"hints": hints,

34
chia/rpc/wallet_rpc_client.py
View File

@ -421,6 +421,14 @@ class WalletRpcClient(RpcClient):
response = await self.fetch("did_get_did", request)
return response
async def get_did_info(self, coin_id: str, latest: bool) -> Dict:
request: Dict[str, Any] = {
"coin_id": coin_id,
"latest": latest,
}
response = await self.fetch("did_get_info", request)
return response
async def create_did_backup_file(self, wallet_id: int, filename: str) -> Dict:
request: Dict[str, Any] = {
"wallet_id": wallet_id,
@ -452,6 +460,17 @@ class WalletRpcClient(RpcClient):
response = await self.fetch("did_get_recovery_list", request)
return response
async def did_message_spend(
self, wallet_id: int, puzzle_announcements: List[str], coin_announcements: List[str]
) -> Dict:
request: Dict[str, Any] = {
"wallet_id": wallet_id,
"coin_announcements": coin_announcements,
"puzzle_announcements": puzzle_announcements,
}
response = await self.fetch("did_message_spend", request)
return response
async def update_did_metadata(
self,
wallet_id: int,
@ -473,6 +492,21 @@ class WalletRpcClient(RpcClient):
response = await self.fetch("did_get_metadata", request)
return response
async def find_lost_did(
self, coin_id: str, recovery_list_hash: Optional[str], metadata: Optional[Dict], num_verification: Optional[int]
) -> Dict:
request: Dict[str, Any] = {
"coin_id": coin_id,
}
if recovery_list_hash is not None:
request["recovery_list_hash"] = recovery_list_hash
if metadata is not None:
request["metadata"] = (metadata,)
if num_verification is not None:
request["num_verification"] = num_verification
response = await self.fetch("did_find_lost_did", request)
return response
async def create_new_did_wallet_from_recovery(self, filename: str) -> Dict:
request: Dict[str, Any] = {
"wallet_type": "did_wallet",

2
tests/wallet/did_wallet/test_did.py
View File

@ -1017,7 +1017,7 @@ class TestDIDWallet:
assert await did_wallet_1.get_confirmed_balance() == did_amount
assert await did_wallet_1.get_unconfirmed_balance() == did_amount
response = await api_0.did_get_info({"coin_id": did_wallet_1.did_info.origin_coin.name().hex()})
assert response["did_id"] == encode_puzzle_hash(did_wallet_1.did_info.origin_coin.name(), AddressType.DID.value)
assert response["launcher_id"] == did_wallet_1.did_info.origin_coin.name().hex()
assert response["full_puzzle"] == create_singleton_puzzle(
did_wallet_1.did_info.current_inner, did_wallet_1.did_info.origin_coin.name()