114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491 | class WalletPSBTSigningMixin:
"""Mixin implementing safe, partial signing of wallet-owned PSBT inputs."""
network: str
mixdepth_count: int
root_path: str
master_key: HDKey
def get_address(self, mixdepth: int, change: int, index: int) -> str: # pragma: no cover
raise NotImplementedError
def get_fidelity_bond_address(self, index: int, locktime: int) -> str: # pragma: no cover
raise NotImplementedError
def get_key_for_address(self, address: str) -> HDKey | None: # pragma: no cover
raise NotImplementedError
def sign_input(
self, tx: ParsedTransaction, input_index: int, utxo: UTXOInfo
) -> SignedInput: # pragma: no cover
raise NotImplementedError
def prepare_psbt_signing(self, psbt_bytes: bytes, scan_range: int) -> PSBTSigningPlan:
"""Parse a PSBT, establish ownership, and validate it without signing."""
if scan_range < 0 or scan_range > MAX_DERIVATION_INDEX:
raise PSBTError(f"PSBT scan range must be between 0 and {MAX_DERIVATION_INDEX:,}")
psbt = parse_psbt(psbt_bytes)
transaction = psbt.transaction
outpoints = [(tx_input.txid, tx_input.vout) for tx_input in transaction.inputs]
if len(set(outpoints)) != len(outpoints):
raise PSBTError("PSBT unsigned transaction contains duplicate input outpoints")
witness_utxos: list[WitnessUTXO] = []
input_types: list[PSBTInputType] = []
sighash_types: list[int] = []
for index, input_map in enumerate(psbt.input_maps):
witness_record = _record_for_type(input_map, PSBT_IN_WITNESS_UTXO)
if witness_record is None:
raise PSBTError(
f"PSBT input {index} is missing witness_utxo; all input amounts are required "
"for fee review"
)
witness_utxo = parse_witness_utxo(witness_record.value)
if witness_utxo.value > MAX_MONEY:
raise PSBTError(f"PSBT input {index} amount exceeds Bitcoin MAX_MONEY")
input_type = _classify_input_script(witness_utxo.script_pubkey)
if input_type == "unsupported":
raise PSBTError(
f"PSBT input {index} has an unsupported prevout script; all inputs must be "
"native P2WPKH or canonical JoinMarket fidelity bond P2WSH"
)
if input_type == "p2wsh":
_validate_fidelity_bond_shape(input_map, witness_utxo, index)
witness_utxos.append(witness_utxo)
input_types.append(input_type)
sighash_types.append(_parse_sighash_type(input_map, index))
for record in _records_for_type(input_map, PSBT_IN_BIP32_DERIVATION):
parse_bip32_derivation(record.key, record.value)
output_total = 0
for index, tx_output in enumerate(transaction.outputs):
if tx_output.value < 0 or tx_output.value > MAX_MONEY:
raise PSBTError(f"PSBT output {index} amount is outside Bitcoin's money range")
output_total += tx_output.value
input_total = sum(witness_utxo.value for witness_utxo in witness_utxos)
if output_total > input_total:
raise PSBTError(
f"PSBT outputs exceed inputs by {output_total - input_total:,} satoshis"
)
plans: list[PSBTInputPlan] = []
unresolved_regular: dict[bytes, list[int]] = {}
for index, (input_map, witness_utxo, input_type) in enumerate(
zip(psbt.input_maps, witness_utxos, input_types, strict=True)
):
finalized = _is_finalized(input_map)
plan: PSBTInputPlan | None = None
if input_type == "p2wpkh":
plan = self._plan_regular_input_from_origins(
psbt, index, input_map, witness_utxo, finalized
)
if plan is None:
unresolved_regular.setdefault(witness_utxo.script_pubkey, []).append(index)
elif input_type == "p2wsh":
plan = self._plan_fidelity_bond_input(
psbt, index, input_map, witness_utxo, finalized
)
plans.append(
plan
or PSBTInputPlan(
index=index,
witness_utxo=witness_utxo,
input_type=input_type,
finalized=finalized,
)
)
if unresolved_regular and scan_range:
self._resolve_regular_inputs_by_scan(psbt, plans, unresolved_regular, scan_range)
for plan, sighash_type in zip(plans, sighash_types, strict=True):
if plan.owned and not plan.finalized and sighash_type != 1:
raise PSBTError(
f"PSBT input {plan.index} requests unsupported sighash type "
f"{sighash_type:#x}; only SIGHASH_ALL (0x01) is allowed"
)
output_types = [_classify_output_script(output.script) for output in transaction.outputs]
if "unknown" in output_types:
unknown_index = output_types.index("unknown")
raise PSBTError(
f"PSBT output {unknown_index} has an unsupported script type; cannot calculate "
"a reliable fee rate"
)
estimated_vsize = estimate_vsize(
["p2wsh" if kind == "p2wsh" else "p2wpkh" for kind in input_types],
output_types,
)
# estimate_vsize assumes one-byte input/output counts. Account for the
# larger CompactSize encodings used at 253 and above.
estimated_vsize += len(encode_varint(len(transaction.inputs))) - 1
estimated_vsize += len(encode_varint(len(transaction.outputs))) - 1
return PSBTSigningPlan(
psbt=psbt,
inputs=tuple(plans),
fee=input_total - output_total,
estimated_vsize=estimated_vsize,
source_psbt=bytes(psbt_bytes),
scan_range=scan_range,
)
def sign_psbt(self, plan: PSBTSigningPlan) -> PSBTSigningResult:
"""Add partial signatures for every signable input in a reviewed plan."""
current_unsigned_tx = serialize_transaction(
plan.psbt.transaction.version,
plan.psbt.transaction.inputs,
plan.psbt.transaction.outputs,
plan.psbt.transaction.locktime,
)
if (
plan.psbt.serialize() != plan.source_psbt
or current_unsigned_tx != plan.psbt.unsigned_tx
):
raise TransactionSigningError("PSBT changed after review; refusing to sign")
# Rebuild from immutable source bytes so nested plan objects cannot be
# altered between review and private-key use.
signing_plan = self.prepare_psbt_signing(plan.source_psbt, plan.scan_range)
signed_indices: list[int] = []
already_signed_indices: list[int] = []
for input_plan in signing_plan.inputs:
if input_plan.already_signed:
already_signed_indices.append(input_plan.index)
continue
if not input_plan.signable:
continue
if input_plan.utxo is None or input_plan.pubkey is None:
raise TransactionSigningError(
f"Incomplete signing plan for PSBT input {input_plan.index}"
)
signed = self.sign_input(
signing_plan.psbt.transaction, input_plan.index, input_plan.utxo
)
if signed.pubkey != input_plan.pubkey:
raise TransactionSigningError(
f"Signing key changed after reviewing PSBT input {input_plan.index}"
)
signing_plan.psbt.append_input_key_value(
input_plan.index,
bytes([PSBT_IN_PARTIAL_SIG]) + signed.pubkey,
signed.signature,
)
signed_indices.append(input_plan.index)
return PSBTSigningResult(
psbt=signing_plan.psbt.serialize(),
signed_indices=tuple(signed_indices),
already_signed_indices=tuple(already_signed_indices),
)
def _plan_regular_input_from_origins(
self,
psbt: ParsedPSBT,
index: int,
input_map: PSBTMap,
witness_utxo: WitnessUTXO,
finalized: bool,
) -> PSBTInputPlan | None:
for record in _records_for_type(input_map, PSBT_IN_BIP32_DERIVATION):
origin = parse_bip32_derivation(record.key, record.value)
path = self._normal_wallet_path(origin.fingerprint, origin.path)
if path is None:
continue
mixdepth, change, address_index = path
address = self.get_address(mixdepth, change, address_index)
key = self.get_key_for_address(address)
if key is None:
continue
pubkey = key.get_public_key_bytes(compressed=True)
if pubkey != origin.pubkey:
continue
if pubkey_to_p2wpkh_script(pubkey) != witness_utxo.script_pubkey:
continue
return self._owned_regular_plan(
psbt,
index,
input_map,
witness_utxo,
address,
mixdepth,
change,
address_index,
finalized,
)
return None
def _plan_fidelity_bond_input(
self,
psbt: ParsedPSBT,
index: int,
input_map: PSBTMap,
witness_utxo: WitnessUTXO,
finalized: bool,
) -> PSBTInputPlan | None:
witness_script_record = _record_for_type(input_map, PSBT_IN_WITNESS_SCRIPT)
if witness_script_record is None:
return None
witness_script = witness_script_record.value
if b"\x00\x20" + sha256(witness_script).digest() != witness_utxo.script_pubkey:
raise PSBTError(
f"PSBT input {index} witness_script does not match its P2WSH witness_utxo"
)
try:
locktime, script_pubkey = parse_freeze_script(witness_script)
timenumber = timestamp_to_timenumber(locktime)
except ValueError:
return None
address = self.get_fidelity_bond_address(timenumber, locktime)
key = self.get_key_for_address(address)
if key is None:
return None
pubkey = key.get_public_key_bytes(compressed=True)
if pubkey != script_pubkey:
return None
tx_input = psbt.transaction.inputs[index]
if psbt.transaction.locktime < locktime:
raise PSBTError(
f"PSBT transaction locktime {psbt.transaction.locktime} is below fidelity bond "
f"input {index} locktime {locktime}"
)
if tx_input.sequence == 0xFFFFFFFF:
raise PSBTError(
f"PSBT fidelity bond input {index} has a final sequence and cannot satisfy CLTV"
)
utxo = UTXOInfo(
txid=tx_input.txid,
vout=tx_input.vout,
value=witness_utxo.value,
address=address,
confirmations=0,
scriptpubkey=witness_utxo.script_pubkey.hex(),
path=f"{self.root_path}/0'/2/{timenumber}",
mixdepth=0,
locktime=locktime,
)
already_signed = _validate_existing_signature(
psbt,
index,
input_map,
witness_utxo,
pubkey,
witness_script=witness_script,
)
return PSBTInputPlan(
index=index,
witness_utxo=witness_utxo,
input_type="p2wsh",
utxo=utxo,
pubkey=pubkey,
wallet_input_type="fidelity-bond",
already_signed=already_signed,
finalized=finalized,
)
def _resolve_regular_inputs_by_scan(
self,
psbt: ParsedPSBT,
plans: list[PSBTInputPlan],
unresolved: dict[bytes, list[int]],
scan_range: int,
) -> None:
for mixdepth in range(self.mixdepth_count):
for change in (0, 1):
for address_index in range(scan_range):
address = self.get_address(mixdepth, change, address_index)
key = self.get_key_for_address(address)
if key is None:
continue
pubkey = key.get_public_key_bytes(compressed=True)
script_pubkey = pubkey_to_p2wpkh_script(pubkey)
indices = unresolved.pop(script_pubkey, None)
if indices is None:
continue
for index in indices:
input_map = psbt.input_maps[index]
plans[index] = self._owned_regular_plan(
psbt,
index,
input_map,
plans[index].witness_utxo,
address,
mixdepth,
change,
address_index,
plans[index].finalized,
)
if not unresolved:
return
def _owned_regular_plan(
self,
psbt: ParsedPSBT,
index: int,
input_map: PSBTMap,
witness_utxo: WitnessUTXO,
address: str,
mixdepth: int,
change: int,
address_index: int,
finalized: bool,
) -> PSBTInputPlan:
key = self.get_key_for_address(address)
if key is None:
raise PSBTError(f"Wallet key disappeared while reviewing PSBT input {index}")
pubkey = key.get_public_key_bytes(compressed=True)
tx_input = psbt.transaction.inputs[index]
utxo = UTXOInfo(
txid=tx_input.txid,
vout=tx_input.vout,
value=witness_utxo.value,
address=address,
confirmations=0,
scriptpubkey=witness_utxo.script_pubkey.hex(),
path=f"{self.root_path}/{mixdepth}'/{change}/{address_index}",
mixdepth=mixdepth,
)
already_signed = _validate_existing_signature(psbt, index, input_map, witness_utxo, pubkey)
return PSBTInputPlan(
index=index,
witness_utxo=witness_utxo,
input_type="p2wpkh",
utxo=utxo,
pubkey=pubkey,
wallet_input_type="regular",
already_signed=already_signed,
finalized=finalized,
)
def _normal_wallet_path(
self, fingerprint: bytes, path: tuple[int, ...]
) -> tuple[int, int, int] | None:
coin_type = 0 if self.network == "mainnet" else 1
if fingerprint != self.master_key.fingerprint or len(path) != 5:
return None
purpose, coin, account, change, address_index = path
if purpose != 84 | HARDENED or coin != coin_type | HARDENED:
return None
if account < HARDENED or account - HARDENED >= self.mixdepth_count:
return None
if change not in (0, 1) or address_index >= MAX_DERIVATION_INDEX:
return None
return account - HARDENED, change, address_index
|