# Copyright (c) The btclib developers
# Distributed under the MIT software license, see the accompanying
# LICENSE file or https://opensource.org/license/mit for the full text.
"""FROST threshold Schnorr signing, according to BIP445.
https://github.com/bitcoin/bips/pull/2070, `bip-0445.md` at v0.10.0,
head `8e25d579` on `siv2r:bip-frost-signing`. The BIP is a draft and not
in `bitcoin/bips` `master`; `tests/_data/README.md` records why v0.10.0
is where this module pins.
FROST turns *t* of *n* participants into one signer: a threshold secret
key is Shamir-shared among *n* participants at key generation, and any
*t* of them can cooperatively produce a signature that verifies under a
single x-only public key with `btclib_ecc.ecc.ssa` and with any other BIP340
verifier -- the same output shape `btclib_ecc.ecc.musig2` produces for an
*n*-of-*n* group, with a chosen subset standing in for the whole group.
**Key generation is out of scope, here as in the BIP.** This module
takes a threshold secret key already split into shares -- a *secshare*
per participant, all of them scalar multiples of the same secret
polynomial's constant term -- and never splits one itself. The vectors
this module is tested against are generated by a trusted dealer, and
`validate_threshold_info` checks key material a dealer or a DKG produced
rather than producing any: it asks whether a threshold info's own public
shares reconstruct its own threshold public key, which functional
compatibility needs and which is not a statement about how that key
material came to exist. `btclib-org/btclib#257` is where interactive
threshold signing is tracked; it stays open, `upstream-dependent`, and is
the hook that brings key generation and ChillDKG back once BIP445 or
BIP-FROST-DKG settle further.
The round shape mirrors `btclib_ecc.ecc.musig2`: a primitive per round and
no function that signs, because signing is interactive. Key material and nonces
come from the caller, not from a session this module holds:
- setup, once per group: `validate_threshold_info` on the *n*-participant
`ThresholdInfo` a dealer or a DKG produced;
- tweaking, optional and once per session: `tweak_ctx_init` and
`apply_tweak`, or `thresh_pubkey_and_tweak` for both in one call,
answering a `TweakContext`;
- round 1, the nonces: `nonce_gen` for each of the *u* selected signers,
then `nonce_agg` over what they published, which anyone can do;
- round 2, the partial signatures: `sign` (or `deterministic_sign`) for
each signer, `partial_sig_verify` to hold a signer to what it sent, and
`partial_sig_agg` for the aggregate signature.
**A secret nonce signs once**, exactly as in `musig2`: `sign` zeroes the
64-byte bytearray it is given before using it, so calling it twice with
the same buffer reads two zero scalars and raises rather than signing
twice -- which is why its argument is a `bytearray` and not `bytes`.
**Participant vs. signer.** *n* participants exist from key generation
on, each with an identifier in `0 .. n-1`, a *secshare* and a *pubshare*.
A signing session selects *u* of them, *t <= u <= n*; that subset is the
*signers*, and every `ids` argument below names their identifiers,
sorted or not as the caller built the list. `ThresholdInfo.pub_shares`
is indexed by identifier over all *n* participants and may hold `None`
where a share is unknown; a `SessionContext`'s own `pub_shares`, when
given at all, is the *u*-long list for the selected signers alone --
absent entirely is also valid, a signer able to produce and verify
against its own partial signature without it, `partial_sig_verify`
alone requiring the list because there is no other way to check a
signature against a key it was not given.
secp256k1 and sha256 are not parameters here, unlike everywhere else in
`btclib_ecc.ecc`, for the reason `musig2`'s own docstring gives: BIP445 is
defined for that pair alone, and a `Curve` argument would advertise a
genericity no vector could check.
The upper bound of 128 participants (`n <= 128`) is a security
requirement and not a format one: FROST3's unforgeability proof assumes
the set of compromised participants is fixed before key generation, and
an adversary that instead chooses whom to compromise after seeing the
public shares can, for large *n*, forge with fewer than *t* compromised
participants -- provided it solves a search problem BIP445 cites as
provably hard whenever *n <= 131*. 128 stays inside that bound with a
margin; it is not raised by passing a larger *n* to any function here.
The tags below are BIP445's own and disjoint from BIP327's: `BIP0445/aux`,
`BIP0445/nonce`, `BIP0445/noncecoef` and `BIP0445/deterministic/nonce`.
Only `BIP0340/challenge` is shared, and it is BIP340's, read the same way
`musig2` reads it -- through `btclib_ecc.ecc.ssa.challenge_`.
**Nothing here is shared with `btclib_ecc.ecc.musig2` by import**, though
BIP445's own reference code says in its header that it is "copied or
modified from" BIP327's: this module's tag bytes, several error
messages (the tweak-range message among them) and the shape of
`ThresholdInfo`/interpolation differ from BIP327's, and the
functions that do look alike -- `nonce_agg`, `apply_tweak` -- were
checked line by line against BIP445's own reference rather than assumed
identical from the name. Where a check found the same construction, this
module keeps its own copy rather than importing `musig2`'s private
helpers: the two modules are each a self-contained mirror of one BIP,
which is the shape `musig2.py` itself already has no sibling to share
with.
"""
from __future__ import annotations
import secrets
from collections.abc import Sequence
from dataclasses import dataclass, field
from hashlib import sha256
from typing import cast
from btclib_ecc._utils import assert_type, bytes_from_octets, is_integer, is_octets
from btclib_ecc.alias import INF, Octets, Point
from btclib_ecc.curves import mult, multi_mult_var, secp256k1
from btclib_ecc.curves.curve import _sum_var, _tweak_add_var
from btclib_ecc.curves.sec_point import bytes_from_point, point_from_octets
from btclib_ecc.ecc import ssa
from btclib_ecc.exceptions import (
BTClibEccTypeError,
BTClibEccValueError,
InvalidContributionError,
)
from btclib_ecc.hashes import tagged_hash
__all__ = [
"ID_SIZE",
"MAX_PARTICIPANTS",
"PK_SIZE",
"SessionContext",
"SessionValues",
"ThresholdInfo",
"TweakContext",
"apply_tweak",
"deterministic_sign",
"nonce_agg",
"nonce_gen",
"nonce_gen_",
"partial_sig_agg",
"partial_sig_verify",
"partial_sig_verify_",
"session_values",
"sign",
"thresh_pubkey_and_tweak",
"tweak_ctx_init",
"validate_threshold_info",
]
# BIP445's own tags: disjoint from BIP327's, so nothing here collides
# with `btclib_ecc.ecc.musig2` even though several are folded the same way
_AUX_TAG = b"BIP0445/aux"
_NONCE_TAG = b"BIP0445/nonce"
_NONCE_COEFF_TAG = b"BIP0445/noncecoef"
_DET_NONCE_TAG = b"BIP0445/deterministic/nonce"
# a compressed point, a scalar, a public nonce (two compressed points),
# a participant identifier. The first and the last are exported with the
# bound below, being the widths a psbt carrying this key material is
# written in
PK_SIZE = 33
_SCALAR_SIZE = 32
_NONCE_SIZE = 66
ID_SIZE = 4
# the infinity point as 33 zero bytes, BIP445's own placeholder for the
# same reason BIP327's is: an aggregate nonce is always 66 bytes on the
# wire even where one half sums to infinity
_INF_BYTES = bytes(PK_SIZE)
# the security bound of the footnote in the module docstring: n <= 128
# keeps the LDVR search problem's hardness inside the range BIP445 cites
MAX_PARTICIPANTS = 128
# BIP445's own verbatim strings, compared byte for byte by the vectors
# that name a "value" error (`assert_error` in `tests/ecc/frost_test.py`,
# mirroring `musig2_test.py`'s own). Several read like BIP327's and are
# not: the tweaks/is_xonly one drops the backticks BIP327's carries,
# and the tweak-range one is a different sentence rather than the same
# one reworded. Both are copied from BIP445's own reference rather than
# from `musig2.py`, per the module docstring
_THRESHOLD_RANGE_ERR = "The threshold must be 1 <= t <= n."
_MAX_PARTICIPANTS_ERR = f"The number of participants must be n <= {MAX_PARTICIPANTS}."
_TWEAK_SIZE_ERR = "The tweak must be a 32-byte array."
_TWEAK_RANGE_ERR = "The tweak value is out of range."
_TWEAK_INF_ERR = "The result of tweaking cannot be infinity."
_TWEAKS_LEN_ERR = "The tweaks and is_xonly arrays must have the same length."
_KEY_MATERIAL_POLY_ERR = (
"The provided key material is incorrect: the public shares do not lie"
" on a single polynomial."
)
_KEY_MATERIAL_MATCH_ERR = (
"The provided key material is incorrect: the public shares do not match"
" the threshold public key."
)
_THRESH_PK_INF_ERR = "The threshold public key must not be the point at infinity."
def _cbytes(P: Point) -> bytes:
return bytes_from_point(P, secp256k1)
def _cbytes_ext(P: Point) -> bytes:
"""Serialize a point that may be the infinity one."""
return _INF_BYTES if P[1] == 0 else _cbytes(P)
def _cpoint(octets: Octets) -> Point:
"""Deserialize a compressed point, refusing anything else."""
return point_from_octets(bytes_from_octets(octets, PK_SIZE), secp256k1)
def _cpoint_ext(octets: Octets) -> Point:
"""Deserialize a point that may be the infinity one."""
data = bytes_from_octets(octets, PK_SIZE)
return INF if data == _INF_BYTES else _cpoint(data)
def _bytes_xor(a: bytes, b: bytes) -> bytes:
return bytes(x ^ y for x, y in zip(a, b, strict=True))
def _assert_octets_sequence(value: object, what: str) -> None:
"""Refuse anything but a Sequence of Octets, naming which.
`musig2._assert_octets_sequence`'s own reason (issue
btclib-org/btclib#1405): every Octets is itself a Sequence, so a caller
passing one where a list of pubshares or tweaks was meant would zip through
its bytes silently.
"""
if is_octets(value) or not isinstance(value, Sequence):
raise BTClibEccTypeError(f"invalid {what} type: {type(value).__name__}")
def _integer(value: object, what: str) -> int:
"""Return `value` if it is an integer, a bool not being one.
`int(value)` would turn 1.9 into 1 and "1" into 1, so a session
would be built for other signers than the ones named.
"""
if not is_integer(value):
raise BTClibEccTypeError(f"invalid {what} type: {type(value).__name__}")
return cast("int", value)
def _ids(ids: Sequence[int]) -> tuple[int, ...]:
"""Return `ids` as a tuple, each one an integer.
A str or bytes is refused whole, as a Sequence of what are not ids:
`_assert_octets_sequence` is the check, its name notwithstanding.
"""
_assert_octets_sequence(ids, "ids")
return tuple(_integer(i, "id") for i in ids)
def _pub_shares(pub_shares: Sequence[Octets]) -> tuple[bytes, ...]:
return tuple(bytes_from_octets(ps, PK_SIZE) for ps in pub_shares)
def _tweaks(tweaks: Sequence[Octets]) -> tuple[bytes, ...]:
return tuple(bytes_from_octets(t) for t in tweaks)
def _require_same_length(tweaks: Sequence[bytes], is_xonly: Sequence[bool]) -> None:
if len(tweaks) != len(is_xonly):
raise BTClibEccValueError(_TWEAKS_LEN_ERR)
def _flag(is_xonly: bool) -> bool:
"""Return the kind of a tweak, which is a `bool` and not a truth.
`musig2._flag`'s own reason: a value read for its truth would tweak
an odd-y key the other way and answer another threshold public key.
"""
assert_type(is_xonly, bool, "is_xonly")
return is_xonly
def _flags(is_xonly: Sequence[bool]) -> tuple[bool, ...]:
return tuple(_flag(flag) for flag in is_xonly)
def _has_duplicates(ids: Sequence[int]) -> bool:
return len(set(ids)) != len(ids)
def _serialize_ids(ids: Sequence[int]) -> bytes:
"""Serialize identifiers sorted, so the nonce coefficient is order-free."""
return b"".join(i.to_bytes(ID_SIZE, "big") for i in sorted(ids))
def _derive_interpolating_value(ids: Sequence[int], my_id: int) -> int:
"""Return the Lagrange coefficient of `my_id` for the secret at x = -1.
Assumes `ids` holds no duplicate, which `_validate_session_params`
has checked before either caller reaches this. `my_id in ids` is
checked by its callers, `sign` and `partial_sig_verify_`: an id the
session does not hold has no `curr_id == my_id` to match, and the
coefficient returned interpolates nothing.
"""
num = 1
deno = 1
for curr_id in ids:
if curr_id == my_id:
continue
num = num * (curr_id + 1) % secp256k1.n
deno = deno * (curr_id - my_id) % secp256k1.n
return num * pow(deno, -1, secp256k1.n) % secp256k1.n
def _derive_pubshare_at(
ids: Sequence[int], pub_shares: Sequence[Point], x: int
) -> Point:
"""Interpolate the polynomial `pub_shares` sit on, evaluated at `x`.
The general form `_derive_interpolating_value` is not: that one fixes
the evaluation point at the secret's own x = -1, this one takes any
x, which is what checking a further pubshare against a base set, or
reconstructing the threshold public key, both need. May answer the
point at infinity, which is a value here and a refusal only where a
caller -- `_derive_thresh_pubkey` -- says it must not be.
"""
coefficients = []
for my_id in ids:
num = 1
deno = 1
for curr_id in ids:
if curr_id == my_id:
continue
num = num * (x - curr_id) % secp256k1.n
deno = deno * (my_id - curr_id) % secp256k1.n
coefficients.append(num * pow(deno, -1, secp256k1.n) % secp256k1.n)
if len(pub_shares) == 1:
return mult(coefficients[0], pub_shares[0], secp256k1)
return multi_mult_var(coefficients, pub_shares, secp256k1)
def _derive_thresh_pubkey(ids: Sequence[int], pub_shares: Sequence[Point]) -> bytes:
"""Reconstruct the threshold public key `pub_shares` were split from."""
Q = _derive_pubshare_at(ids, pub_shares, -1)
if Q[1] == 0:
raise BTClibEccValueError(_THRESH_PK_INF_ERR)
return _cbytes(Q)
[docs]
@dataclass(frozen=True, init=False)
class ThresholdInfo:
"""Key material a signer holds after key generation, common to a session.
`t`, the threshold public key, and every participant's public share
it knows -- `pub_shares` is `n` long, indexed by identifier, `None`
where a share is unknown to this holder. `validate_threshold_info`
checks it: that every subset of `t` present shares it selects
reconstructs the same threshold public key, which is what makes the
rest of this module answer the same signature regardless of which
`t` of the present shares a caller happened to check against.
A key generation protocol answers this, and this module never builds
one: `btclib-org/btclib#257` is where producing it -- a trusted
dealer, ChillDKG -- is tracked.
"""
t: int
thresh_pk: bytes
pub_shares: tuple[bytes | None, ...]
def __init__(
self, t: int, thresh_pk: Octets, pub_shares: Sequence[Octets | None]
) -> None:
_assert_octets_sequence(pub_shares, "pub_shares")
object.__setattr__(self, "t", _integer(t, "t"))
object.__setattr__(self, "thresh_pk", bytes_from_octets(thresh_pk, PK_SIZE))
object.__setattr__(
self,
"pub_shares",
tuple(
None if ps is None else bytes_from_octets(ps, PK_SIZE)
for ps in pub_shares
),
)
def _parse_present_pub_shares(
pub_shares: Sequence[bytes | None],
) -> tuple[list[int], list[Point]]:
"""Return the identifiers and points of every non-`None` entry.
Split out of `validate_threshold_info` so that function reads as one
check per line rather than a loop with two nested reasons to raise.
"""
parsed_ids: list[int] = []
parsed_points: list[Point] = []
for i, pub_share in enumerate(pub_shares):
if pub_share is None:
continue
try:
point = _cpoint(pub_share)
except BTClibEccValueError as e:
raise BTClibEccValueError(f"Invalid pubshare at index {i}.") from e
parsed_ids.append(i)
parsed_points.append(point)
return parsed_ids, parsed_points
[docs]
def validate_threshold_info(info: ThresholdInfo) -> None:
"""Check a `ThresholdInfo`'s own shares reconstruct its own key.
Passing this is functional compatibility with the signing protocol,
not a statement about the key generation protocol that produced the
shares: a dealer or a DKG that already colluded against the group
could still pass it, having every incentive to (BIP445's own
warning). What it rules out is key material that cannot be signed
with consistently -- shares that do not sit on one polynomial, or
that sit on one whose constant term is not the given threshold
public key.
No test vector exercises this: `tests/_data/README.md` records why.
"""
t, thresh_pk, pub_shares = info.t, info.thresh_pk, info.pub_shares
n = len(pub_shares)
if not 1 <= t <= n:
raise BTClibEccValueError(_THRESHOLD_RANGE_ERR)
if n > MAX_PARTICIPANTS:
raise BTClibEccValueError(_MAX_PARTICIPANTS_ERR)
try:
_cpoint(thresh_pk)
except BTClibEccValueError as e:
raise BTClibEccValueError("Invalid threshold public key.") from e
parsed_ids, parsed_points = _parse_present_pub_shares(pub_shares)
if len(parsed_ids) < t:
raise BTClibEccValueError("At least t pubshares must be present.")
base_ids, base_points = parsed_ids[:t], parsed_points[:t]
for i, point in zip(parsed_ids[t:], parsed_points[t:], strict=True):
if _derive_pubshare_at(base_ids, base_points, i) != point:
raise BTClibEccValueError(_KEY_MATERIAL_POLY_ERR)
if _derive_thresh_pubkey(base_ids, base_points) != thresh_pk:
raise BTClibEccValueError(_KEY_MATERIAL_MATCH_ERR)
[docs]
@dataclass(frozen=True)
class TweakContext:
"""The threshold public key, and what tweaking it has accumulated.
`musig2.KeyAggContext`'s own shape and reason: Q is the tweaked
point, `gacc` the product of the negations x-only tweaking forced,
`tacc` the sum of the tweaks -- so that the partial signatures still
add up under the even-y Q a BIP340 verifier assumes, and
`partial_sig_agg` adds the tweak contribution in once for the group
rather than each signer adding a share of it. Unlike
`KeyAggContext`, this is not built by aggregating several keys --
FROST's threshold public key comes from key generation whole --
which is why `tweak_ctx_init` takes one key and not a list.
"""
Q: Point
gacc: int
tacc: int
@property
def x_only_pub_key(self) -> bytes:
"""Return the 32-byte x-only key, for BIP341 Taproot tweaking."""
return self.Q[0].to_bytes(secp256k1.p_size, "big")
@property
def plain_pub_key(self) -> bytes:
"""Return the 33-byte compressed key, for BIP32 plain tweaking."""
return _cbytes(self.Q)
[docs]
def tweak_ctx_init(thresh_pk: Octets) -> TweakContext:
"""Return a fresh `TweakContext` over an untweaked threshold public key."""
return TweakContext(_cpoint(thresh_pk), 1, 0)
[docs]
def apply_tweak(tweak_ctx: TweakContext, tweak: Octets, is_xonly: bool) -> TweakContext:
"""Return the context tweaked by t, x-only or plain.
`musig2.apply_tweak`'s own reasoning for the negation and the
accumulators holds identically here; only the range-error message
differs from BIP327's, per the module docstring.
"""
tweak = bytes_from_octets(tweak)
if len(tweak) != _SCALAR_SIZE:
raise BTClibEccValueError(_TWEAK_SIZE_ERR)
Q, gacc, tacc = tweak_ctx.Q, tweak_ctx.gacc, tweak_ctx.tacc
g = secp256k1.n - 1 if (_flag(is_xonly) and Q[1] % 2) else 1
t = int.from_bytes(tweak, "big")
if t >= secp256k1.n:
raise BTClibEccValueError(_TWEAK_RANGE_ERR)
Q = _tweak_add_var(Q if g == 1 else secp256k1.negate(Q), t, secp256k1)
if Q[1] == 0:
raise BTClibEccValueError(_TWEAK_INF_ERR)
return TweakContext(Q, g * gacc % secp256k1.n, (t + g * tacc) % secp256k1.n)
[docs]
def thresh_pubkey_and_tweak(
thresh_pk: Octets, tweaks: Sequence[Octets], is_xonly: Sequence[bool]
) -> TweakContext:
"""Build a `TweakContext` and apply the tweaks in order."""
_assert_octets_sequence(tweaks, "tweaks")
tweaks_ = _tweaks(tweaks)
is_xonly_ = _flags(is_xonly)
_require_same_length(tweaks_, is_xonly_)
tweak_ctx = tweak_ctx_init(thresh_pk)
for tweak, xonly in zip(tweaks_, is_xonly_, strict=True):
tweak_ctx = apply_tweak(tweak_ctx, tweak, xonly)
return tweak_ctx
def _nonce_hash(
rand: bytes,
pub_share: bytes,
thresh_pk_xonly: bytes,
i: int,
msg_prefixed: bytes,
extra_in: bytes,
) -> int:
buf = b"".join(
[
rand,
len(pub_share).to_bytes(1, "big"),
pub_share,
len(thresh_pk_xonly).to_bytes(1, "big"),
thresh_pk_xonly,
msg_prefixed,
len(extra_in).to_bytes(4, "big"),
extra_in,
i.to_bytes(1, "big"),
]
)
return int.from_bytes(tagged_hash(_NONCE_TAG, buf), "big")
[docs]
def nonce_gen_(
rand_: Octets,
sec_share: Octets | None,
pub_share: Octets | None,
thresh_pk_xonly: Octets | None = None,
msg: Octets | None = None,
extra_in: Octets | None = None,
) -> tuple[bytearray, bytes]:
"""Return the (secnonce, pubnonce) pair of one signer, given ``rand_``.
`musig2.nonce_gen_`'s own reasons for the trailing underscore and for
the returned secnonce being a mutable `bytearray` hold identically
here. Where MuSig2's secnonce also carries the signer's own plain
public key -- 97 bytes, so `sign` can check it was not swapped --
FROST's is 64 bytes exactly, `k_1 || k_2` and nothing else: BIP445
defines no such check, and this module raises nothing this pair does
not already answer.
Every optional argument beside `rand_` is a defense-in-depth
measure and none, `sec_share` included, is required. `msg` is the
one whose absence differs from an empty value: absent hashes as a
single mode byte, present as the other mode byte followed by the
length and the bytes. An absent `extra_in` and an empty one are the
same input, and `sec_share`, `pub_share` and `thresh_pk_xonly` have
no empty value at all, each being of a fixed size.
"""
rand_ = bytes_from_octets(rand_, _SCALAR_SIZE)
if sec_share is None:
rand = rand_
else:
rand = _bytes_xor(
bytes_from_octets(sec_share, _SCALAR_SIZE), tagged_hash(_AUX_TAG, rand_)
)
pub_share_bytes = (
b"" if pub_share is None else bytes_from_octets(pub_share, PK_SIZE)
)
thresh_pk_xonly_bytes = (
b""
if thresh_pk_xonly is None
else bytes_from_octets(thresh_pk_xonly, secp256k1.p_size)
)
if msg is None:
msg_prefixed = b"\x00"
else:
msg = bytes_from_octets(msg)
msg_prefixed = b"\x01" + len(msg).to_bytes(8, "big") + msg
extra = b"" if extra_in is None else bytes_from_octets(extra_in)
k_1 = (
_nonce_hash(
rand, pub_share_bytes, thresh_pk_xonly_bytes, 0, msg_prefixed, extra
)
% secp256k1.n
)
k_2 = (
_nonce_hash(
rand, pub_share_bytes, thresh_pk_xonly_bytes, 1, msg_prefixed, extra
)
% secp256k1.n
)
pub_nonce = _cbytes(mult(k_1, ec=secp256k1)) + _cbytes(mult(k_2, ec=secp256k1))
sec_nonce = bytearray(
k_1.to_bytes(_SCALAR_SIZE, "big") + k_2.to_bytes(_SCALAR_SIZE, "big")
)
return sec_nonce, pub_nonce
[docs]
def nonce_gen(
sec_share: Octets | None = None,
pub_share: Octets | None = None,
thresh_pk_xonly: Octets | None = None,
msg: Octets | None = None,
extra_in: Octets | None = None,
) -> tuple[bytearray, bytes]:
"""Return the (secnonce, pubnonce) pair of one signer.
Fresh randomness is drawn here; ``nonce_gen_`` is the spelling that
takes it, for the test vectors.
"""
return nonce_gen_(
secrets.token_bytes(_SCALAR_SIZE),
sec_share,
pub_share,
thresh_pk_xonly,
msg,
extra_in,
)
[docs]
def nonce_agg(pub_nonces: Sequence[Octets]) -> bytes:
"""Aggregate the public nonces of round 1 into the 66-byte aggnonce.
`musig2.nonce_agg`'s own construction, checked line by line against
BIP445's reference rather than assumed identical from the name: both
sum each half of the pair independently and serialize a half that
sums to infinity as the 33-zero-byte placeholder, refusing that from
one signer alone would let it stop the session at will.
"""
_assert_octets_sequence(pub_nonces, "pub_nonces")
nonces = [bytes_from_octets(on, _NONCE_SIZE) for on in pub_nonces]
agg_nonce = b""
for j in (0, 1):
R_j: list[Point] = []
for i, pub_nonce in enumerate(nonces):
try:
R_j.append(_cpoint(pub_nonce[j * PK_SIZE : (j + 1) * PK_SIZE]))
except BTClibEccValueError as e:
raise InvalidContributionError(i, "pubnonce") from e
agg_nonce += _cbytes_ext(_sum_var(R_j, secp256k1))
return agg_nonce
def _parse_session_pub_shares(
ids: Sequence[int], n: int, pub_shares: Sequence[bytes] | None
) -> list[Point]:
"""Return every pubshare's point, checking each id and share in step.
Split out of `_validate_session_params` for the reason
`_parse_present_pub_shares` is split out of `validate_threshold_info`.
"""
pub_share_points: list[Point] = []
for idx, signer_id in enumerate(ids):
if not 0 <= signer_id <= n - 1:
raise BTClibEccValueError(f"Invalid id at index {idx}")
if pub_shares is not None:
try:
pub_share_points.append(_cpoint(pub_shares[idx]))
except BTClibEccValueError as e:
raise BTClibEccValueError(f"Invalid pubshare at index {idx}.") from e
return pub_share_points
def _validate_session_params(
n: int,
t: int,
ids: Sequence[int],
pub_shares: Sequence[bytes] | None,
thresh_pk: bytes,
) -> None:
if not 1 <= t <= n:
raise BTClibEccValueError(_THRESHOLD_RANGE_ERR)
if n > MAX_PARTICIPANTS:
raise BTClibEccValueError(_MAX_PARTICIPANTS_ERR)
if not t <= len(ids) <= n:
raise BTClibEccValueError("The number of signers must be between t and n.")
if pub_shares is not None and len(pub_shares) != len(ids):
raise BTClibEccValueError(
"The pubshares and ids lists must have the same length."
)
pub_share_points = _parse_session_pub_shares(ids, n, pub_shares)
if _has_duplicates(ids):
raise BTClibEccValueError("The ids list contains duplicate elements.")
if (
pub_shares is not None
and _derive_thresh_pubkey(ids, pub_share_points) != thresh_pk
):
raise BTClibEccValueError(_KEY_MATERIAL_MATCH_ERR)
[docs]
@dataclass(frozen=True, init=False)
class SessionContext:
"""Everything the signers of one session have to agree on.
The aggregate nonce, the total and threshold participant counts, the
selected signers' identifiers, their public shares (or `None`, a
session that does not carry them), the threshold public key, the
tweaks with their kinds, and the message. `musig2.SessionContext`'s
own shape, with FROST's extra fields -- `n`, `t`, `ids`, `pub_shares`
-- in place of MuSig2's aggregated `pub_keys`: nothing here
aggregates a key list, the threshold public key already being one
key from key generation.
"""
n: int
t: int
ids: tuple[int, ...]
pub_shares: tuple[bytes, ...] | None
thresh_pk: bytes
agg_nonce: bytes
tweaks: tuple[bytes, ...]
is_xonly: tuple[bool, ...]
msg: bytes
# `musig2.SessionContext._values`'s own idiom: memoized on first call
# to `session_values`, excluded from equality by `compare=False` so
# two contexts spelling the same session stay equal regardless of
# which has already been used to sign
_values: SessionValues | None = field(
default=None, init=False, repr=False, compare=False
)
def __init__(
self,
n: int,
t: int,
ids: Sequence[int],
pub_shares: Sequence[Octets] | None,
thresh_pk: Octets,
agg_nonce: Octets,
tweaks: Sequence[Octets],
is_xonly: Sequence[bool],
msg: Octets,
) -> None:
if pub_shares is not None:
_assert_octets_sequence(pub_shares, "pub_shares")
_assert_octets_sequence(tweaks, "tweaks")
object.__setattr__(self, "n", _integer(n, "n"))
object.__setattr__(self, "t", _integer(t, "t"))
object.__setattr__(self, "ids", _ids(ids))
object.__setattr__(
self, "pub_shares", None if pub_shares is None else _pub_shares(pub_shares)
)
object.__setattr__(self, "thresh_pk", bytes_from_octets(thresh_pk, PK_SIZE))
object.__setattr__(self, "agg_nonce", bytes_from_octets(agg_nonce, _NONCE_SIZE))
object.__setattr__(self, "tweaks", _tweaks(tweaks))
object.__setattr__(self, "is_xonly", _flags(is_xonly))
object.__setattr__(self, "msg", bytes_from_octets(msg))
_require_same_length(self.tweaks, self.is_xonly)
[docs]
@dataclass(frozen=True)
class SessionValues:
"""What every signer derives from a `SessionContext` before signing.
- Q, gacc and tacc are the tweaked threshold public key and its
tweak accumulators, from `thresh_pubkey_and_tweak`
- b is the coefficient combining the two halves of the nonce
- R is the effective nonce point, R_1 + b*R_2
- e is the BIP340 challenge, over R, Q and the message
Unlike `musig2.SessionValues`, there is no cached key-aggregation
coefficient here: `_derive_interpolating_value` is a pure function of
`session_ctx.ids` and one signer's own id, and both `sign` and
`partial_sig_verify_` read `ids` straight off the context rather than
off a value cached here.
"""
Q: Point
gacc: int
tacc: int
b: int
R: Point
e: int
[docs]
def session_values(session_ctx: SessionContext) -> SessionValues:
"""Derive the session values from the context, as every party does.
Memoized on `session_ctx`, for `musig2.session_values`'s own reason:
`sign`, `partial_sig_verify_` and `partial_sig_agg` each call this
once per signer a session touches, and the tweaked threshold public
key does not change between them.
"""
if session_ctx._values is not None:
return session_ctx._values
_validate_session_params(
session_ctx.n,
session_ctx.t,
session_ctx.ids,
session_ctx.pub_shares,
session_ctx.thresh_pk,
)
tweak_ctx = thresh_pubkey_and_tweak(
session_ctx.thresh_pk, session_ctx.tweaks, session_ctx.is_xonly
)
Q, gacc, tacc = tweak_ctx.Q, tweak_ctx.gacc, tweak_ctx.tacc
coeff_hash = tagged_hash(
_NONCE_COEFF_TAG,
len(session_ctx.ids).to_bytes(4, "big")
+ _serialize_ids(session_ctx.ids)
+ session_ctx.agg_nonce
+ tweak_ctx.x_only_pub_key
+ session_ctx.msg,
)
b = int.from_bytes(coeff_hash, "big") % secp256k1.n
try:
R_1 = _cpoint_ext(session_ctx.agg_nonce[:PK_SIZE])
R_2 = _cpoint_ext(session_ctx.agg_nonce[PK_SIZE:])
except BTClibEccValueError as err:
raise InvalidContributionError(None, "aggnonce") from err
R = secp256k1.add_var(R_1, mult(b, R_2, secp256k1))
# an aggregate nonce of infinity is a session the signers can still
# complete: musig2.session_values's own reasoning, BIP445 keeping the
# same modification for the same reason (module docstring's citation)
if R[1] == 0:
R = secp256k1.G
e = ssa.challenge_(session_ctx.msg, Q[0], R[0], secp256k1, sha256)
values = SessionValues(Q, gacc, tacc, b, R, e)
object.__setattr__(session_ctx, "_values", values)
return values
[docs]
def sign(
sec_nonce: bytearray, sec_share: Octets, my_id: int, session_ctx: SessionContext
) -> bytes:
"""Return the 32-byte partial signature of one signer.
The secnonce is consumed exactly as `musig2.sign` consumes its own:
zeroed the moment it is read, before either half is used, so calling
this twice with the same bytearray raises "out of range" instead of
handing out the secret share.
`session_values` runs first, for the same reason `musig2.sign`'s own
docstring gives: a session that does not assemble raises before the
nonce is touched, so the same bytearray may be used for the
corrected session.
`sec_nonce` is checked exactly `bytearray` first, for the reason
`musig2.sign`'s own docstring gives (issue
btclib-org/btclib-ecc#11).
"""
assert_type(sec_nonce, bytearray, "sec_nonce")
_integer(my_id, "my_id")
values = session_values(session_ctx)
k_1_ = int.from_bytes(sec_nonce[:_SCALAR_SIZE], "big")
k_2_ = int.from_bytes(sec_nonce[_SCALAR_SIZE : 2 * _SCALAR_SIZE], "big")
sec_nonce[: 2 * _SCALAR_SIZE] = bytearray(2 * _SCALAR_SIZE)
if not 0 < k_1_ < secp256k1.n:
raise BTClibEccValueError("first secnonce value is out of range.")
if not 0 < k_2_ < secp256k1.n:
raise BTClibEccValueError("second secnonce value is out of range.")
if values.R[1] % 2:
k_1_, k_2_ = secp256k1.n - k_1_, secp256k1.n - k_2_
d_bytes = bytes_from_octets(sec_share, _SCALAR_SIZE)
d_ = int.from_bytes(d_bytes, "big")
if not 0 < d_ < secp256k1.n:
raise BTClibEccValueError("The signer's secret share value is out of range.")
pub_share = _cbytes(mult(d_, ec=secp256k1))
if my_id not in session_ctx.ids:
raise BTClibEccValueError("The signer's id is missing from the ids list.")
if session_ctx.pub_shares is not None:
j = session_ctx.ids.index(my_id)
if session_ctx.pub_shares[j] != pub_share:
raise BTClibEccValueError(
"The signer's pubshare is missing from the pubshares list."
)
a = _derive_interpolating_value(session_ctx.ids, my_id)
g = 1 if values.Q[1] % 2 == 0 else secp256k1.n - 1
d = g * values.gacc * d_ % secp256k1.n
s = (k_1_ + values.b * k_2_ + values.e * a * d) % secp256k1.n
return s.to_bytes(_SCALAR_SIZE, "big")
def _det_nonce_hash(
sec_share_: bytes,
my_id: int,
ids: Sequence[int],
agg_other_nonce: bytes,
tweaked_thresh_pk_xonly: bytes,
msg: bytes,
i: int,
) -> int:
buf = b"".join(
[
sec_share_,
my_id.to_bytes(ID_SIZE, "big"),
len(ids).to_bytes(ID_SIZE, "big"),
_serialize_ids(ids),
agg_other_nonce,
tweaked_thresh_pk_xonly,
len(msg).to_bytes(8, "big"),
msg,
i.to_bytes(1, "big"),
]
)
return int.from_bytes(tagged_hash(_DET_NONCE_TAG, buf), "big")
[docs]
def deterministic_sign(
sec_share: Octets,
my_id: int,
agg_other_nonce: Octets | None,
n: int,
t: int,
ids: Sequence[int],
pub_shares: Sequence[Octets] | None,
thresh_pk: Octets,
tweaks: Sequence[Octets],
is_xonly: Sequence[bool],
msg: Octets,
aux_rand: Octets | None = None,
) -> tuple[bytes, bytes]:
"""Return the (pubnonce, partial signature) of a signer with no RNG.
`musig2.deterministic_sign`'s own reason: the two rounds collapse
into one for the *last* signer to act, given the aggregate of
everybody else's nonces. Unlike MuSig2, which is always n-of-n, the
derived nonce also binds the *signer set* `ids` -- without that, a
malicious coordinator could recover the secret share by replaying
the session with the signer set varied each time, which cannot
happen to MuSig2's `DeterministicSign` because its signer set is
fixed by the protocol.
`agg_other_nonce` is omitted (`None`) by a sole signer, who has no
other signers' nonces to aggregate; passed, it should be the
aggregate of every other signer's pubnonce, and may come from an
untrusted coordinator -- `sign`, reached at the end of this, is what
catches a session it does not assemble into.
"""
n = _integer(n, "n")
t = _integer(t, "t")
ids_ = _ids(ids)
my_id = _integer(my_id, "my_id")
_pub_shares_ = None if pub_shares is None else _pub_shares(pub_shares)
thresh_pk_bytes = bytes_from_octets(thresh_pk, PK_SIZE)
_validate_session_params(n, t, ids_, _pub_shares_, thresh_pk_bytes)
sec_share_bytes = bytes_from_octets(sec_share, _SCALAR_SIZE)
if aux_rand is not None:
sec_share_ = _bytes_xor(
sec_share_bytes, tagged_hash(_AUX_TAG, bytes_from_octets(aux_rand))
)
else:
sec_share_ = sec_share_bytes
agg_other_nonce_bytes = (
b""
if agg_other_nonce is None
else bytes_from_octets(agg_other_nonce, _NONCE_SIZE)
)
tweak_ctx = thresh_pubkey_and_tweak(thresh_pk_bytes, tweaks, is_xonly)
tweaked_thresh_pk_xonly = tweak_ctx.x_only_pub_key
msg_bytes = bytes_from_octets(msg)
k_1 = (
_det_nonce_hash(
sec_share_,
my_id,
ids_,
agg_other_nonce_bytes,
tweaked_thresh_pk_xonly,
msg_bytes,
0,
)
% secp256k1.n
)
k_2 = (
_det_nonce_hash(
sec_share_,
my_id,
ids_,
agg_other_nonce_bytes,
tweaked_thresh_pk_xonly,
msg_bytes,
1,
)
% secp256k1.n
)
pub_nonce = _cbytes(mult(k_1, ec=secp256k1)) + _cbytes(mult(k_2, ec=secp256k1))
sec_nonce = bytearray(
k_1.to_bytes(_SCALAR_SIZE, "big") + k_2.to_bytes(_SCALAR_SIZE, "big")
)
if agg_other_nonce is None:
agg_nonce = pub_nonce
else:
try:
agg_nonce = nonce_agg([pub_nonce, agg_other_nonce])
except InvalidContributionError as e:
# the coordinator's own aggregate is what could be wrong:
# pub_nonce is this call's own, freshly built and valid
raise InvalidContributionError(None, "aggothernonce") from e
session_ctx = SessionContext(
n, t, ids, pub_shares, thresh_pk, agg_nonce, tweaks, is_xonly, msg
)
return pub_nonce, sign(sec_nonce, sec_share, my_id, session_ctx)
[docs]
def partial_sig_verify_(
psig: Octets,
my_id: int,
pub_nonce: Octets,
pub_share: Octets,
session_ctx: SessionContext,
) -> bool:
"""Verify a partial signature against a prepared session context.
`partial_sig_verify` is the other spelling: it aggregates the public
nonces itself. No bindings back this, unlike `musig2`'s: BIP445 has
no C implementation this library wraps, so every case here is the
Python arithmetic and stays that way.
"""
_integer(my_id, "my_id")
values = session_values(session_ctx)
if my_id not in session_ctx.ids:
raise BTClibEccValueError("The signer's id is missing from the ids list.")
psig_bytes = bytes_from_octets(psig, _SCALAR_SIZE)
s = int.from_bytes(psig_bytes, "big")
if s >= secp256k1.n:
return False
pub_nonce = bytes_from_octets(pub_nonce, _NONCE_SIZE)
try:
R_s1 = _cpoint(pub_nonce[:PK_SIZE])
R_s2 = _cpoint(pub_nonce[PK_SIZE:])
except BTClibEccValueError:
return False
R_s = secp256k1.add_var(R_s1, mult(values.b, R_s2, secp256k1))
if values.R[1] % 2:
R_s = secp256k1.negate(R_s)
try:
P = _cpoint(pub_share)
except BTClibEccValueError:
return False
if (
session_ctx.pub_shares is not None
and _cbytes(P) != session_ctx.pub_shares[session_ctx.ids.index(my_id)]
):
return False
a = _derive_interpolating_value(session_ctx.ids, my_id)
g = 1 if values.Q[1] % 2 == 0 else secp256k1.n - 1
g = g * values.gacc % secp256k1.n
lhs = mult(s, secp256k1.G, secp256k1)
rhs = secp256k1.add_var(R_s, mult(values.e * a * g % secp256k1.n, P, secp256k1))
return lhs == rhs
[docs]
def partial_sig_verify(
psig: Octets,
pub_nonces: Sequence[Octets],
n: int,
t: int,
ids: Sequence[int],
pub_shares: Sequence[Octets],
thresh_pk: Octets,
tweaks: Sequence[Octets],
is_xonly: Sequence[bool],
msg: Octets,
i: int,
) -> bool:
"""Verify the partial signature of signer i, against its own nonce.
`pub_shares` is required here where every other entry point takes it
optionally: there is no other way to check a signature against a
key it was not given, `tests/_data/README.md`'s cited summary of the
vendored vectors makes the same point of BIP445's own reference.
"""
ids = _ids(ids)
i = _integer(i, "i")
if len(pub_nonces) != len(ids) or len(pub_shares) != len(ids):
err_msg = "The pubnonces, pubshares and ids lists must have the same length."
raise BTClibEccValueError(err_msg)
if not 0 <= i < len(ids):
raise BTClibEccValueError("The signer index must satisfy 0 <= i <= u - 1.")
agg_nonce = nonce_agg(pub_nonces)
session_ctx = SessionContext(
n, t, ids, pub_shares, thresh_pk, agg_nonce, tweaks, is_xonly, msg
)
pub_shares_ = _pub_shares(pub_shares)
return partial_sig_verify_(psig, ids[i], pub_nonces[i], pub_shares_[i], session_ctx)
[docs]
def partial_sig_agg(psigs: Sequence[Octets], session_ctx: SessionContext) -> ssa.Sig:
"""Aggregate the partial signatures into one BIP340 signature.
`musig2.partial_sig_agg`'s own reason for answering an `ssa.Sig`
rather than 64 bytes: the result verifies under the x-only tweaked
threshold public key with `btclib_ecc.ecc.ssa.verify_` and with every
other BIP340 verifier, and handing back raw bytes would only make
the caller parse them again to find out.
"""
_assert_octets_sequence(psigs, "psigs")
values = session_values(session_ctx)
if len(psigs) != len(session_ctx.ids):
raise BTClibEccValueError("The psigs and ids lists must have the same length.")
s = 0
for i, psig in enumerate(psigs):
s_i = int.from_bytes(bytes_from_octets(psig, _SCALAR_SIZE), "big")
if s_i >= secp256k1.n:
raise InvalidContributionError(i, "psig")
s = (s + s_i) % secp256k1.n
g = 1 if values.Q[1] % 2 == 0 else secp256k1.n - 1
s = (s + values.e * g * values.tacc) % secp256k1.n
return ssa.Sig(values.R[0], s, secp256k1)