Add parsing of Musig2 pubnonces and partial signatures as yielded val…

…ues in sign_psbt in the python client

bitcoin_client/ledger_bitcoin/client.py

@@ -10,8 +10,8 @@
 
 from .command_builder import BitcoinCommandBuilder, BitcoinInsType
 from .common import Chain, read_uint, read_varint
-from .client_command import ClientCommandInterpreter
-from .client_base import Client, TransportClient, PartialSignature
+from .client_command import ClientCommandInterpreter, CCMD_YIELD_MUSIG_PARTIALSIGNATURE_TAG, CCMD_YIELD_MUSIG_PUBNONCE_TAG
+from .client_base import Client, MusigPartialSignature, MusigPubNonce, SignPsbtYieldedObject, TransportClient, PartialSignature
 from .client_legacy import LegacyClient
 from .exception import DeviceException
 from .errors import UnknownDeviceError
@@ -105,6 +105,60 @@ def _make_partial_signature(pubkey_augm: bytes, signature: bytes) -> PartialSign
         return PartialSignature(signature=signature, pubkey=pubkey_augm)
 
 
+def _decode_signpsbt_yielded_value(res: bytes) -> Tuple[int, SignPsbtYieldedObject]:
+    res_buffer = BytesIO(res)
+    input_index_or_tag = read_varint(res_buffer)
+    if input_index_or_tag == CCMD_YIELD_MUSIG_PUBNONCE_TAG:
+        input_index = read_varint(res_buffer)
+        pubnonce = res_buffer.read(66)
+        participant_pk = res_buffer.read(33)
+        agg_xonlykey = res_buffer.read(33)
+        tapleaf_hash = res_buffer.read()
+        if len(tapleaf_hash) == 0:
+            tapleaf_hash = None
+
+        return (
+            input_index,
+            MusigPubNonce(
+                participant_pubkey=participant_pk,
+                agg_xonlykey=agg_xonlykey,
+                tapleaf_hash=tapleaf_hash,
+                pubnonce=pubnonce
+            )
+        )
+    elif input_index_or_tag == CCMD_YIELD_MUSIG_PARTIALSIGNATURE_TAG:
+        input_index = read_varint(res_buffer)
+        participant_pk = res_buffer.read(33)
+        agg_xonlykey = res_buffer.read(33)
+        partial_signature = res_buffer.read(32)
+        tapleaf_hash = res_buffer.read()
+        if len(tapleaf_hash) == 0:
+            tapleaf_hash = None
+
+        return (
+            input_index,
+            MusigPartialSignature(
+                participant_pubkey=participant_pk,
+                agg_xonlykey=agg_xonlykey,
+                tapleaf_hash=tapleaf_hash,
+                partial_signature=partial_signature
+            )
+        )
+    else:
+        # other values follow an encoding without an explicit tag, where the
+        # first element is the input index. All the signature types are implemented
+        # by the PartialSignature type (not to be confused with the musig Partial Signature).
+        input_index = input_index_or_tag
+
+        pubkey_augm_len = read_uint(res_buffer, 8)
+        pubkey_augm = res_buffer.read(pubkey_augm_len)
+
+        signature = res_buffer.read()
+
+        return((input_index, _make_partial_signature(pubkey_augm, signature)))
+
+
+
 class NewClient(Client):
     # internal use for testing: if set to True, sign_psbt will not clone the psbt before converting to psbt version 2
     _no_clone_psbt: bool = False
