diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index f59df4da..20e6cac5 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -65,13 +65,14 @@ jobs: ./autogen.sh ./configure --disable-coverage --disable-benchmark --disable-tests \ --disable-exhaustive-tests --enable-module-recovery \ - --enable-module-ecdh --enable-module-schnorrsig + --enable-module-ecdh --enable-module-schnorrsig --enable-module-musig make sudo make install else cmake -B build -DSECP256K1_ENABLE_MODULE_RECOVERY=ON \ -DSECP256K1_ENABLE_MODULE_ECDH=ON \ -DSECP256K1_ENABLE_MODULE_SCHNORRSIG=ON \ + -DSECP256K1_ENABLE_MODULE_MUSIG=ON \ -DSECP256K1_BUILD_TESTS=OFF \ -DSECP256K1_BUILD_BENCHMARK=OFF \ -DSECP256K1_BUILD_EXHAUSTIVE_TESTS=OFF diff --git a/README.md b/README.md index 448c1a5b..fd4d27ff 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,8 @@ a focus on correctness, consistency, and developer ergonomics. * Some API have changed and may be not compatible with old code (see below) * libsecp256k1 is used for signing and verifying. Signing by libsecp256k1 is deterministic, per RFC6979. +* Native BIP327 MuSig2 support is available when libsecp256k1 is built with + its `musig` module (enabled by default since v0.6.0); it does not use a Python backend. * Support for PSBT (BIP174 Partially-signed transactions) * HD keys support * Easier to build code that supports and interacts with other bitcoin-based blockchains @@ -38,17 +40,34 @@ the library and v1.0.0 release in particular, and also has some code examples. ## Requirements - Python >= 3.11 -- [libsecp256k1](https://github.com/bitcoin-core/secp256k1) +- [libsecp256k1](https://github.com/bitcoin-core/secp256k1) >= 0.3.0, < 1.0.0 - [libbitcoinconsensus](https://github.com/bitcoin/bitcoin/blob/master/doc/shared-libraries.md) (optional, for consensus-compatible script verification) -Tests use the following libsecp256k1 versions: +Native feature requirements within that supported version range: + +| Feature | Minimum libsecp256k1 | Required modules | +| --- | --- | --- | +| ECDSA and ordinary key operations | 0.3.0 | Core library | +| Taproot keys/tweaks and BIP340 Schnorr signing/verification | 0.3.0 | `extrakeys`, `schnorrsig` | +| BIP327 MuSig2 | 0.6.0 | `musig` (also enables `schnorrsig` and `extrakeys`) | + +The Taproot primitives were already present in upstream v0.2.0, but this +package's supported baseline is v0.3.0. ECDH and recoverable ECDSA signatures +additionally require the `ecdh` and `recovery` modules, respectively. + +CI tests use the following libsecp256k1 versions: [//]: # (!LIBSECP256K1_VERSION_MARKER_DO_NOT_MOVE_OR_EDIT! this marker is used by automatic tests to extract the version that is in the following line from this README.md, and use it to run tests with this specific version of libsecp256k1) `v0.4.0`, `v0.7.0`, and `v0.8.0` -Libsecp256k1 is not linked as a git submodule in python-bitcointx git repository, because python-bitcointx -can still be used with other versions of libsecp256k1 as long as experimental modules with unstable ABI -of are not used, or are compatible with the vesion listed above. +MuSig2 tests run only when the loaded library provides the module; v0.4.0 +covers compatibility without MuSig2. The bindings support the v0.8.0 removal +of the deprecated `secp256k1_schnorrsig_sign` symbol by using `sign32`. + +Libsecp256k1 is loaded dynamically, not bundled or installed by pip. Forks and +experimental builds must provide a compatible ABI, not just matching symbol +names. Native opaque objects must not be persisted or exchanged between +library versions or platforms. While allowing dynamic linkage with libsecp256k1 adds these complications, it is at the same time allows more flexibility for advanced uses. For example, one can use libsecp256k1-zkp instead of libsecp256k1 to @@ -58,6 +77,75 @@ For best results, use one of the versions listed above, as these are the version use to build libsecp256k1. Then make sure that this version of the library is loaded by python-bitcointx, by using `bitcointx.set_custom_secp256k1_path()` or `LD_LIBRARY_PATH ` environment variable. +### MuSig2 in Distribution Packages + +MuSig2 was [introduced in libsecp256k1 v0.6.0](https://github.com/bitcoin-core/secp256k1/blob/v0.6.0/CHANGELOG.md). +It is enabled by default in both upstream +[CMake](https://github.com/bitcoin-core/secp256k1/blob/v0.6.0/CMakeLists.txt) +and [Autotools](https://github.com/bitcoin-core/secp256k1/blob/v0.6.0/configure.ac). +An explicit build can use `-DSECP256K1_ENABLE_MODULE_MUSIG=ON` with CMake or +`--enable-module-musig` with Autotools where available. + +Do not assume every distribution package has MuSig2. Examples checked in +September 2026 (stock packages, excluding backports and third-party builds): + +| Distribution | libsecp256k1 version | MuSig2 | +| --- | --- | --- | +| [Debian 13 (trixie)](https://tracker.debian.org/pkg/libsecp256k1) | 0.5.0-2 | Unavailable in this upstream version | +| [Ubuntu 24.04 LTS](https://packages.ubuntu.com/noble/libsecp256k1-dev) | 0.2.0-2 | Unavailable; also below this package's supported baseline | +| [Ubuntu 26.04 LTS](https://git.launchpad.net/ubuntu/+source/libsecp256k1/tree/debian/rules?h=applied/ubuntu/resolute) | 0.7.0-2 | Enabled by the upstream default | +| [Fedora 44](https://packages.fedoraproject.org/pkgs/libsecp256k1/libsecp256k1-devel/fedora-44.html) | 0.6.0-4.fc44 | Enabled by the upstream default | + +Check the library actually loaded by Python, not just the installed headers: + +```python +from bitcointx.core.secp256k1 import get_secp256k1 + +secp = get_secp256k1() +print(secp.cap.has_xonly_pubkeys, secp.cap.has_schnorrsig, secp.cap.has_musig) +``` + +### Taproot and MuSig2 Scope + +Taproot support includes x-only keys, Schnorr signatures, P2TR addresses, +signature hashes, and `TaprootScriptTree`. As discussed in +[issue #57](https://github.com/Simplexum/python-bitcointx/issues/57), this is +not complete Taproot support: `core.psbt` does not implement Taproot input +signing or BIP371-aware signature combining, and `VerifyScript` does not +validate Taproot spends. + +`bitcointx.core.musig` provides low-level BIP327 key/nonce aggregation, x-only +tweaks, and partial signing/verification for 32-byte message hashes. It does +not implement plain EC tweaks, arbitrary-length messages, a signing transport, +or MuSig2 PSBT coordination. Construct opaque contexts, sessions, and secret +nonces through `key_agg`, `get_session`, and `nonce_gen`, not their constructors. +Callers must agree on participant order, derive TapTweak hashes according to +BIP341, and verify the final signature with `XOnlyPubKey.verify_schnorr`. +Partial signature aggregation alone does not verify the result. Verify your +own partial signature before sharing it, as recommended by libsecp256k1. +`partial_sig_verify` raises `MuSig2Error` for malformed contributions or an +invalid session argument; well-formed contributions that fail verification +return `False`. Coordinators should handle this exception separately for each +peer's contribution. + +**Nonce safety:** use `nonce_gen(pubkey)` for fresh operating-system randomness. +When known, pass `privkey=`, `msg32=`, and `extra_input32=` to include the signing +key, message, and extra input in nonce derivation. These optional inputs are +32-byte `bytes` values; `privkey` must correspond to `pubkey`. They do not +replace the requirement for fresh randomness. +The optional `rand` is secret nonce-generation material, not public auxiliary +randomness: it must be uniformly random, kept secret, and never reused, even +after a failed or abandoned signing attempt. Fixed values in tests are not +production examples. `rand` also accepts a 32-byte `bytearray`, which is wiped +in place once input validation succeeds, including when generation fails. +Do not share or mutate that buffer during the call. Previously supplied +randomness is not tracked globally. Each `SecNonce` is consumed on the first +signing attempt, including failures, and cannot be copied or pickled. Do not +fork a process or restore a memory snapshot containing live secret nonces: process copies can +bypass per-object single-use protection. Native secret buffers are wiped on +consumption, but immutable Python `bytes` (including supplied randomness and +private keys) cannot be reliably erased by this wrapper. + ## Installation ``` @@ -81,6 +169,7 @@ consensus critical and non-consensus-critical. bitcointx.core - Basic core definitions, datastructures, and (context-independent) validation bitcointx.core.key - ECC keys, BIP32Paths + bitcointx.core.musig - Native BIP327 MuSig2 signing primitives bitcointx.core.script - Scripts and opcodes bitcointx.core.scripteval - Script evaluation/verification bitcointx.core.psbt - BIP174 Partially-signed transactions diff --git a/bitcointx/core/musig.py b/bitcointx/core/musig.py new file mode 100644 index 00000000..e9e6af42 --- /dev/null +++ b/bitcointx/core/musig.py @@ -0,0 +1,687 @@ +# Copyright (C) 2026 The python-bitcointx developers +# +# This file is part of python-bitcointx. +# +# It is subject to the license terms in the LICENSE file found in the top-level +# directory of this distribution. + +"""Native BIP327 MuSig2 wrappers for libsecp256k1's v0.6.0+ musig module. + +These primitives support 32-byte messages and x-only tweaks. See the README's +Taproot and MuSig2 scope section for protocol responsibilities and nonce safety. +""" + +import ctypes +import os +import threading +from typing import Any, Sequence, SupportsIndex, Tuple, cast + +from bitcointx.core.secp256k1 import ( + SECP256K1_EC_COMPRESSED, + Secp256k1, + get_secp256k1, +) + + +KEYAGG_CACHE_SIZE = 197 +SECNONCE_SIZE = 132 +PUBNONCE_SIZE = 132 +AGGNONCE_SIZE = 132 +SESSION_SIZE = 133 +PARTIAL_SIG_SIZE = 36 + + +class MuSig2Error(ValueError): + """Raised when a MuSig2 contribution or session is invalid.""" + + +def _zero_buffer(buffer: Any, size: int) -> None: + ctypes.memset(buffer, 0, size) + + +def _require_musig() -> Secp256k1: + secp256k1 = get_secp256k1() + if not secp256k1.cap.has_musig: + raise MuSig2Error( + "libsecp256k1 MuSig2 support is unavailable; rebuild it with --enable-module-musig" + ) + return secp256k1 + + +def _require_bytes(value: object, size: int, name: str) -> bytes: + if not isinstance(value, bytes): + raise MuSig2Error(f"{name} must be bytes") + if len(value) != size: + raise MuSig2Error(f"{name} must be exactly {size} bytes long") + return value + + +def _require_bytes_sequence(value: object, name: str) -> Tuple[bytes, ...]: + if not isinstance(value, Sequence) or isinstance(value, (bytes, bytearray, str)): + raise MuSig2Error(f"{name} must be a sequence of bytes") + if not value: + raise MuSig2Error(f"{name} must not be empty") + return tuple(cast(bytes, item) for item in value) + + +def _validate_scalar(secp256k1: Secp256k1, value: object, name: str) -> bytes: + scalar = _require_bytes(value, 32, name) + if secp256k1.lib.secp256k1_ec_seckey_verify(secp256k1.ctx.sign, scalar) != 1: + raise MuSig2Error(f"{name} is not a valid secp256k1 scalar") + return scalar + + +def _parse_pubkey(secp256k1: Secp256k1, value: object, name: str) -> Tuple[bytes, Any]: + pubkey = _require_bytes(value, 33, name) + if pubkey[0] not in (2, 3): + raise MuSig2Error(f"{name} must use compressed serialization") + + parsed = ctypes.create_string_buffer(64) + if ( + secp256k1.lib.secp256k1_ec_pubkey_parse(secp256k1.ctx.verify, parsed, pubkey, len(pubkey)) + != 1 + ): + _zero_buffer(parsed, 64) + raise MuSig2Error(f"{name} is not a valid secp256k1 public key") + return pubkey, parsed + + +def _parse_pubnonce(secp256k1: Secp256k1, value: object, name: str) -> Any: + pubnonce = _require_bytes(value, 66, name) + parsed = ctypes.create_string_buffer(PUBNONCE_SIZE) + if secp256k1.lib.secp256k1_musig_pubnonce_parse(secp256k1.ctx.verify, parsed, pubnonce) != 1: + _zero_buffer(parsed, PUBNONCE_SIZE) + raise MuSig2Error(f"{name} is not a valid MuSig2 public nonce") + return parsed + + +def _parse_aggnonce(secp256k1: Secp256k1, value: object, name: str) -> Any: + aggnonce = _require_bytes(value, 66, name) + parsed = ctypes.create_string_buffer(AGGNONCE_SIZE) + if secp256k1.lib.secp256k1_musig_aggnonce_parse(secp256k1.ctx.verify, parsed, aggnonce) != 1: + _zero_buffer(parsed, AGGNONCE_SIZE) + raise MuSig2Error(f"{name} is not a valid MuSig2 aggregate nonce") + return parsed + + +def _parse_partial_sig(secp256k1: Secp256k1, value: object, name: str) -> Any: + partial_sig = _require_bytes(value, 32, name) + parsed = ctypes.create_string_buffer(PARTIAL_SIG_SIZE) + if ( + secp256k1.lib.secp256k1_musig_partial_sig_parse(secp256k1.ctx.verify, parsed, partial_sig) + != 1 + ): + _zero_buffer(parsed, PARTIAL_SIG_SIZE) + raise MuSig2Error(f"{name} is not a valid MuSig2 partial signature") + return parsed + + +def _serialize_pubnonce(secp256k1: Secp256k1, pubnonce: Any) -> bytes: + serialized = ctypes.create_string_buffer(66) + try: + if ( + secp256k1.lib.secp256k1_musig_pubnonce_serialize( + secp256k1.ctx.verify, serialized, pubnonce + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not serialize the MuSig2 public nonce") + return serialized.raw + finally: + _zero_buffer(serialized, 66) + + +def _serialize_aggnonce(secp256k1: Secp256k1, aggnonce: Any) -> bytes: + serialized = ctypes.create_string_buffer(66) + try: + if ( + secp256k1.lib.secp256k1_musig_aggnonce_serialize( + secp256k1.ctx.verify, serialized, aggnonce + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not serialize the MuSig2 aggregate nonce") + return serialized.raw + finally: + _zero_buffer(serialized, 66) + + +def _serialize_partial_sig(secp256k1: Secp256k1, partial_sig: Any) -> bytes: + serialized = ctypes.create_string_buffer(32) + try: + if ( + secp256k1.lib.secp256k1_musig_partial_sig_serialize( + secp256k1.ctx.verify, serialized, partial_sig + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not serialize the MuSig2 partial signature") + return serialized.raw + finally: + _zero_buffer(serialized, 32) + + +def _serialize_pubkey(secp256k1: Secp256k1, pubkey: Any) -> bytes: + serialized = ctypes.create_string_buffer(33) + size = ctypes.c_size_t(33) + try: + if ( + secp256k1.lib.secp256k1_ec_pubkey_serialize( + secp256k1.ctx.verify, + serialized, + ctypes.byref(size), + pubkey, + SECP256K1_EC_COMPRESSED, + ) + != 1 + or size.value != 33 + ): + raise MuSig2Error("libsecp256k1 could not serialize the aggregate public key") + return serialized.raw + finally: + _zero_buffer(serialized, 33) + + +def _serialize_xonly_pubkey(secp256k1: Secp256k1, pubkey: Any) -> bytes: + xonly = ctypes.create_string_buffer(64) + serialized = ctypes.create_string_buffer(32) + try: + if ( + secp256k1.lib.secp256k1_xonly_pubkey_from_pubkey( + secp256k1.ctx.verify, xonly, None, pubkey + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not convert the aggregate public key") + if ( + secp256k1.lib.secp256k1_xonly_pubkey_serialize(secp256k1.ctx.verify, serialized, xonly) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not serialize the aggregate x-only key") + return serialized.raw + finally: + _zero_buffer(xonly, 64) + _zero_buffer(serialized, 32) + + +def _opaque_ptr_array(buffers: Sequence[Any]) -> Any: + pointers = (ctypes.c_void_p * len(buffers))() + for index, buffer in enumerate(buffers): + pointers[index] = ctypes.addressof(buffer) + return pointers + + +class KeyAggContext: + """BIP327 key aggregation state, backed by a native opaque cache.""" + + __slots__ = ("_cache", "_participants", "_aggregate_xonly", "_aggregate_pubkey") + + def __init__( + self, + cache: Any, + participants: Tuple[bytes, ...], + aggregate_xonly: bytes, + aggregate_pubkey: bytes, + ) -> None: + self._cache = cache + self._participants = participants + self._aggregate_xonly = aggregate_xonly + self._aggregate_pubkey = aggregate_pubkey + + def aggregate_xonly(self) -> bytes: + """The 32-byte BIP340 aggregate public key.""" + + return self._aggregate_xonly + + def aggregate_pubkey(self) -> bytes: + """The compressed 33-byte aggregate public key.""" + + return self._aggregate_pubkey + + def _copy_cache(self) -> Any: + copied = ctypes.create_string_buffer(KEYAGG_CACHE_SIZE) + ctypes.memmove(copied, self._cache, KEYAGG_CACHE_SIZE) + return copied + + +class Session: + """Native MuSig2 signing session bound to its participants and context.""" + + __slots__ = ("ctx", "_session", "_participants") + + def __init__(self, ctx: KeyAggContext, session: Any, participants: Tuple[bytes, ...]) -> None: + self.ctx = ctx + self._session = session + self._participants = participants + + +class SecNonce: + """A single-use native MuSig2 secret nonce. + + The native secnonce structure is never serialized or exposed. It is wiped + after every signing attempt, including one that raises an exception. + Obtain instances from nonce_gen, not this constructor. Do not fork or + restore a process snapshot containing a live nonce: that copies its state. + """ + + __slots__ = ("__buffer", "__pubkey", "__used", "__lock") + + def __init__(self, buffer: Any, pubkey: bytes) -> None: + self.__buffer = buffer + self.__pubkey = pubkey + self.__used = False + self.__lock = threading.Lock() + + def __repr__(self) -> str: + return "SecNonce()" + + __str__ = __repr__ + + def __copy__(self) -> "SecNonce": + raise TypeError("SecNonce instances cannot be copied") + + def __deepcopy__(self, memo: object) -> "SecNonce": + raise TypeError("SecNonce instances cannot be copied") + + def __reduce__(self) -> str | tuple[Any, ...]: + raise TypeError("SecNonce instances cannot be pickled") + + def __reduce_ex__(self, protocol: SupportsIndex) -> str | tuple[Any, ...]: + raise TypeError("SecNonce instances cannot be pickled") + + def __getstate__(self) -> object: + raise TypeError("SecNonce instances cannot be pickled") + + def __del__(self) -> None: + try: + _zero_buffer(self.__buffer, SECNONCE_SIZE) + except (AttributeError, TypeError): + pass + + def _sign(self, privkey: object, session: object) -> bytes: + with self.__lock: + if self.__used: + raise MuSig2Error("MuSig2 secret nonce has already been consumed") + self.__used = True + + keypair = signer_pubkey = partial_sig = None + try: + keypair = ctypes.create_string_buffer(96) + signer_pubkey = ctypes.create_string_buffer(64) + partial_sig = ctypes.create_string_buffer(PARTIAL_SIG_SIZE) + secp256k1 = _require_musig() + if not isinstance(session, Session): + raise MuSig2Error("session must be a MuSig2 Session") + + secret = _validate_scalar(secp256k1, privkey, "privkey") + if secp256k1.lib.secp256k1_keypair_create(secp256k1.ctx.sign, keypair, secret) != 1: + raise MuSig2Error("libsecp256k1 could not create a keypair") + if ( + secp256k1.lib.secp256k1_ec_pubkey_create( + secp256k1.ctx.sign, signer_pubkey, secret + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not derive the signer public key") + + signer_serialized = _serialize_pubkey(secp256k1, signer_pubkey) + if signer_serialized != self.__pubkey: + raise MuSig2Error("MuSig2 secret nonce is bound to a different public key") + if signer_serialized not in session._participants: + raise MuSig2Error("signer public key is not a session participant") + + if ( + secp256k1.lib.secp256k1_musig_partial_sign( + secp256k1.ctx.sign, + partial_sig, + self.__buffer, + keypair, + session.ctx._cache, + session._session, + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 MuSig2 partial signing failed") + return _serialize_partial_sig(secp256k1, partial_sig) + finally: + _zero_buffer(self.__buffer, SECNONCE_SIZE) + if partial_sig is not None: + _zero_buffer(partial_sig, PARTIAL_SIG_SIZE) + if signer_pubkey is not None: + _zero_buffer(signer_pubkey, 64) + if keypair is not None: + _zero_buffer(keypair, 96) + + +def key_agg(pubkeys: Sequence[bytes]) -> KeyAggContext: + """Aggregate compressed BIP327 participant public keys without sorting them.""" + + secp256k1 = _require_musig() + values = _require_bytes_sequence(pubkeys, "pubkeys") + parsed: list[Any] = [] + participants: list[bytes] = [] + cache = ctypes.create_string_buffer(KEYAGG_CACHE_SIZE) + aggregate_xonly = ctypes.create_string_buffer(64) + aggregate_pubkey = ctypes.create_string_buffer(64) + try: + for index, value in enumerate(values): + participant, parsed_pubkey = _parse_pubkey(secp256k1, value, f"pubkeys[{index}]") + participants.append(participant) + parsed.append(parsed_pubkey) + + if ( + secp256k1.lib.secp256k1_musig_pubkey_agg( + secp256k1.ctx.verify, + aggregate_xonly, + cache, + _opaque_ptr_array(parsed), + len(parsed), + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 MuSig2 key aggregation failed") + if ( + secp256k1.lib.secp256k1_musig_pubkey_get(secp256k1.ctx.verify, aggregate_pubkey, cache) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not retrieve the aggregate public key") + + serialized_xonly = ctypes.create_string_buffer(32) + try: + if ( + secp256k1.lib.secp256k1_xonly_pubkey_serialize( + secp256k1.ctx.verify, serialized_xonly, aggregate_xonly + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not serialize the aggregate x-only key") + return KeyAggContext( + cache, + tuple(participants), + serialized_xonly.raw, + _serialize_pubkey(secp256k1, aggregate_pubkey), + ) + finally: + _zero_buffer(serialized_xonly, 32) + except Exception: + _zero_buffer(cache, KEYAGG_CACHE_SIZE) + raise + finally: + _zero_buffer(aggregate_xonly, 64) + _zero_buffer(aggregate_pubkey, 64) + for parsed_pubkey in parsed: + _zero_buffer(parsed_pubkey, 64) + + +def apply_xonly_tweak(ctx: KeyAggContext, tweak: bytes) -> KeyAggContext: + """Return a new aggregation context with a BIP341-compatible x-only tweak.""" + + secp256k1 = _require_musig() + if not isinstance(ctx, KeyAggContext): + raise MuSig2Error("ctx must be a KeyAggContext") + tweak_value = _require_bytes(tweak, 32, "tweak") + cache = ctx._copy_cache() + aggregate_pubkey = ctypes.create_string_buffer(64) + try: + if ( + secp256k1.lib.secp256k1_musig_pubkey_xonly_tweak_add( + secp256k1.ctx.verify, aggregate_pubkey, cache, tweak_value + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 MuSig2 x-only tweak failed") + return KeyAggContext( + cache, + ctx._participants, + _serialize_xonly_pubkey(secp256k1, aggregate_pubkey), + _serialize_pubkey(secp256k1, aggregate_pubkey), + ) + except Exception: + _zero_buffer(cache, KEYAGG_CACHE_SIZE) + raise + finally: + _zero_buffer(aggregate_pubkey, 64) + + +def nonce_gen( + pubkey: bytes, + rand: bytes | bytearray | None = None, + *, + privkey: bytes | None = None, + msg32: bytes | None = None, + extra_input32: bytes | None = None, +) -> Tuple[SecNonce, bytes]: + """Generate a single-use secret nonce and its 66-byte public nonce. + + Omit rand to use fresh operating-system randomness. If supplied, rand must + be 32 uniformly random SECRET bytes, unique to this call even if signing + fails or is abandoned. It is not public auxiliary randomness. Reusing it + can repeat the nonce and expose the private key. + Once input validation succeeds, a supplied bytearray is wiped in place, + even if generation fails. Do not share or mutate it during this call. + Immutable rand bytes cannot be wiped by this function. There is no global + tracking of previously supplied randomness. + + Pass privkey, msg32, and extra_input32 when known to include them in nonce + derivation. They must be 32-byte bytes values; privkey must be a valid + scalar corresponding to pubkey. These inputs do not replace fresh rand. + """ + + secp256k1 = _require_musig() + participant, parsed_pubkey = _parse_pubkey(secp256k1, pubkey, "pubkey") + seed = None + signer_pubkey = secnonce = pubnonce = None + success = False + try: + seckey = None if privkey is None else _validate_scalar(secp256k1, privkey, "privkey") + message = None if msg32 is None else _require_bytes(msg32, 32, "msg32") + extra = None if extra_input32 is None else _require_bytes(extra_input32, 32, "extra_input32") + if rand is not None: + if not isinstance(rand, (bytes, bytearray)): + raise MuSig2Error("rand must be bytes or bytearray") + if len(rand) != 32: + raise MuSig2Error("rand must be exactly 32 bytes long") + if seckey is not None: + signer_pubkey = ctypes.create_string_buffer(64) + if ( + secp256k1.lib.secp256k1_ec_pubkey_create(secp256k1.ctx.sign, signer_pubkey, seckey) + != 1 + ): + raise MuSig2Error("libsecp256k1 could not derive the signer public key") + if _serialize_pubkey(secp256k1, signer_pubkey) != participant: + raise MuSig2Error("privkey does not correspond to pubkey") + + if isinstance(rand, bytearray): + seed = rand + else: + seed = bytearray(os.urandom(32) if rand is None else rand) + session_secrand = (ctypes.c_char * 32).from_buffer(seed) + secnonce = ctypes.create_string_buffer(SECNONCE_SIZE) + pubnonce = ctypes.create_string_buffer(PUBNONCE_SIZE) + if ( + secp256k1.lib.secp256k1_musig_nonce_gen( + secp256k1.ctx.sign, + secnonce, + pubnonce, + session_secrand, + seckey, + parsed_pubkey, + message, + None, + extra, + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 MuSig2 nonce generation failed") + serialized = _serialize_pubnonce(secp256k1, pubnonce) + result = SecNonce(secnonce, participant) + success = True + return result, serialized + finally: + if seed is not None: + seed[:] = b"\x00" * 32 + _zero_buffer(parsed_pubkey, 64) + if signer_pubkey is not None: + _zero_buffer(signer_pubkey, 64) + if pubnonce is not None: + _zero_buffer(pubnonce, PUBNONCE_SIZE) + if not success and secnonce is not None: + _zero_buffer(secnonce, SECNONCE_SIZE) + + +def nonce_agg(pubnonces: Sequence[bytes]) -> bytes: + """Aggregate 66-byte MuSig2 public nonces.""" + + secp256k1 = _require_musig() + values = _require_bytes_sequence(pubnonces, "pubnonces") + parsed: list[Any] = [] + aggregate = ctypes.create_string_buffer(AGGNONCE_SIZE) + try: + for index, value in enumerate(values): + parsed.append(_parse_pubnonce(secp256k1, value, f"pubnonces[{index}]")) + if ( + secp256k1.lib.secp256k1_musig_nonce_agg( + secp256k1.ctx.verify, aggregate, _opaque_ptr_array(parsed), len(parsed) + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 MuSig2 nonce aggregation failed") + return _serialize_aggnonce(secp256k1, aggregate) + finally: + _zero_buffer(aggregate, AGGNONCE_SIZE) + for parsed_nonce in parsed: + _zero_buffer(parsed_nonce, PUBNONCE_SIZE) + + +def get_session( + aggnonce: bytes, + pubkeys: Sequence[bytes], + tweaks: Sequence[bytes], + msg32: bytes, +) -> Session: + """Create a signing session from its aggregate nonce and ordered key setup.""" + + secp256k1 = _require_musig() + message = _require_bytes(msg32, 32, "msg32") + ctx = key_agg(pubkeys) + if not isinstance(tweaks, Sequence) or isinstance(tweaks, (bytes, bytearray, str)): + raise MuSig2Error("tweaks must be a sequence of bytes") + for index, tweak in enumerate(tweaks): + ctx = apply_xonly_tweak(ctx, _require_bytes(tweak, 32, f"tweaks[{index}]")) + + parsed_aggnonce = _parse_aggnonce(secp256k1, aggnonce, "aggnonce") + native_session = ctypes.create_string_buffer(SESSION_SIZE) + try: + if ( + secp256k1.lib.secp256k1_musig_nonce_process( + secp256k1.ctx.sign, native_session, parsed_aggnonce, message, ctx._cache + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 MuSig2 session creation failed") + return Session(ctx, native_session, ctx._participants) + except Exception: + _zero_buffer(native_session, SESSION_SIZE) + raise + finally: + _zero_buffer(parsed_aggnonce, AGGNONCE_SIZE) + + +def sign_partial(secnonce: SecNonce, privkey: bytes, session: Session) -> bytes: + """Create a partial signature and irrevocably consume ``secnonce``. + + Verify the result with partial_sig_verify before sharing it. The native + signing function does not perform this recommended fault-detection check. + """ + + if not isinstance(secnonce, SecNonce): + raise MuSig2Error("secnonce must be a SecNonce") + return secnonce._sign(privkey, session) + + +def partial_sig_verify(psig: bytes, pubnonce: bytes, pubkey: bytes, session: Session) -> bool: + """Verify a participant's 32-byte partial signature for ``session``. + + Malformed contributions or an invalid session argument raise MuSig2Error. + Well-formed contributions return False if verification fails or the public + key is not a session participant. Coordinators should handle MuSig2Error + for each peer's contribution separately. + """ + + secp256k1 = _require_musig() + if not isinstance(session, Session): + raise MuSig2Error("session must be a MuSig2 Session") + participant, parsed_pubkey = _parse_pubkey(secp256k1, pubkey, "pubkey") + parsed_pubnonce = _parse_pubnonce(secp256k1, pubnonce, "pubnonce") + parsed_psig = _parse_partial_sig(secp256k1, psig, "psig") + try: + if participant not in session._participants: + return False + return bool( + secp256k1.lib.secp256k1_musig_partial_sig_verify( + secp256k1.ctx.verify, + parsed_psig, + parsed_pubnonce, + parsed_pubkey, + session.ctx._cache, + session._session, + ) + == 1 + ) + finally: + _zero_buffer(parsed_psig, PARTIAL_SIG_SIZE) + _zero_buffer(parsed_pubnonce, PUBNONCE_SIZE) + _zero_buffer(parsed_pubkey, 64) + + +def partial_sig_agg(psigs: Sequence[bytes], session: Session) -> bytes: + """Aggregate one partial signature per session participant into a BIP340 signature. + + This does not verify the contributions or the result. Verify the returned + signature against the session's aggregate x-only key and message. + """ + + secp256k1 = _require_musig() + if not isinstance(session, Session): + raise MuSig2Error("session must be a MuSig2 Session") + values = _require_bytes_sequence(psigs, "psigs") + if len(values) != len(session._participants): + raise MuSig2Error("psigs must contain exactly one signature per session participant") + + parsed: list[Any] = [] + signature = ctypes.create_string_buffer(64) + try: + for index, value in enumerate(values): + parsed.append(_parse_partial_sig(secp256k1, value, f"psigs[{index}]")) + if ( + secp256k1.lib.secp256k1_musig_partial_sig_agg( + secp256k1.ctx.verify, + signature, + session._session, + _opaque_ptr_array(parsed), + len(parsed), + ) + != 1 + ): + raise MuSig2Error("libsecp256k1 MuSig2 partial signature aggregation failed") + return signature.raw + finally: + _zero_buffer(signature, 64) + for parsed_sig in parsed: + _zero_buffer(parsed_sig, PARTIAL_SIG_SIZE) + + +__all__ = ( + "MuSig2Error", + "KeyAggContext", + "SecNonce", + "Session", + "key_agg", + "apply_xonly_tweak", + "nonce_gen", + "nonce_agg", + "get_session", + "sign_partial", + "partial_sig_verify", + "partial_sig_agg", +) diff --git a/bitcointx/core/secp256k1.py b/bitcointx/core/secp256k1.py index 097a40e0..9912bde9 100644 --- a/bitcointx/core/secp256k1.py +++ b/bitcointx/core/secp256k1.py @@ -101,6 +101,7 @@ class Secp256k1_Capabilities: has_ecdh: bool has_xonly_pubkeys: bool has_schnorrsig: bool + has_musig: bool @dataclass(frozen=True) @@ -126,6 +127,126 @@ def _get_schnorrsig_sign_function(lib: ctypes.CDLL) -> Optional[Any]: return sign_function +_MUSIG_FUNCTION_NAMES = ( + "secp256k1_musig_pubnonce_parse", + "secp256k1_musig_pubnonce_serialize", + "secp256k1_musig_aggnonce_parse", + "secp256k1_musig_aggnonce_serialize", + "secp256k1_musig_partial_sig_parse", + "secp256k1_musig_partial_sig_serialize", + "secp256k1_musig_pubkey_agg", + "secp256k1_musig_pubkey_get", + "secp256k1_musig_pubkey_xonly_tweak_add", + "secp256k1_musig_nonce_gen", + "secp256k1_musig_nonce_agg", + "secp256k1_musig_nonce_process", + "secp256k1_musig_partial_sign", + "secp256k1_musig_partial_sig_verify", + "secp256k1_musig_partial_sig_agg", +) + + +def _add_musig_function_definitions(lib: ctypes.CDLL) -> bool: + """Configure the v0.6.0+ MuSig2 ABI only when its full symbol group exists.""" + + if not all(getattr(lib, name, None) is not None for name in _MUSIG_FUNCTION_NAMES): + return False + + opaque_ptr = ctypes.c_void_p + opaque_ptr_array = ctypes.POINTER(opaque_ptr) + + for name in ( + "secp256k1_musig_pubnonce_parse", + "secp256k1_musig_aggnonce_parse", + "secp256k1_musig_partial_sig_parse", + ): + function = getattr(lib, name) + function.restype = ctypes.c_int + function.argtypes = [opaque_ptr, opaque_ptr, ctypes.c_char_p] + + for name in ( + "secp256k1_musig_pubnonce_serialize", + "secp256k1_musig_aggnonce_serialize", + "secp256k1_musig_partial_sig_serialize", + ): + function = getattr(lib, name) + function.restype = ctypes.c_int + function.argtypes = [opaque_ptr, ctypes.c_char_p, opaque_ptr] + + lib.secp256k1_musig_pubkey_agg.restype = ctypes.c_int + lib.secp256k1_musig_pubkey_agg.argtypes = [ + opaque_ptr, + opaque_ptr, + opaque_ptr, + opaque_ptr_array, + ctypes.c_size_t, + ] + lib.secp256k1_musig_pubkey_get.restype = ctypes.c_int + lib.secp256k1_musig_pubkey_get.argtypes = [opaque_ptr, opaque_ptr, opaque_ptr] + lib.secp256k1_musig_pubkey_xonly_tweak_add.restype = ctypes.c_int + lib.secp256k1_musig_pubkey_xonly_tweak_add.argtypes = [ + opaque_ptr, + opaque_ptr, + opaque_ptr, + ctypes.c_char_p, + ] + lib.secp256k1_musig_nonce_gen.restype = ctypes.c_int + lib.secp256k1_musig_nonce_gen.argtypes = [ + opaque_ptr, + opaque_ptr, + opaque_ptr, + ctypes.c_char_p, + ctypes.c_char_p, + opaque_ptr, + ctypes.c_char_p, + opaque_ptr, + ctypes.c_char_p, + ] + lib.secp256k1_musig_nonce_agg.restype = ctypes.c_int + lib.secp256k1_musig_nonce_agg.argtypes = [ + opaque_ptr, + opaque_ptr, + opaque_ptr_array, + ctypes.c_size_t, + ] + lib.secp256k1_musig_nonce_process.restype = ctypes.c_int + lib.secp256k1_musig_nonce_process.argtypes = [ + opaque_ptr, + opaque_ptr, + opaque_ptr, + ctypes.c_char_p, + opaque_ptr, + ] + lib.secp256k1_musig_partial_sign.restype = ctypes.c_int + lib.secp256k1_musig_partial_sign.argtypes = [ + opaque_ptr, + opaque_ptr, + opaque_ptr, + opaque_ptr, + opaque_ptr, + opaque_ptr, + ] + lib.secp256k1_musig_partial_sig_verify.restype = ctypes.c_int + lib.secp256k1_musig_partial_sig_verify.argtypes = [ + opaque_ptr, + opaque_ptr, + opaque_ptr, + opaque_ptr, + opaque_ptr, + opaque_ptr, + ] + lib.secp256k1_musig_partial_sig_agg.restype = ctypes.c_int + lib.secp256k1_musig_partial_sig_agg.argtypes = [ + opaque_ptr, + ctypes.c_char_p, + opaque_ptr, + opaque_ptr_array, + ctypes.c_size_t, + ] + + return True + + def get_secp256k1() -> Secp256k1: """Will create and initialize an instance of Secp256k1 class, and store it as attribute of the module this function resides in. If this attribute @@ -148,6 +269,7 @@ def _add_function_definitions(lib: ctypes.CDLL) -> Secp256k1_Capabilities: has_ecdh = False has_xonly_pubkeys = False has_schnorrsig = False + has_musig = False if getattr(lib, 'secp256k1_ecdsa_sign_recoverable', None): has_pubkey_recovery = True @@ -293,13 +415,17 @@ def _add_function_definitions(lib: ctypes.CDLL) -> Secp256k1_Capabilities: schnorrsig_sign.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p] lib.secp256k1_schnorrsig_sign = schnorrsig_sign # type: ignore[attr-defined] + if has_xonly_pubkeys: + has_musig = _add_musig_function_definitions(lib) + return Secp256k1_Capabilities( has_pubkey_recovery=has_pubkey_recovery, has_privkey_negate=has_privkey_negate, has_pubkey_negate=has_pubkey_negate, has_ecdh=has_ecdh, has_xonly_pubkeys=has_xonly_pubkeys, - has_schnorrsig=has_schnorrsig) + has_schnorrsig=has_schnorrsig, + has_musig=has_musig) def secp256k1_create_and_init_context(lib: ctypes.CDLL, flags: int diff --git a/bitcointx/tests/data/bip327-musig2-vectors.json b/bitcointx/tests/data/bip327-musig2-vectors.json new file mode 100644 index 00000000..de192376 --- /dev/null +++ b/bitcointx/tests/data/bip327-musig2-vectors.json @@ -0,0 +1,90 @@ +{ + "sources": { + "key_agg": "https://github.com/bitcoin/bips/blob/master/bip-0327/vectors/key_agg_vectors.json", + "nonce_gen": "https://github.com/bitcoin/bips/blob/master/bip-0327/vectors/nonce_gen_vectors.json", + "nonce_agg": "https://github.com/bitcoin/bips/blob/master/bip-0327/vectors/nonce_agg_vectors.json", + "nonce_coefficient_partial_verify": "https://github.com/bitcoin/bips/blob/master/bip-0327/vectors/sign_verify_vectors.json", + "xonly_tweak_partial_verify": "https://github.com/bitcoin/bips/blob/master/bip-0327/vectors/tweak_vectors.json", + "signature_agg": "https://github.com/bitcoin/bips/blob/master/bip-0327/vectors/sig_agg_vectors.json", + "zero_xonly_tweak": "BIP327 KeyAgg case [2, 1, 0] with BIP340 x-only zero tweak" + }, + "key_agg": { + "pubkeys": [ + "02F9308A019258C31049344F85F89D5229B531C845836F99B08601F113BCE036F9", + "03DFF1D77F2A671C5F36183726DB2341BE58FEAE1DA2DECED843240F7B502BA659", + "023590A94E768F8E1815C2F24B4D80A8E3149316C3518CE7B7AD338368D038CA66" + ], + "test_cases": [ + {"key_indices": [0, 1, 2], "expected": "90539EEDE565F5D054F32CC0C220126889ED1E5D193BAF15AEF344FE59D4610C"}, + {"key_indices": [2, 1, 0], "expected": "6204DE8B083426DC6EAF9502D27024D53FC826BF7D2012148A0575435DF54B2B"}, + {"key_indices": [0, 0, 0], "expected": "B436E3BAD62B8CD409969A224731C193D051162D8C5AE8B109306127DA3AA935"}, + {"key_indices": [0, 0, 1, 1], "expected": "69BC22BFA5D106306E48A20679DE1D7389386124D07571D0D872686028C26A3E"} + ] + }, + "nonce_gen": { + "rand": "0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F0F", + "pubkey": "02F9308A019258C31049344F85F89D5229B531C845836F99B08601F113BCE036F9", + "expected_pubnonce": "02C96E7CB1E8AA5DAC64D872947914198F607D90ECDE5200DE52978AD5DED63C000299EC5117C2D29EDEE8A2092587C3909BE694D5CFF0667D6C02EA4059F7CD9786" + }, + "nonce_agg": { + "pubnonces": [ + "020151C80F435648DF67A22B749CD798CE54E0321D034B92B709B567D60A42E66603BA47FBC1834437B3212E89A84D8425E7BF12E0245D98262268EBDCB385D50641", + "03FF406FFD8ADB9CD29877E4985014F66A59F6CD01C0E88CAA8E5F3166B1F676A60248C264CDD57D3C24D79990B0F865674EB62A0F9018277A95011B41BFC193B833" + ], + "expected": "035FE1873B4F2967F52FEA4A06AD5A8ECCBE9D0FD73068012C894E2E87CCB5804B024725377345BDE0E9C33AF3C43C0A29A9249F2F2956FA8CFEB55C8573D0262DC8" + }, + "nonce_coefficient_partial_verify": { + "pubkeys": [ + "03935F972DA013F80AE011890FA89B67A27B7BE6CCB24D3274D18B2D4067F261A9", + "02F9308A019258C31049344F85F89D5229B531C845836F99B08601F113BCE036F9", + "02DFF1D77F2A671C5F36183726DB2341BE58FEAE1DA2DECED843240F7B502BA661" + ], + "pubnonces": [ + "0337C87821AFD50A8644D820A8F3E02E499C931865C2360FB43D0A0D20DAFE07EA0287BF891D2A6DEAEBADC909352AA9405D1428C15F4B75F04DAE642A95C2548480", + "0279BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F817980279BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798", + "032DE2662628C90B03F5E720284EB52FF7D71F4284F627B68A853D78C78E1FFE9303E4C5524E83FFE1493B9077CF1CA6BEB2090C93D930321071AD40B2F44E599046" + ], + "aggnonce": "028465FCF0BBDBCF443AABCCE533D42B4B5A10966AC09A49655E8C42DAAB8FCD61037496A3CC86926D452CAFCFD55D25972CA1675D549310DE296BFF42F72EEEA8C9", + "msg": "F95466D086770E689964664219266FE5ED215C92AE20BAB5C9D79ADDDDF3C0CF", + "pubkey_index": 0, + "pubnonce_index": 0, + "psig": "012ABBCB52B3016AC03AD82395A1A415C48B93DEF78718E62A7A90052FE224FB" + }, + "xonly_tweak_partial_verify": { + "pubkeys": [ + "02F9308A019258C31049344F85F89D5229B531C845836F99B08601F113BCE036F9", + "02DFF1D77F2A671C5F36183726DB2341BE58FEAE1DA2DECED843240F7B502BA659", + "03935F972DA013F80AE011890FA89B67A27B7BE6CCB24D3274D18B2D4067F261A9" + ], + "pubnonces": [ + "0279BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F817980279BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798", + "032DE2662628C90B03F5E720284EB52FF7D71F4284F627B68A853D78C78E1FFE9303E4C5524E83FFE1493B9077CF1CA6BEB2090C93D930321071AD40B2F44E599046", + "0337C87821AFD50A8644D820A8F3E02E499C931865C2360FB43D0A0D20DAFE07EA0287BF891D2A6DEAEBADC909352AA9405D1428C15F4B75F04DAE642A95C2548480" + ], + "aggnonce": "028465FCF0BBDBCF443AABCCE533D42B4B5A10966AC09A49655E8C42DAAB8FCD61037496A3CC86926D452CAFCFD55D25972CA1675D549310DE296BFF42F72EEEA8C9", + "tweak": "E8F791FF9225A2AF0102AFFF4A9A723D9612A682A25EBE79802B263CDFCD83BB", + "msg": "F95466D086770E689964664219266FE5ED215C92AE20BAB5C9D79ADDDDF3C0CF", + "pubkey_index": 2, + "pubnonce_index": 2, + "psig": "E28A5C66E61E178C2BA19DB77B6CF9F7E2F0F56C17918CD13135E60CC848FE91" + }, + "zero_xonly_tweak": { + "key_indices": [2, 1, 0], + "tweak": "0000000000000000000000000000000000000000000000000000000000000000", + "expected_xonly": "6204DE8B083426DC6EAF9502D27024D53FC826BF7D2012148A0575435DF54B2B", + "expected_pubkey": "026204DE8B083426DC6EAF9502D27024D53FC826BF7D2012148A0575435DF54B2B" + }, + "signature_agg": { + "pubkeys": [ + "03935F972DA013F80AE011890FA89B67A27B7BE6CCB24D3274D18B2D4067F261A9", + "02D2DC6F5DF7C56ACF38C7FA0AE7A759AE30E19B37359DFDE015872324C7EF6E05" + ], + "aggnonce": "0341432722C5CD0268D829C702CF0D1CBCE57033EED201FD335191385227C3210C03D377F2D258B64AADC0E16F26462323D701D286046A2EA93365656AFD9875982B", + "msg": "599C67EA410D005B9DA90817CF03ED3B1C868E4DA4EDF00A5880B0082C237869", + "psigs": [ + "B15D2CD3C3D22B04DAE438CE653F6B4ECF042F42CFDED7C41B64AAF9B4AF53FB", + "6193D6AC61B354E9105BBDC8937A3454A6D705B6D57322A5A472A02CE99FCB64" + ], + "expected": "041DA22223CE65C92C9A0D6C2CAC828AAF1EEE56304FEC371DDF91EBB2B9EF0912F1038025857FEDEB3FF696F8B99FA4BB2C5812F6095A2E0004EC99CE18DE1E" + } +} diff --git a/bitcointx/tests/test_musig.py b/bitcointx/tests/test_musig.py new file mode 100644 index 00000000..9f4dff12 --- /dev/null +++ b/bitcointx/tests/test_musig.py @@ -0,0 +1,510 @@ +# Copyright (C) 2026 The python-bitcointx developers +# +# This file is part of python-bitcointx. +# +# It is subject to the license terms in the LICENSE file found in the top-level +# directory of this distribution. + +import copy +import ctypes +import json +import pickle +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace +from typing import Any, ClassVar, cast +from unittest.mock import patch + +from bitcointx.core.key import CKey, XOnlyPubKey +from bitcointx.core.musig import ( + MuSig2Error, + SecNonce, + Session, + apply_xonly_tweak, + get_session, + key_agg, + nonce_agg, + nonce_gen, + partial_sig_agg, + partial_sig_verify, + sign_partial, +) +from bitcointx.core.secp256k1 import ( + _MUSIG_FUNCTION_NAMES, + _add_musig_function_definitions, + get_secp256k1, +) + + +def _hex(value: str) -> bytes: + return bytes.fromhex(value) + + +HAS_MUSIG = get_secp256k1().cap.has_musig + + +class TestMuSig2Capability(unittest.TestCase): + def test_capability_requires_every_musig_symbol(self) -> None: + symbols = {name: SimpleNamespace() for name in _MUSIG_FUNCTION_NAMES[:-1]} + partial_library = cast(ctypes.CDLL, SimpleNamespace(**symbols)) + self.assertFalse(_add_musig_function_definitions(partial_library)) + + def test_missing_capability_fails_clearly(self) -> None: + unavailable = SimpleNamespace(cap=SimpleNamespace(has_musig=False)) + with patch("bitcointx.core.musig.get_secp256k1", return_value=unavailable): + with self.assertRaisesRegex(MuSig2Error, "--enable-module-musig"): + key_agg([b"\x02" + b"\x01" * 32]) + + +@unittest.skipUnless(HAS_MUSIG, "libsecp256k1 was not built with the musig module") +class TestMuSig2(unittest.TestCase): + vectors: ClassVar[dict[str, Any]] + + @classmethod + def setUpClass(cls) -> None: + vectors_path = Path(__file__).with_name("data") / "bip327-musig2-vectors.json" + cls.vectors = json.loads(vectors_path.read_text(encoding="ascii")) + + def _two_party_session( + self, + ) -> tuple[ + list[CKey], + list[bytes], + bytes, + tuple[Session, SecNonce, SecNonce], + bytes, + bytes, + ]: + keys = [ + CKey(b"\x00" * 31 + b"\x01"), + CKey(b"\x00" * 31 + b"\x02"), + ] + pubkeys = [bytes(key.pub) for key in keys] + secnonce_one, pubnonce_one = nonce_gen(pubkeys[0]) + secnonce_two, pubnonce_two = nonce_gen(pubkeys[1]) + message = b"\x33" * 32 + session = get_session(nonce_agg([pubnonce_one, pubnonce_two]), pubkeys, [], message) + return ( + keys, + pubkeys, + message, + (session, secnonce_one, secnonce_two), + pubnonce_one, + pubnonce_two, + ) + + def test_official_key_aggregation_vectors(self) -> None: + vectors = self.vectors["key_agg"] + pubkeys = [_hex(value) for value in vectors["pubkeys"]] + for case in vectors["test_cases"]: + ctx = key_agg([pubkeys[index] for index in case["key_indices"]]) + self.assertEqual(ctx.aggregate_xonly(), _hex(case["expected"])) + self.assertEqual(len(ctx.aggregate_pubkey()), 33) + + def test_official_nonce_vectors(self) -> None: + nonce_gen_vector = self.vectors["nonce_gen"] + _, pubnonce = nonce_gen(_hex(nonce_gen_vector["pubkey"]), _hex(nonce_gen_vector["rand"])) + self.assertEqual(pubnonce, _hex(nonce_gen_vector["expected_pubnonce"])) + + nonce_agg_vector = self.vectors["nonce_agg"] + self.assertEqual( + nonce_agg([_hex(value) for value in nonce_agg_vector["pubnonces"]]), + _hex(nonce_agg_vector["expected"]), + ) + + def test_official_nonce_vector_supports_mutable_seed(self) -> None: + nonce_gen_vector = self.vectors["nonce_gen"] + pubkey = _hex(nonce_gen_vector["pubkey"]) + seed = bytearray(_hex(nonce_gen_vector["rand"])) + + _, pubnonce = nonce_gen(pubkey, seed) + + self.assertEqual(pubnonce, _hex(nonce_gen_vector["expected_pubnonce"])) + self.assertEqual(seed, bytearray(32)) + with self.assertRaises(MuSig2Error): + nonce_gen(pubkey, seed) + + def test_nonce_gen_preserves_immutable_seed(self) -> None: + vector = self.vectors["nonce_gen"] + seed = _hex(vector["rand"]) + _, pubnonce = nonce_gen(_hex(vector["pubkey"]), seed) + self.assertEqual(seed.hex().upper(), vector["rand"]) + self.assertEqual(pubnonce, _hex(vector["expected_pubnonce"])) + + def test_nonce_gen_validates_optional_arguments_before_native_generation(self) -> None: + key = CKey(b"\x00" * 31 + b"\x01") + other_key = CKey(b"\x00" * 31 + b"\x02") + pubkey = bytes(key.pub) + invalid_inputs: list[tuple[str, Any]] = [ + ("rand", memoryview(bytes(32))), + ("rand", bytearray(31)), + ("rand", bytearray(33)), + ("rand", bytes(31)), + ("rand", bytes(33)), + ("pubkey", b"\x04" + pubkey[1:]), + ("privkey", bytearray(bytes(key))), + ("privkey", b"\x01" * 31), + ("privkey", b"\x01" * 33), + ("privkey", bytes(32)), + ("privkey", b"\xff" * 32), + ("privkey", bytes(other_key)), + ("msg32", bytearray(32)), + ("msg32", bytes(31)), + ("msg32", bytes(33)), + ("extra_input32", bytearray(32)), + ("extra_input32", bytes(31)), + ("extra_input32", bytes(33)), + ] + secp256k1 = get_secp256k1() + native_nonce_gen = secp256k1.lib.secp256k1_musig_nonce_gen + + with patch.object( + secp256k1.lib, "secp256k1_musig_nonce_gen", wraps=native_nonce_gen + ) as mocked_nonce_gen: + for name, value in invalid_inputs: + with self.subTest(argument=name, value=value): + kwargs: dict[str, Any] = { + "pubkey": pubkey, "rand": bytearray(b"\xa1" * 32), name: value + } + seed = kwargs["rand"] + seed_before = bytes(seed) + with self.assertRaises(MuSig2Error): + nonce_gen(**kwargs) + self.assertEqual(seed, seed_before) + + mocked_nonce_gen.assert_not_called() + + def test_nonce_gen_forwards_optional_inputs_to_native(self) -> None: + key = CKey(b"\x00" * 31 + b"\x01") + pubkey = bytes(key.pub) + privkey = bytes(key) + msg32 = b"\x31" * 32 + extra_input32 = b"\x32" * 32 + cases: tuple[ + tuple[str, dict[str, bytes], bytes | None, bytes | None, bytes | None], ... + ] = ( + ("defaults", {}, None, None, None), + ("privkey", {"privkey": privkey}, privkey, None, None), + ("msg32", {"msg32": msg32}, None, msg32, None), + ("extra input", {"extra_input32": extra_input32}, None, None, extra_input32), + ( + "all optional inputs", + { + "privkey": privkey, + "msg32": msg32, + "extra_input32": extra_input32, + }, + privkey, + msg32, + extra_input32, + ), + ) + secp256k1 = get_secp256k1() + native_nonce_gen = secp256k1.lib.secp256k1_musig_nonce_gen + + with patch.object( + secp256k1.lib, "secp256k1_musig_nonce_gen", wraps=native_nonce_gen + ) as mocked_nonce_gen: + for name, kwargs, expected_privkey, expected_msg32, expected_extra_input32 in cases: + with self.subTest(optional_inputs=name): + nonce_gen(pubkey, **kwargs) + + mocked_nonce_gen.assert_called_once() + args = mocked_nonce_gen.call_args.args + self.assertEqual(args[4], expected_privkey) + self.assertEqual(args[6], expected_msg32) + self.assertIsNone(args[7]) + self.assertEqual(args[8], expected_extra_input32) + mocked_nonce_gen.reset_mock() + + def test_optional_nonce_inputs_support_two_party_signing(self) -> None: + keys = [ + CKey(b"\x00" * 31 + b"\x01"), + CKey(b"\x00" * 31 + b"\x02"), + ] + pubkeys = [bytes(key.pub) for key in keys] + message = b"\x41" * 32 + secnonce_one, pubnonce_one = nonce_gen( + pubkeys[0], + privkey=bytes(keys[0]), + msg32=message, + extra_input32=b"\x51" * 32, + ) + secnonce_two, pubnonce_two = nonce_gen( + pubkeys[1], + privkey=bytes(keys[1]), + msg32=message, + extra_input32=b"\x52" * 32, + ) + session = get_session(nonce_agg([pubnonce_one, pubnonce_two]), pubkeys, [], message) + psig_one = sign_partial(secnonce_one, bytes(keys[0]), session) + psig_two = sign_partial(secnonce_two, bytes(keys[1]), session) + + self.assertTrue(partial_sig_verify(psig_one, pubnonce_one, pubkeys[0], session)) + self.assertTrue(partial_sig_verify(psig_two, pubnonce_two, pubkeys[1], session)) + signature = partial_sig_agg([psig_one, psig_two], session) + self.assertTrue( + XOnlyPubKey(session.ctx.aggregate_xonly()).verify_schnorr(message, signature) + ) + + def test_official_nonce_coefficient_partial_signature_vector(self) -> None: + vector = self.vectors["nonce_coefficient_partial_verify"] + pubkeys = [_hex(value) for value in vector["pubkeys"]] + pubnonces = [_hex(value) for value in vector["pubnonces"]] + session = get_session(_hex(vector["aggnonce"]), pubkeys, [], _hex(vector["msg"])) + self.assertTrue( + partial_sig_verify( + _hex(vector["psig"]), + pubnonces[vector["pubnonce_index"]], + pubkeys[vector["pubkey_index"]], + session, + ) + ) + + tweak_vector = self.vectors["xonly_tweak_partial_verify"] + tweak_pubkeys = [_hex(value) for value in tweak_vector["pubkeys"]] + tweak_pubnonces = [_hex(value) for value in tweak_vector["pubnonces"]] + tweak_session = get_session( + _hex(tweak_vector["aggnonce"]), + tweak_pubkeys, + [_hex(tweak_vector["tweak"])], + _hex(tweak_vector["msg"]), + ) + self.assertTrue( + partial_sig_verify( + _hex(tweak_vector["psig"]), + tweak_pubnonces[tweak_vector["pubnonce_index"]], + tweak_pubkeys[tweak_vector["pubkey_index"]], + tweak_session, + ) + ) + + def test_official_partial_signature_aggregation_vector(self) -> None: + vector = self.vectors["signature_agg"] + session = get_session( + _hex(vector["aggnonce"]), + [_hex(value) for value in vector["pubkeys"]], + [], + _hex(vector["msg"]), + ) + self.assertEqual( + partial_sig_agg([_hex(value) for value in vector["psigs"]], session), + _hex(vector["expected"]), + ) + self.assertTrue( + XOnlyPubKey(session.ctx.aggregate_xonly()).verify_schnorr( + _hex(vector["msg"]), _hex(vector["expected"]) + ) + ) + + def test_zero_xonly_tweak_matches_independent_expected_key(self) -> None: + key_agg_vector = self.vectors["key_agg"] + zero_tweak_vector = self.vectors["zero_xonly_tweak"] + pubkeys = [_hex(value) for value in key_agg_vector["pubkeys"]] + ctx = key_agg([pubkeys[index] for index in zero_tweak_vector["key_indices"]]) + self.assertEqual(ctx.aggregate_pubkey()[0], 3) + + tweaked_ctx = apply_xonly_tweak(ctx, _hex(zero_tweak_vector["tweak"])) + self.assertEqual(tweaked_ctx.aggregate_xonly(), _hex(zero_tweak_vector["expected_xonly"])) + self.assertEqual(tweaked_ctx.aggregate_pubkey(), _hex(zero_tweak_vector["expected_pubkey"])) + + def test_two_party_tweaked_signature_verifies(self) -> None: + keys = [ + CKey(b"\x00" * 31 + b"\x01"), + CKey(b"\x00" * 31 + b"\x02"), + ] + pubkeys = [bytes(key.pub) for key in keys] + tweak = b"\x00" * 31 + b"\x03" + direct_ctx = apply_xonly_tweak(key_agg(pubkeys), tweak) + secnonce_one, pubnonce_one = nonce_gen(pubkeys[0]) + secnonce_two, pubnonce_two = nonce_gen(pubkeys[1]) + message = b"\x33" * 32 + session = get_session(nonce_agg([pubnonce_one, pubnonce_two]), pubkeys, [tweak], message) + psig_one = sign_partial(secnonce_one, bytes(keys[0]), session) + psig_two = sign_partial(secnonce_two, bytes(keys[1]), session) + + self.assertEqual(session.ctx.aggregate_xonly(), direct_ctx.aggregate_xonly()) + self.assertTrue(partial_sig_verify(psig_one, pubnonce_one, pubkeys[0], session)) + self.assertTrue(partial_sig_verify(psig_two, pubnonce_two, pubkeys[1], session)) + signature = partial_sig_agg([psig_one, psig_two], session) + self.assertTrue( + XOnlyPubKey(session.ctx.aggregate_xonly()).verify_schnorr(message, signature) + ) + + def test_zero_xonly_tweak_full_signing(self) -> None: + keys = [ + CKey(b"\x00" * 31 + b"\x01"), + CKey(b"\x00" * 31 + b"\x02"), + ] + pubkeys = [bytes(key.pub) for key in keys] + tweak = b"\x00" * 32 + direct_ctx = apply_xonly_tweak(key_agg(pubkeys), tweak) + secnonce_one, pubnonce_one = nonce_gen(pubkeys[0]) + secnonce_two, pubnonce_two = nonce_gen(pubkeys[1]) + message = b"\x99" * 32 + session = get_session(nonce_agg([pubnonce_one, pubnonce_two]), pubkeys, [tweak], message) + psig_one = sign_partial(secnonce_one, bytes(keys[0]), session) + psig_two = sign_partial(secnonce_two, bytes(keys[1]), session) + signature = partial_sig_agg([psig_one, psig_two], session) + + self.assertEqual(session.ctx.aggregate_xonly(), direct_ctx.aggregate_xonly()) + self.assertTrue( + XOnlyPubKey(session.ctx.aggregate_xonly()).verify_schnorr(message, signature) + ) + + def test_duplicate_participants_and_partial_count(self) -> None: + key = CKey(b"\x00" * 31 + b"\x01") + pubkey = bytes(key.pub) + secnonce_one, pubnonce_one = nonce_gen(pubkey) + secnonce_two, pubnonce_two = nonce_gen(pubkey) + message = b"\x66" * 32 + session = get_session( + nonce_agg([pubnonce_one, pubnonce_two]), [pubkey, pubkey], [], message + ) + psig_one = sign_partial(secnonce_one, bytes(key), session) + psig_two = sign_partial(secnonce_two, bytes(key), session) + signature = partial_sig_agg([psig_one, psig_two], session) + self.assertTrue( + XOnlyPubKey(session.ctx.aggregate_xonly()).verify_schnorr(message, signature) + ) + with self.assertRaises(MuSig2Error): + partial_sig_agg([psig_one], session) + + def test_invalid_encodings_are_rejected_before_session_use(self) -> None: + key = CKey(b"\x00" * 31 + b"\x01") + pubkey = bytes(key.pub) + with self.assertRaises(MuSig2Error): + key_agg([]) + with self.assertRaises(MuSig2Error): + key_agg([b"\x04" + pubkey[1:]]) + with self.assertRaises(MuSig2Error): + nonce_agg([b"\x00" * 65]) + with self.assertRaises(MuSig2Error): + get_session(b"\x00" * 66, [pubkey], [], b"\x00" * 31) + + _, _, _, session_data, pubnonce_one, _ = self._two_party_session() + session, _, _ = session_data + with self.assertRaises(MuSig2Error): + partial_sig_verify(b"\xff" * 32, pubnonce_one, pubkey, session) + + def test_well_formed_invalid_partial_signature_returns_false(self) -> None: + _, pubkeys, _, session_data, pubnonce, _ = self._two_party_session() + session, _, _ = session_data + self.assertFalse(partial_sig_verify(bytes(32), pubnonce, pubkeys[0], session)) + + def test_secret_nonce_is_single_use_and_nonserializable(self) -> None: + keys, pubkeys, _, session_data, _, _ = self._two_party_session() + session, secnonce_one, _ = session_data + self.assertEqual(repr(secnonce_one), "SecNonce()") + with self.assertRaises(TypeError): + copy.copy(secnonce_one) + with self.assertRaises(TypeError): + copy.deepcopy(secnonce_one) + with self.assertRaises(TypeError): + pickle.dumps(secnonce_one) + + with self.assertRaises(MuSig2Error): + sign_partial(secnonce_one, bytes(keys[1]), session) + with self.assertRaisesRegex(MuSig2Error, "already been consumed"): + sign_partial(secnonce_one, bytes(keys[0]), session) + + def test_secret_nonce_consumption_is_thread_safe(self) -> None: + keys, _, _, session_data, _, _ = self._two_party_session() + session, secnonce_one, _ = session_data + barrier = threading.Barrier(2) + + def attempt() -> bytes | None: + barrier.wait() + try: + return sign_partial(secnonce_one, bytes(keys[0]), session) + except MuSig2Error: + return None + + with ThreadPoolExecutor(max_workers=2) as executor: + outcomes = list(executor.map(lambda _: attempt(), range(2))) + self.assertEqual(sum(outcome is not None for outcome in outcomes), 1) + + def test_secret_nonce_is_wiped_when_signing_allocation_fails(self) -> None: + create_buffer = ctypes.create_string_buffer + for fail_at in (1, 2, 3): + with self.subTest(allocation=fail_at): + keys, _, _, session_data, _, _ = self._two_party_session() + session, secnonce, _ = session_data + nonce_buffer = cast(Any, secnonce)._SecNonce__buffer + self.assertTrue(any(nonce_buffer.raw)) + allocated: list[Any] = [] + + def allocate(size: int) -> Any: + if len(allocated) + 1 == fail_at: + raise MemoryError("injected allocation failure") + buffer = create_buffer(size) + # Nonzero contents distinguish cleanup from initial zero allocation. + ctypes.memset(buffer, 0x7F, size) + allocated.append(buffer) + return buffer + + with patch("bitcointx.core.musig.ctypes.create_string_buffer", side_effect=allocate): + with self.assertRaisesRegex(MemoryError, "injected allocation failure"): + sign_partial(secnonce, bytes(keys[0]), session) + + self.assertEqual(nonce_buffer.raw, bytes(ctypes.sizeof(nonce_buffer))) + for buffer in allocated: + self.assertEqual(buffer.raw, bytes(ctypes.sizeof(buffer))) + with self.assertRaisesRegex(MuSig2Error, "already been consumed"): + sign_partial(secnonce, bytes(keys[0]), session) + + def test_mutable_seed_is_wiped_after_nonce_generation_failures(self) -> None: + key = CKey(b"\x00" * 31 + b"\x01") + pubkey = bytes(key.pub) + secp256k1 = get_secp256k1() + + seed = bytearray(b"\xb1" * 32) + with patch.object(secp256k1.lib, "secp256k1_musig_nonce_gen", return_value=0) as mocked: + with self.assertRaisesRegex(MuSig2Error, "nonce generation failed"): + nonce_gen(pubkey, seed) + mocked.assert_called_once() + self.assertEqual(seed, bytearray(32)) + + seed = bytearray(b"\xb2" * 32) + create_buffer = ctypes.create_string_buffer + allocation_sizes: list[int] = [] + + def allocate(size: int) -> Any: + allocation_sizes.append(size) + if len(allocation_sizes) == 2: + raise MemoryError("injected allocation failure") + return create_buffer(size) + + with patch("bitcointx.core.musig.ctypes.create_string_buffer", side_effect=allocate): + with self.assertRaisesRegex(MemoryError, "injected allocation failure"): + nonce_gen(pubkey, seed) + self.assertEqual(allocation_sizes, [64, 132]) + self.assertEqual(seed, bytearray(32)) + + seed = bytearray(b"\xb3" * 32) + with patch.object( + secp256k1.lib, "secp256k1_musig_pubnonce_serialize", return_value=0 + ) as mocked: + with self.assertRaisesRegex(MuSig2Error, "could not serialize the MuSig2 public nonce"): + nonce_gen(pubkey, seed) + mocked.assert_called_once() + self.assertEqual(seed, bytearray(32)) + + keys, _, message, session_data, _, _ = self._two_party_session() + session, secnonce_one, secnonce_two = session_data + signature = partial_sig_agg( + [ + sign_partial(secnonce_one, bytes(keys[0]), session), + sign_partial(secnonce_two, bytes(keys[1]), session), + ], + session, + ) + self.assertTrue( + XOnlyPubKey(session.ctx.aggregate_xonly()).verify_schnorr(message, signature) + ) + + +if __name__ == "__main__": + unittest.main()