did-btcr2-py/beacon_manager_example.py at main · dcdpr/did-btcr2-py · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
from buidl.mnemonic import secure_mnemonic
from buidl.hd import HDPrivateKey
from buidl.tx import Tx
from buidl.helper import sha256
from pydid.verification_method import Multikey
from pydid.doc.builder import VerificationMethodBuilder
from pydid.did import DIDUrl
import json
import asyncio
from libbtc1.did import encode_identifier, decode_identifier, PLACEHOLDER_DID
from libbtc1.resolver import Btc1Resolver
from libbtc1.service import SingletonBeaconService
from libbtc1.diddoc.builder import Btc1DIDDocumentBuilder, IntermediateBtc1DIDDocumentBuilder
from libbtc1.multikey import get_public_key_multibase
import os
from bitcoinrpc import BitcoinRPC
from libbtc1.did_manager import DIDManager
from libbtc1.diddoc.builder import Btc1ServiceBuilder
from libbtc1.beacon_manager import BeaconManager

async def test_beacon_manager():

    didkey_purpose = "11"
    ## Run this if you want a new hardware key
    mnemonic = secure_mnemonic()

    # mnemonic = "prosper can dial lumber write coconut express imitate husband isolate inside release brush media please kind comic pill science repeat basic also endorse bronze"
    root_hdpriv = HDPrivateKey.from_mnemonic(mnemonic, network="regtest")
    print("Mnemonic : ", mnemonic)

    builder = IntermediateBtc1DIDDocumentBuilder(controller=[PLACEHOLDER_DID])

    auth_sk = root_hdpriv.get_private_key(didkey_purpose, address_num=0)
    auth_pk = auth_sk.point

    print("Secp256k1 PrivateKey", auth_sk.hex())
    print("Secp256k1 Public Key", auth_pk.sec())

    pk_multibase = get_public_key_multibase(auth_pk.sec())

    vm = builder.verification_method.add(Multikey, controller=PLACEHOLDER_DID, public_key_multibase = pk_multibase)

    builder.authentication.reference(vm.id)

    # builder.authentication.add(Multikey, public_key_multibase=pk_multibase)

    cap_sk = root_hdpriv.get_private_key(didkey_purpose, address_num=1)
    cap_pk = cap_sk.point

    print("Secp256k1 PrivateKey", cap_sk.hex())
    print("Secp256k1 Public Key", cap_pk.sec())


    pk_multibase = get_public_key_multibase(cap_pk.sec())

    cap_vm = builder.verification_method.add(Multikey, controller=PLACEHOLDER_DID, public_key_multibase=pk_multibase)

    builder.capability_invocation.reference(cap_vm.id)
    builder.capability_delegation.reference(cap_vm.id)

    beacon_sk = root_hdpriv.get_private_key(didkey_purpose, address_num=2)
    beacon_pk = beacon_sk.point

    script_pubkey = beacon_pk.p2wpkh_script()
    address = script_pubkey.address(network="regtest")

    print("Beacon Address", address, beacon_pk.p2wpkh_address(network="regtest"))
    beacon_service = builder.service.add_singleton_beacon(beacon_address=address)


    intermediate_doc = builder.build()


    rpc_endpoint = "http://localhost:18443"
    network = "regtest"

    bitcoinrpc = BitcoinRPC.from_config(rpc_endpoint, ("polaruser", "polarpass"))

    did_manager = DIDManager(network=network, rpc_endpoint=rpc_endpoint, rpcuser="polaruser", rpcpassword="polarpass")

    identifier, did_doc = did_manager.create_external(intermediate_doc, network)

    beacon_service = did_doc.service[0]

    beacon_manager = BeaconManager(bitcoinrpc, network, beacon_service.id, beacon_sk, script_pubkey)

    result = await bitcoinrpc.acall("send", {"outputs": { beacon_service.address(): 0.2}})

    funding_txid = result["txid"]
    funding_tx_hex = await bitcoinrpc.acall("getrawtransaction", {"txid": funding_txid})
    funding_tx = Tx.parse_hex(funding_tx_hex)

    beacon_manager.add_funding_tx(funding_tx)


    test_commitment = sha256(b'TESTBEACONCOMMITMENT')

    pending_signal = beacon_manager.construct_beacon_signal(test_commitment)

    signed_signal = beacon_manager.sign_beacon_signal(pending_signal)

    signed_hex = signed_signal.serialize().hex()
    signal_id = await bitcoinrpc.acall("sendrawtransaction", {"hexstring": signed_hex})

    print("Successfully broadcast beacon signal", signal_id)
    print("UTXOs", len(beacon_manager.utxos))
    print(beacon_manager.utxos)

asyncio.run(test_beacon_manager())