@@ -211,7 +265,7 @@ def get_wallet_address(
 
         return result
 
-    def sign_psbt(self, psbt: Union[PSBT, bytes, str], wallet: WalletPolicy, wallet_hmac: Optional[bytes]) -> List[Tuple[int, PartialSignature]]:
+    def sign_psbt(self, psbt: Union[PSBT, bytes, str], wallet: WalletPolicy, wallet_hmac: Optional[bytes]) -> List[Tuple[int, SignPsbtYieldedObject]]:
 
         psbt = normalize_psbt(psbt)
 
@@ -280,17 +334,10 @@ def sign_psbt(self, psbt: Union[PSBT, bytes, str], wallet: WalletPolicy, wallet_
         if any(len(x) <= 1 for x in results):
             raise RuntimeError("Invalid response")
 
-        results_list: List[Tuple[int, PartialSignature]] = []
+        results_list: List[Tuple[int, SignPsbtYieldedObject]] = []
         for res in results:
-            res_buffer = BytesIO(res)
-            input_index = read_varint(res_buffer)
-
-            pubkey_augm_len = read_uint(res_buffer, 8)
-            pubkey_augm = res_buffer.read(pubkey_augm_len)
-
-            signature = res_buffer.read()
-
-            results_list.append((input_index, _make_partial_signature(pubkey_augm, signature)))
+            input_index, obj = _decode_signpsbt_yielded_value(res)
+            results_list.append((input_index, obj))
 
         return results_list
 

bitcoin_client/ledger_bitcoin/client_base.py

@@ -28,7 +28,8 @@ def __init__(self, sw: int, data: bytes) -> None:
 
 class TransportClient:
     def __init__(self, interface: Literal['hid', 'tcp'] = "tcp", *, server: str = "127.0.0.1", port: int = 9999, path: Optional[str] = None, hid: Optional[HID] = None, debug: bool = False):
-        self.transport = Transport('hid', path=path, hid=hid, debug=debug) if interface == 'hid' else Transport(interface, server=server, port=port, debug=debug)
+        self.transport = Transport('hid', path=path, hid=hid, debug=debug) if interface == 'hid' else Transport(
+            interface, server=server, port=port, debug=debug)
 
     def apdu_exchange(
         self, cla: int, ins: int, data: bytes = b"", p1: int = 0, p2: int = 0
@@ -67,18 +68,60 @@ def print_response(sw: int, data: bytes) -> None:
 
 @dataclass(frozen=True)
 class PartialSignature:
-    """Represents a partial signature returned by sign_psbt.
+    """Represents a partial signature returned by sign_psbt. Such objects can be added to the PSBT.
 
     It always contains a pubkey and a signature.
-    The pubkey
+    The pubkey is a compressed 33-byte for legacy and segwit Scripts, or 32-byte x-only key for taproot.
+    The signature is in the format it would be pushed on the scriptSig or the witness stack, therefore of
+    variable length, and possibly concatenated with the SIGHASH flag byte if appropriate.
 
-    The tapleaf_hash is also filled if signing a for a tapscript.
+    The tapleaf_hash is also filled if signing for a tapscript.
+
+    Note: not to be confused with 'partial signature' of protocols like MuSig2; 
     """
     pubkey: bytes
     signature: bytes
     tapleaf_hash: Optional[bytes] = None
 
 
+@dataclass(frozen=True)
+class MusigPubNonce:
+    """Represents a pubnonce returned by sign_psbt during the first round of a Musig2 signing session.
+
+    It always contains
+    - the participant_pubkey, a 33-byte compressed pubkey;
+    - agg_xonlykey, the 32-byte xonly key that is the aggregate and tweaked key present in the script;
+    - the 66-byte pubnonce.
+
+    The tapleaf_hash is also filled if signing for a tapscript; `None` otherwise.
+    """
+    participant_pubkey: bytes
+    agg_xonlykey: bytes
+    tapleaf_hash: Optional[bytes]
+    pubnonce: bytes
+
+
+@dataclass(frozen=True)
+class MusigPartialSignature:
+    """Represents a partial signature returned by sign_psbt during the second round of a Musig2 signing session.
+
+    It always contains
+    - the participant_pubkey, a 33-byte compressed pubkey;
+    - agg_xonlykey, the 32-byte xonly key that is the aggregate and tweaked key present in the script;
+    - the partial_signature, the 32-byte partial signature for this participant.
+
+    The tapleaf_hash is also filled if signing for a tapscript; `None` otherwise
+    """
+    participant_pubkey: bytes
+    agg_xonlykey: bytes
+    tapleaf_hash: Optional[bytes]
+    partial_signature: bytes
+
+
+SignPsbtYieldedObject = Union[PartialSignature,
+                              MusigPubNonce, MusigPartialSignature]
+
+
 class Client:
     def __init__(self, transport_client: TransportClient, chain: Chain = Chain.MAIN, debug: bool = False) -> None:
         self.transport_client = transport_client
@@ -218,7 +261,7 @@ def get_wallet_address(
 
         raise NotImplementedError
 
-    def sign_psbt(self, psbt: Union[PSBT, bytes, str], wallet: WalletPolicy, wallet_hmac: Optional[bytes]) -> List[Tuple[int, PartialSignature]]:
+    def sign_psbt(self, psbt: Union[PSBT, bytes, str], wallet: WalletPolicy, wallet_hmac: Optional[bytes]) -> List[Tuple[int, SignPsbtYieldedObject]]:
         """Signs a PSBT using a registered wallet (or a standard wallet that does not need registration).
 
         Signature requires explicit approval from the user.

bitcoin_client/ledger_bitcoin/client_command.py

@@ -15,6 +15,10 @@ class ClientCommandCode(IntEnum):
     GET_MORE_ELEMENTS = 0xA0
 
 
+CCMD_YIELD_MUSIG_PUBNONCE_TAG = 0xFFFFFFFF
+CCMD_YIELD_MUSIG_PARTIALSIGNATURE_TAG = 0xFFFFFFFE
+
+
 class ClientCommand:
     def execute(self, request: bytes) -> bytes:
         raise NotImplementedError("Subclasses should implement this method.")