7a17ffd12e
Open-source 2-of-3 policy-enforced threshold HSM: auto-signs cold→hot treasury refills under on-device Coldcard policy, no human in the loop. Includes the full operator manual + quick-start, the reference coordinator/signing code, and a signer-host bootstrap. No keys, seeds, or secrets — placeholders only. Live signet demo: https://multisighsm.mineracks.com Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
62 lines
3.1 KiB
Python
62 lines
3.1 KiB
Python
import subprocess, json, os, io, time, re
|
|
from base64 import b64decode, b64encode
|
|
from ckcc.client import ColdcardDevice
|
|
from ckcc.protocol import CCProtocolPacker
|
|
from ckcc.cli import real_file_upload
|
|
V='27.0'; DATA=os.path.expanduser('~/cksim/regtest-data')
|
|
CLI=os.path.expanduser('~/bitcoin-%s/bin/bitcoin-cli'%V)
|
|
def bc(*a, wallet=None):
|
|
base=[CLI,'-datadir=%s'%DATA,'-rpcport=18999','-rpcuser=ck','-rpcpassword=ckms']
|
|
if wallet: base.append('-rpcwallet=%s'%wallet)
|
|
r=subprocess.run(base+list(a),capture_output=True,text=True)
|
|
if r.returncode: raise SystemExit('%s FAIL %s'%(a,r.stderr.strip()))
|
|
return r.stdout.strip()
|
|
def dev(n): return ColdcardDevice(sn='/tmp/cksim-%d.sock'%n)
|
|
def E(d,c):
|
|
r=d.send_recv(b'EXEC'+c.encode(),encrypt=False); return r.decode() if isinstance(r,(bytes,bytearray)) else r
|
|
def K(d,k): d.send_recv(CCProtocolPacker.sim_keypress(k.encode('ascii')))
|
|
def hstat(d):
|
|
try:
|
|
r=d.send_recv(CCProtocolPacker.hsm_status()); r=r.decode() if isinstance(r,(bytes,bytearray)) else r
|
|
return json.loads(r)
|
|
except Exception as e: return {'err':str(e)[:60]}
|
|
def story(d): return E(d,"RV.write(repr(sim_display.story))")
|
|
def activate(d):
|
|
pf=os.path.expanduser('~/cksim/policy.json')
|
|
if hstat(d).get('active'): return True
|
|
flen,sha=real_file_upload(open(pf,'rb'),d)
|
|
d.send_recv(CCProtocolPacker.hsm_start(flen,sha)); time.sleep(1.4)
|
|
for _ in range(20):
|
|
if hstat(d).get('active'): return True
|
|
s=story(d)
|
|
m=re.search(r'Press \((\d)\) to save', s)
|
|
if m: K(d,m.group(1)); time.sleep(1.6); continue
|
|
if 'Press OK to enable' in s or s.startswith("('Start HSM?'"): K(d,'y'); time.sleep(1.6); continue
|
|
time.sleep(0.5)
|
|
return hstat(d).get('active')
|
|
d1,d2=dev(1),dev(2)
|
|
print('sim1 active(before)=', hstat(d1).get('active'))
|
|
print('sim2 activate ->', activate(d2))
|
|
print('sim1 active(now)=', hstat(d1).get('active'), ' sim2 active(now)=', hstat(d2).get('active'))
|
|
print('== build PSBT + AUTO-SIGN (no keypress) on sims 1 & 2 ==')
|
|
dest=bc('getnewaddress',wallet='miner')
|
|
funded=json.loads(bc('walletcreatefundedpsbt','[]',json.dumps([{dest:0.4}]),'0',json.dumps({'fee_rate':2,'change_type':'bech32'}),wallet='ckms23-watch'))
|
|
psbt=b64decode(funded['psbt']); sigs=[]
|
|
for n,d in ((1,d1),(2,d2)):
|
|
flen,sha=real_file_upload(io.BytesIO(psbt),d)
|
|
d.send_recv(CCProtocolPacker.sign_transaction(flen,sha,flags=0x0),timeout=None)
|
|
done=None
|
|
for i in range(40):
|
|
time.sleep(0.5)
|
|
r=d.send_recv(CCProtocolPacker.get_signed_txn(),timeout=None)
|
|
if r is not None: done=r; break
|
|
if not done: print('sim%d TIMEOUT hsm=%s'%(n,hstat(d).get('active'))); continue
|
|
rl,rsha=done; res=d.download_file(rl,rsha,file_number=1)
|
|
sigs.append(b64encode(res).decode()); print('sim%d AUTO-SIGNED no-keypress bytes=%d'%(n,rl))
|
|
fin=json.loads(bc('finalizepsbt', bc('combinepsbt',json.dumps(sigs))))
|
|
print('PSBT_COMPLETE:', fin['complete'])
|
|
if fin['complete']:
|
|
txid=bc('sendrawtransaction',fin['hex']); print('TXID:',txid)
|
|
bc('generatetoaddress','1',bc('getnewaddress',wallet='miner'),wallet='miner')
|
|
print('BALANCE_AFTER:',bc('getbalance',wallet='ckms23-watch'))
|